diff --git a/locks.go b/locks.go new file mode 100644 index 0000000..977fed3 --- /dev/null +++ b/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) + } +}