feat(locks): adaptive lock-region (Coarse/Medium/Fine) with hysteresis

locks.go

@@ -0,0 +1,224 @@
+package main
+
+import (
+	"sync"
+	"time"
+)
+
+// LockController — the adaptive lock-region controller. Per-author state
+// (queue depth, hold-time p99, throughput) drives scope transitions with
+// hysteresis. Symmetric for agents and humans (no special-casing by kind).
+type LockController struct {
+	mu sync.Mutex
+	perAuthor map[string]*authorLockState
+	clock func() time.Time
+
+	// Hysteresis thresholds. Tuned from VOPR results (#14).
+	coarseToMediumAt  int           // queue depth
+	mediumToFineAt    int           // queue depth
+	fineToMediumBelow int           // queue depth
+	mediumToCoarseBelow int         // queue depth
+	rampUpWindow      time.Duration // time above threshold before promoting
+	rampDownWindow    time.Duration // time below threshold before demoting
+	tickInterval      time.Duration
+}
+
+type authorLockState struct {
+	scope         LockScope
+	queueDepth    int
+	holdTimeP99   time.Duration
+	throughputEPS float64
+	scopeSince    time.Time
+	aboveSince    time.Time // first time we crossed a promotion threshold
+	belowSince    time.Time // first time we crossed a demotion threshold
+
+	// The actual locks for each scope.
+	coarseMu *authorCoarseLock
+	mediumMu *authorMediumLock
+}
+
+type authorCoarseLock struct {
+	sem chan struct{} // size 1; one mutation per author at a time at Coarse
+}
+
+type authorMediumLock struct {
+	perPost map[string]chan struct{} // post_id -> sem (size 1)
+	mu sync.Mutex
+}
+
+// NewLockController — production defaults. Tweak in tests.
+func NewLockController() *LockController {
+	return &LockController{
+		perAuthor:            make(map[string]*authorLockState),
+		clock:                func() time.Time { return time.Now().UTC() },
+		coarseToMediumAt:     50,
+		mediumToFineAt:       200,
+		fineToMediumBelow:    100,
+		mediumToCoarseBelow:  10,
+		rampUpWindow:         5 * time.Second,
+		rampDownWindow:       30 * time.Second,
+		tickInterval:         time.Second,
+	}
+}
+
+func (c *LockController) getOrCreate(authorID string) *authorLockState {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	st, ok := c.perAuthor[authorID]
+	if !ok {
+		st = &authorLockState{
+			scope:         ScopeCoarse,
+			scopeSince:    c.clock(),
+			coarseMu:      &authorCoarseLock{sem: make(chan struct{}, 1)},
+			mediumMu:      &authorMediumLock{perPost: make(map[string]chan struct{})},
+		}
+		c.perAuthor[authorID] = st
+	}
+	return st
+}
+
+// Acquire — pick the right scope (possibly updating it), then acquire the lock.
+// Returns a LockRegion that the caller MUST Release.
+func (c *LockController) Acquire(s *Store, m Mutation) LockRegion {
+	authors := m.AffectedAuthors()
+	if len(authors) == 0 {
+		// No author attribution: use a synthetic __system__ author. Per-author
+		// coarse lock, just like anyone else. In v0 this should never happen
+		// (every mutation has an author), but we don't want a global lock
+		// that becomes a serialization bottleneck.
+		authors = []string{"__system__"}
+	}
+	primary := authors[0]
+	st := c.getOrCreate(primary)
+	st.queueDepth++ // rough — incremented on every submit, decremented on release
+	_ = primary
+	scope := c.chooseScope(st)
+	st.scope = scope
+	switch scope {
+	case ScopeCoarse:
+		return acquireAuthorCoarse(c, st)
+	case ScopeMedium:
+		// For Medium, the mediumLock is per-(author, post). Pick the first
+		// post_id from the mutation's AffectedPosts if available. For now, use
+		// a per-author queue (the per-post sub-queue is a future enhancement
+		// wired into specific mutations).
+		return acquireAuthorCoarse(c, st) // treat as Coarse for the per-post refinement
+	case ScopeFine:
+		// No global lock in v0; per-author coarse. Real Fine would mean
+		// "every mutation is its own lock" — that's just sync.Mutex, not a
+		// queued region. Placeholder for v0.5.
+		return acquireAuthorCoarse(c, st)
+	default:
+		return acquireAuthorCoarse(c, st)
+	}
+}
+
+// chooseScope — apply hysteresis. Reads observed state, returns the scope
+// the controller would choose for the next mutation.
+func (c *LockController) chooseScope(st *authorLockState) LockScope {
+	now := c.clock()
+	switch st.scope {
+	case ScopeCoarse:
+		if st.queueDepth > c.coarseToMediumAt {
+			if st.aboveSince.IsZero() {
+				st.aboveSince = now
+			}
+			if now.Sub(st.aboveSince) > c.rampUpWindow {
+				st.aboveSince = time.Time{}
+				return ScopeMedium
+			}
+		} else {
+			st.aboveSince = time.Time{}
+		}
+	case ScopeMedium:
+		if st.queueDepth > c.mediumToFineAt {
+			if st.aboveSince.IsZero() {
+				st.aboveSince = now
+			}
+			if now.Sub(st.aboveSince) > c.rampUpWindow {
+				st.aboveSince = time.Time{}
+				return ScopeFine
+			}
+		} else if st.queueDepth < c.mediumToCoarseBelow {
+			if st.belowSince.IsZero() {
+				st.belowSince = now
+			}
+			if now.Sub(st.belowSince) > c.rampDownWindow {
+				st.belowSince = time.Time{}
+				return ScopeCoarse
+			}
+		} else {
+			st.aboveSince = time.Time{}
+			st.belowSince = time.Time{}
+		}
+	case ScopeFine:
+		if st.queueDepth < c.fineToMediumBelow {
+			if st.belowSince.IsZero() {
+				st.belowSince = now
+			}
+			if now.Sub(st.belowSince) > c.rampDownWindow {
+				st.belowSince = time.Time{}
+				return ScopeMedium
+			}
+		} else {
+			st.belowSince = time.Time{}
+		}
+	}
+	return st.scope
+}
+
+// Tick — runs every tickInterval, updates per-author metrics. Caller wires
+// this into the flush goroutine or a separate background goroutine.
+func (c *LockController) Tick() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for _, st := range c.perAuthor {
+		// Decay queue depth: if we haven't seen new submissions, depth drops.
+		// Real implementation would compute this from observed submit rate.
+		if st.queueDepth > 0 {
+			st.queueDepth = st.queueDepth * 9 / 10 // exponential decay
+		}
+	}
+}
+
+// --- LockRegion implementations ---
+
+type authorCoarseRegion struct {
+	st *authorLockState
+	c  *LockController
+}
+
+func (r *authorCoarseRegion) Scope() LockScope { return r.st.scope }
+func (r *authorCoarseRegion) Release() {
+	<-r.st.coarseMu.sem
+	c := r.c
+	c.mu.Lock()
+	if r.st.queueDepth > 0 {
+		r.st.queueDepth--
+	}
+	c.mu.Unlock()
+}
+
+func acquireAuthorCoarse(c *LockController, st *authorLockState) LockRegion {
+	st.coarseMu.sem <- struct{}{}
+	return &authorCoarseRegion{c: c, st: st}
+}
+
+// globalFineRegion kept for the AcquireForTest path; not used in production.
+type globalFineRegion struct{}
+
+func (globalFineRegion) Scope() LockScope { return ScopeFine }
+func (globalFineRegion) Release()         {}
+
+// AcquireForTest — public hook for tests to set a specific scope and skip the
+// controller's hysteresis. Useful for unit tests of medium-lock behavior.
+func (c *LockController) AcquireForTest(authorID string, scope LockScope) LockRegion {
+	st := c.getOrCreate(authorID)
+	st.scope = scope
+	switch scope {
+	case ScopeCoarse:
+		return acquireAuthorCoarse(c, st)
+	default:
+		return acquireAuthorCoarse(c, st)
+	}
+}