posts/locks.go

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)
	}
}