diff --git a/CONTEXT.md b/CONTEXT.md index c2f572a..2276f26 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -424,8 +424,9 @@ _Avoid_: Full platform schema, strict external SDK model - Default **Runtime Options** use a 24 hour dedupe TTL and a 2 minute **Thread Lock** TTL. - **Runtime Options** TTL values must be positive. - **Runtime Options** include a **Concurrency Strategy** that defaults to drop. -- The runtime implements the drop (default), queue, debounce, and concurrent **Concurrency Strategy** values plus a **Lock Scope** option (thread default, channel opt-in); the burst strategy and force/steerability remain reserved pending the deferred-dispatch admission and fenced-coordination design. +- The runtime implements the drop (default), queue, debounce, concurrent, and burst **Concurrency Strategy** values plus a **Lock Scope** option (thread default, channel opt-in); force/steerability remains reserved pending ADR 0015's formal-design bar. - Debounce coalesces on a configured quiet period and requires deferred **Dispatch Mode**; concurrent takes no **Thread Lock** and is bounded by a configured maximum; skipped (superseded) events are always observable, never silent. +- Burst batches routed events per lock scope during a fixed collection window (anchored at the first member, optionally sealed early by a batch cap) and dispatches the batch in join order under one **Thread Lock** hold, each member with its own detach budget; batch shaping is delivery-preserving (an accepted member is never dropped), parked members count against the **Admission Bound**, batches for one scope dispatch in seal order, and batching is per runtime instance. - A deferred handler whose **Lock Lease** is lost mid-run is cancelled rather than left running unserialized. - A **Thread Lock** coordinates processing of distinct **Webhook Events** for the same **Thread**; it never deduplicates them, and what happens to a conflicting event is decided by the **Concurrency Strategy** (drop acknowledges and drops it; queue coalesces waiters per process and runs the most recent after the lock releases). - A **Thread Lock** is represented as a **Lock Lease** with an ownership token. diff --git a/burst.go b/burst.go new file mode 100644 index 0000000..a7b35f5 --- /dev/null +++ b/burst.go @@ -0,0 +1,307 @@ +package chat + +import ( + "context" + "errors" + "time" +) + +// burstScope is one scope's burst coordination state: the currently collecting +// window, the FIFO of sealed batches awaiting dispatch, and whether a runner +// goroutine owns the scope. Exactly one runner is active per scope while any +// member is retained, which is what serializes rolled batches: a batch sealed +// while its predecessor dispatches waits in the FIFO and can never overtake it +// to the Thread Lock. An idle scope (no members, no runner work) deletes its +// entry, so scope cardinality never grows the map unboundedly. +type burstScope struct { + // sealed is the FIFO of closed collection windows awaiting dispatch, in + // seal order. + sealed [][]preludeWork + // open is the currently collecting window; nil when none is open. A + // window exists only while it has members: it opens on its first member's + // join and every member holds an Admission Bound slot, so parked burst + // retention is always counted against MaxDetached. + open []preludeWork + // openedAt anchors the open window's deadline at its first member's join. + // Later members never move it: the window seals at openedAt+BurstWindow + // (or at MaxBurstBatch), so a steady sub-window stream cannot defer + // dispatch indefinitely. + openedAt time.Time + // windowID identifies the open window so the runner's timer seal can + // never close a successor window early: the timer carries the id of the + // window it was armed for and a mismatch means that window already sealed. + windowID uint64 + // wake tells a runner parked on the window timer that a window sealed at + // its cap. Buffered so a seal never blocks a joining dispatch; a dropped + // token is never a lost seal because a busy runner re-reads the FIFO when + // it loops. + wake chan struct{} + // runnerActive is true while a runner goroutine owns the scope, so joins + // never double-start a runner. + runnerActive bool +} + +// seal closes the open window into the sealed FIFO. Callers hold burstMu. +func (b *burstScope) seal() { + assert(len(b.open) > 0, "sealed an empty burst window") + b.sealed = append(b.sealed, b.open) + b.open = nil +} + +// joinBurstBatch parks a routed burst member in its scope's collection window, +// opening a new window (anchored now) when none is open, sealing at +// MaxBurstBatch with the cap-reaching member included, and starting the +// scope's runner when none is active. The member's admission slot travels with +// it: the runner releases it at the member's terminal disposition, so a parked +// member counts against MaxDetached for as long as it is retained (ADR 0015 +// bounded retention). +func (c *Chat) joinBurstBatch(work preludeWork) { + assert(work.releaseAdmission != nil, "burst member parked without a held admission slot") + scope := work.scope + c.burstMu.Lock() + b := c.burstScopes[scope] + if b == nil { + b = &burstScope{wake: make(chan struct{}, 1)} + c.burstScopes[scope] = b + } + // An open window whose anchored deadline already passed (the runner is + // still dispatching a predecessor batch) seals before this member joins: + // window boundaries follow the anchor, not the runner's availability, so + // a late arrival never rides a window it missed. + if b.open != nil && !time.Now().Before(b.openedAt.Add(c.options.BurstWindow)) { + b.seal() + } + if b.open == nil { + b.windowID++ + b.openedAt = time.Now() + } + b.open = append(b.open, work) + if c.options.MaxBurstBatch > 0 && len(b.open) >= c.options.MaxBurstBatch { + // The cap-reaching member seals its own window as the batch's last + // member; the next arrival opens a rolled window dispatched strictly + // after this one. + b.seal() + select { + case b.wake <- struct{}{}: + default: + } + } + startRunner := !b.runnerActive + if startRunner { + b.runnerActive = true + // The runner joins the shutdown drain while this member's admission + // slot is still held, so no WaitGroup Add can happen after Shutdown's + // admission-slot drain completes. + c.inflight.Add(1) + } + c.burstMu.Unlock() + c.logger.Debug("chat burst member joined", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "route", work.route) + if startRunner { + go c.runBurstScope(scope) + } +} + +// runBurstScope is the per-scope burst runner: it dispatches sealed batches in +// FIFO order, waits out the open window's anchored deadline when the FIFO is +// empty, and retires (deleting the scope entry) when the scope holds no work. +// The runner is bounded only by Runtime Shutdown — per-batch coordination and +// per-member execution carry their own DetachTimeout budgets — and on +// shutdown it disposes every parked member observably before exiting. +func (c *Chat) runBurstScope(scope string) { + defer c.inflight.Done() + for { + batch, wait, wake, windowID, retired := c.nextBurstWork(scope) + if retired { + return + } + if batch != nil { + c.dispatchBurstBatch(scope, batch) + continue + } + timer := time.NewTimer(wait) + select { + case <-c.baseCtx.Done(): + timer.Stop() + c.drainBurstScope(scope) + return + case <-wake: + timer.Stop() + case <-timer.C: + c.sealBurstWindow(scope, windowID) + } + } +} + +// nextBurstWork hands the runner its next unit of work: the oldest sealed +// batch when one waits, otherwise the open window's remaining anchored wait, +// otherwise retirement — the runner flag clears and the idle scope's entry is +// deleted so an inactive scope retains nothing. Retirement and joins serialize +// under burstMu, so a join racing retirement either sees the active runner or +// starts a fresh one; a runner is never lost or doubled. +func (c *Chat) nextBurstWork(scope string) (batch []preludeWork, wait time.Duration, wake <-chan struct{}, windowID uint64, retired bool) { + c.burstMu.Lock() + defer c.burstMu.Unlock() + b := c.burstScopes[scope] + assert(b != nil && b.runnerActive, "burst runner without live scope state") + if len(b.sealed) > 0 { + batch = b.sealed[0] + b.sealed[0] = nil + b.sealed = b.sealed[1:] + return batch, 0, nil, 0, false + } + if b.open != nil { + return nil, time.Until(b.openedAt.Add(c.options.BurstWindow)), b.wake, b.windowID, false + } + b.runnerActive = false + delete(c.burstScopes, scope) + return nil, 0, nil, 0, true +} + +// sealBurstWindow closes the open window when the runner's window timer fires. +// The window id guards the seal: a cap-sealed window whose successor already +// opened must not lose its successor's remaining collection time to a stale +// timer. +func (c *Chat) sealBurstWindow(scope string, windowID uint64) { + c.burstMu.Lock() + defer c.burstMu.Unlock() + b := c.burstScopes[scope] + assert(b != nil, "burst window sealed without scope state") + if b.open == nil || b.windowID != windowID { + return + } + b.seal() +} + +// dispatchBurstBatch runs one sealed batch: a single Thread Lock hold with the +// lease refreshed across the batch, members in join order, each with its own +// DetachTimeout execution budget. The lock-wait budget is a fresh +// DetachTimeout starting when the batch reaches the head of its scope's FIFO, +// so neither collection time nor a predecessor batch's execution consumes it. +// Every member reaches an observable terminal outcome and releases its +// admission slot at disposition; the final member's slot is retained until the +// batch's lock cleanup completes so the tail work stays counted against +// MaxDetached. +func (c *Chat) dispatchBurstBatch(scope string, batch []preludeWork) { + assert(len(batch) > 0, "burst batch dispatched with no members") + // Only the identifying strings are copied out of the first member: holding + // its *Event here would keep the payload reachable after the member's + // references are cleared below. + adapter, threadID := batch[0].event.Adapter, batch[0].event.ThreadID + + waitCtx, waitCancel := context.WithTimeout(c.baseCtx, c.options.DetachTimeout) + lease, outcome, err := c.pollForLock(waitCtx, scope, nil) + waitErr := waitCtx.Err() + waitCancel() + if outcome != acquireHeld { + // A batch that never held the lock ran nothing: every member closes + // observably — a state failure as an error outcome, an abandoned wait + // as ignored — and frees its admission slot. Never silent (ADR 0015). + if outcome == acquireFailed { + c.logger.Error("chat burst acquire lock failed", "error", err, "adapter", adapter, "thread_id", threadID, "size", len(batch)) + } else { + c.logger.Info("chat burst wait abandoned", "adapter", adapter, "thread_id", threadID, "size", len(batch), "error", waitErr) + } + for i := range batch { + c.safeEnd(batch[i].span, waitOutcome(outcome), RouteAttr(batch[i].route)) + batch[i].releaseAdmission() + batch[i] = preludeWork{} + } + return + } + + // Like runLockedTail, every member is cancellable on lease loss with cause + // ErrPreempted: mutual exclusion gone means the batch must stop, not run + // alongside the lease's next holder. + batchCtx, batchCancel := context.WithCancelCause(c.baseCtx) + defer batchCancel(nil) + c.logger.Info("chat burst batch dispatch", "adapter", adapter, "thread_id", threadID, "size", len(batch)) + stopRefresh, leaseLost := c.startLockRefresh(batchCtx, lease, threadID, batchCancel) + + // finalRelease is the last member's admission release, deferred past the + // lock cleanup below. + var finalRelease func() + for i := 0; i < len(batch); i++ { + work := batch[i] + if batchCtx.Err() != nil { + // Lease loss or Runtime Shutdown: members that never started are + // skipped, never run unserialized — each closes observably as + // ignored, distinct from the preempted outcome of a member that + // was actually cancelled mid-run. + c.logger.Info("chat burst batch abandoned", "adapter", adapter, "thread_id", threadID, "remaining", len(batch)-i, "error", context.Cause(batchCtx)) + for j := i; j < len(batch); j++ { + rest := batch[j] + c.safeEnd(rest.span, OutcomeIgnored, RouteAttr(rest.route)) + if j == len(batch)-1 { + finalRelease = rest.releaseAdmission + } else { + rest.releaseAdmission() + } + batch[j] = preludeWork{} + } + break + } + // Each member gets its own DetachTimeout execution budget: a slow + // earlier member never consumes a later member's allowance. A member + // whose handler ignores cancellation blocks here cooperatively, like + // every deferred tail; its timeout outcome is recorded when it + // returns. + memberCtx, memberCancel := context.WithTimeout(batchCtx, c.options.DetachTimeout) + c.logger.Info("chat deferred dispatch started", "adapter", work.event.Adapter, "event_id", work.event.ID, "route", work.route) + runErr := work.run(memberCtx) + // The classification snapshot mirrors runLockedTail: within the run, + // the outcome follows the cancellation cause, not the handler's return + // convention — a member that observed ctx.Done, shut down cleanly, and + // returned nil still lost its lease. A handler error without lease + // loss is recorded and does not abort the batch: the remaining members + // are accepted deliveries and the lease is intact. + if errors.Is(context.Cause(memberCtx), ErrPreempted) { + c.logger.Info("chat handler preempted", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "route", work.route) + c.safeEnd(work.span, OutcomePreempted, RouteAttr(work.route)) + } else { + c.endHandlerRun(memberCtx, work.event, work.route, work.span, runErr) + } + memberCancel() + if i == len(batch)-1 { + finalRelease = work.releaseAdmission + } else { + work.releaseAdmission() + } + // A disposed member's references clear immediately so its payload and + // closure are collectable while later members still execute: batch + // retention shrinks as the batch drains instead of pinning every + // member until the tail exits. + batch[i] = preludeWork{} + } + stopRefresh() + benign := batchCtx.Err() != nil || leaseLost() + c.releaseTailLock(batchCtx, lease, threadID, benign) + assert(finalRelease != nil, "burst batch finished without a final admission release") + finalRelease() +} + +// drainBurstScope disposes every parked member — sealed batches and the open +// window — when Runtime Shutdown cancels the runner: each member's span closes +// as ignored (the same observable abandonment contract debounce waiters +// follow) and its admission slot releases, so Shutdown's admission drain can +// complete and no admitted delivery is silently lost. +func (c *Chat) drainBurstScope(scope string) { + c.burstMu.Lock() + b := c.burstScopes[scope] + assert(b != nil && b.runnerActive, "burst drain without live scope state") + var members []preludeWork + for _, batch := range b.sealed { + members = append(members, batch...) + } + members = append(members, b.open...) + b.sealed, b.open = nil, nil + b.runnerActive = false + delete(c.burstScopes, scope) + c.burstMu.Unlock() + for i := range members { + work := members[i] + c.logger.Info("chat burst wait abandoned", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "error", c.baseCtx.Err()) + c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) + work.releaseAdmission() + members[i] = preludeWork{} + } +} diff --git a/burst_hardening_test.go b/burst_hardening_test.go new file mode 100644 index 0000000..97226c1 --- /dev/null +++ b/burst_hardening_test.go @@ -0,0 +1,891 @@ +package chat_test + +import ( + "context" + "errors" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + "weak" + + "github.com/coder/chat" +) + +// This file is the burst lifecycle hardening suite: each test proves one of +// the burst-lifecycle findings from the ADR 0015 review history (PR #54) in +// code, per the issue #55 contract. The finding each test covers is named in +// its comment. + +// countingLockState counts AcquireLock calls so tests can prove the burst +// prelude never touches the Thread Lock before acknowledgement. +type countingLockState struct { + *fakeState + acquires atomic.Int64 +} + +func (s *countingLockState) AcquireLock(ctx context.Context, key string, ttl time.Duration) (chat.LockLease, bool, error) { + s.acquires.Add(1) + return s.fakeState.AcquireLock(ctx, key, ttl) +} + +// blockingReleaseState blocks the first ReleaseLock call on a gate so tests +// can observe admission accounting while a batch's lock cleanup is still in +// flight. +type blockingReleaseState struct { + *fakeState + mu sync.Mutex + blockOne bool + started chan struct{} + gate chan struct{} +} + +func newBlockingReleaseState() *blockingReleaseState { + return &blockingReleaseState{ + fakeState: newFakeState(), + blockOne: true, + started: make(chan struct{}), + gate: make(chan struct{}), + } +} + +func (s *blockingReleaseState) ReleaseLock(ctx context.Context, lease chat.LockLease) (bool, error) { + s.mu.Lock() + shouldBlock := s.blockOne + s.blockOne = false + s.mu.Unlock() + if shouldBlock { + close(s.started) + <-s.gate + } + return s.fakeState.ReleaseLock(ctx, lease) +} + +func outcomeCounts(observer *recordingObserver) map[chat.DispatchOutcome]int { + counts := map[chat.DispatchOutcome]int{} + for _, outcome := range observer.terminalOutcomes() { + counts[outcome]++ + } + return counts +} + +// Finding: window anchor — the collection window is anchored at its first +// member's join and later arrivals never extend it, so a steady sub-window +// stream cannot defer dispatch indefinitely (latency starvation). +func TestBurstWindowAnchoredAtFirstMemberNotResetByArrivals(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 250 * time.Millisecond + }) + + recorder := &burstRecorder{} + var firstHandled sync.Once + firstBatch := make(chan struct{}) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + firstHandled.Do(func() { close(firstBatch) }) + return recorder.handle(ctx, ev) + }) + + // Post events faster than the window for far longer than one window: if + // arrivals reset the timer (debounce semantics), no batch would dispatch + // while the stream continues. + posted := 0 + streaming := true + deadline := time.Now().Add(6 * time.Second) + for streaming && time.Now().Before(deadline) { + id := "event-" + string(rune('a'+posted%26)) + "-" + time.Now().Format("150405.000000") + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + posted++ + select { + case <-firstBatch: + streaming = false + case <-time.After(40 * time.Millisecond): + } + } + if streaming { + t.Fatalf("no batch dispatched while sub-window arrivals continued (%d posted); logs:\n%s", posted, logs.String()) + } + + // Delivery-preserving: every accepted member is eventually handled even + // though the anchored window moved batch boundaries mid-stream. + eventually(t, 10*time.Second, func() bool { + return len(recorder.snapshot()) == posted + }, "accepted members were dropped by window rolling") +} + +// Finding: cap-reaching-member ownership — the member that reaches +// MaxBurstBatch seals its own window as that batch's last member (the batch +// dispatches immediately, well before the window deadline), and the next +// event opens a rolled window. +func TestBurstCapReachingMemberSealsItsOwnWindow(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + // The window is far longer than the test: only the cap can seal it. + o.BurstWindow = 10 * time.Second + o.MaxBurstBatch = 3 + }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + for _, id := range []string{"event-1", "event-2", "event-3"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) == 3 + }, "cap-sealed batch did not dispatch before the window deadline") + + for _, id := range []string{"event-4", "event-5", "event-6"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) == 6 + }, "rolled window's cap-sealed batch did not dispatch") + + if got := recorder.snapshot(); !equalStrings(got, []string{"event-1", "event-2", "event-3", "event-4", "event-5", "event-6"}) { + t.Fatalf("cap-sealed batches ran out of order: %v", got) + } + if got := strings.Count(logs.String(), "chat burst batch dispatch"); got != 2 { + t.Fatalf("expected exactly 2 batch dispatches, got %d; logs:\n%s", got, logs.String()) + } + if got := strings.Count(logs.String(), "size=3"); got != 2 { + t.Fatalf("expected both batches to carry 3 members (cap-reaching member included); logs:\n%s", logs.String()) + } +} + +// Finding: lock sequencing — a burst delivery acknowledges promptly without +// acquiring or contending on the Thread Lock in the synchronous prelude; the +// batch acquires the lock only after the window seals. +func TestBurstAcknowledgesPromptlyWithoutPreludeLockAcquisition(t *testing.T) { + t.Parallel() + + state := &countingLockState{fakeState: newFakeState()} + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 300 * time.Millisecond + }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + // A foreign holder owns the scope's lock; a pre-ack lock acquisition + // would park or conflict the webhook request. + lease, acquired, err := state.fakeState.AcquireLock(context.Background(), "fake:v1:thread-1", time.Minute) + if err != nil || !acquired { + t.Fatalf("foreign lock hold: acquired=%v err=%v", acquired, err) + } + + if status := postEvent(t, bot, "fake", mentionEvent("event-1", "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch with held lock: status %d", status) + } + if got := state.acquires.Load(); got != 0 { + t.Fatalf("prelude touched the Thread Lock %d times before acknowledgement", got) + } + + if _, err := state.fakeState.ReleaseLock(context.Background(), lease); err != nil { + t.Fatalf("release foreign lock: %v", err) + } + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) == 1 + }, "batch never dispatched after the foreign lock released") +} + +// Finding: FIFO ordering of rolled batches — batches sealed in order for one +// scope acquire the lock and execute strictly in seal order, even when they +// seal while the scope's lock is held elsewhere. +func TestBurstRolledBatchesDispatchInFIFOOrder(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 10 * time.Second + o.MaxBurstBatch = 2 + }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + // Both batches seal (at the cap) while a foreign holder owns the lock, so + // they are simultaneously ready: only FIFO sequencing keeps batch 2 from + // overtaking batch 1. + lease, acquired, err := state.AcquireLock(context.Background(), "fake:v1:thread-1", time.Minute) + if err != nil || !acquired { + t.Fatalf("foreign lock hold: acquired=%v err=%v", acquired, err) + } + for _, id := range []string{"event-1", "event-2", "event-3", "event-4"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + time.Sleep(50 * time.Millisecond) + if _, err := state.ReleaseLock(context.Background(), lease); err != nil { + t.Fatalf("release foreign lock: %v", err) + } + + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) == 4 + }, "rolled batches were not all dispatched") + if got := recorder.snapshot(); !equalStrings(got, []string{"event-1", "event-2", "event-3", "event-4"}) { + t.Fatalf("rolled batches ran out of FIFO order: %v", got) + } +} + +// Finding: bounded pre-execution coordination — a batch whose lock wait +// exhausts its budget terminates: every member closes observably and frees +// its admission slot instead of leaking it behind an unavailable lock. +func TestBurstLockWaitBoundedAndMembersDisposedObservably(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + o.MaxDetached = 2 + o.DetachTimeout = 250 * time.Millisecond + }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + // The foreign holder never releases: the sealed batch must give up after + // its DetachTimeout lock-wait budget. + if _, acquired, err := state.AcquireLock(context.Background(), "fake:v1:thread-1", time.Minute); err != nil || !acquired { + t.Fatalf("foreign lock hold: acquired=%v err=%v", acquired, err) + } + for _, id := range []string{"event-1", "event-2"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + + eventually(t, 5*time.Second, func() bool { + return outcomeCounts(observer)[chat.OutcomeIgnored] == 2 + }, "abandoned batch members did not close observably") + if !strings.Contains(logs.String(), "chat burst wait abandoned") { + t.Fatalf("abandoned lock wait was not surfaced; logs:\n%s", logs.String()) + } + if got := recorder.snapshot(); len(got) != 0 { + t.Fatalf("members ran without the lock: %v", got) + } + + // Both admission slots freed: the saturated instance admits new work. + if status := postEvent(t, bot, "fake", mentionEvent("event-3", "fake:v1:thread-2")); status != 200 { + t.Fatalf("admission slot leaked behind the abandoned batch: status %d", status) + } + if status := postEvent(t, bot, "fake", mentionEvent("event-4", "fake:v1:thread-2")); status != 200 { + t.Fatalf("admission slot leaked behind the abandoned batch: status %d", status) + } +} + +// Finding: FIFO wait budget — a rolled batch's lock-wait budget starts when +// it reaches the head of its scope's FIFO, so a healthy predecessor batch +// whose members legitimately run longer than one DetachTimeout does not time +// the successor out. +func TestBurstQueuedBatchLockBudgetStartsAtHeadOfQueue(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 200 * time.Millisecond + o.MaxBurstBatch = 2 + o.DetachTimeout = 700 * time.Millisecond + }) + + recorder := &burstRecorder{} + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "event-1" || ev.Event.ID == "event-2" { + // Each first-batch member consumes most of its own budget: the + // batch as a whole runs ~900ms, past one 700ms DetachTimeout. + time.Sleep(450 * time.Millisecond) + } + return recorder.handle(ctx, ev) + }) + + for _, id := range []string{"event-1", "event-2", "event-3", "event-4"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + + eventually(t, 10*time.Second, func() bool { + return len(recorder.snapshot()) == 4 + }, "successor batch timed out while its healthy predecessor was executing") + if strings.Contains(logs.String(), "chat burst wait abandoned") { + t.Fatalf("successor batch's lock budget was consumed by its predecessor; logs:\n%s", logs.String()) + } + if got := recorder.snapshot(); !equalStrings(got, []string{"event-1", "event-2", "event-3", "event-4"}) { + t.Fatalf("batches ran out of order: %v", got) + } +} + +// Finding: per-member execution budget — every member's handler starts with a +// fresh DetachTimeout, regardless of how much of theirs earlier members used. +func TestBurstEveryMemberGetsItsOwnExecutionBudget(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 250 * time.Millisecond + o.DetachTimeout = 400 * time.Millisecond + }) + + var remaining atomic.Int64 + recorder := &burstRecorder{} + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "event-1" { + // The first member consumes most of one DetachTimeout. + time.Sleep(250 * time.Millisecond) + } + if ev.Event.ID == "event-2" { + deadline, ok := ctx.Deadline() + if !ok { + t.Error("member context has no deadline") + } + remaining.Store(int64(time.Until(deadline))) + } + return recorder.handle(ctx, ev) + }) + + for _, id := range []string{"event-1", "event-2"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) == 2 + }, "batch members were not all handled") + // A shared budget would leave event-2 ~150ms; a fresh one leaves ~400ms. + if got := time.Duration(remaining.Load()); got < 300*time.Millisecond { + t.Fatalf("second member inherited a consumed budget: %v remaining", got) + } +} + +// Finding: collection time is not execution time — the window wait consumes +// no DetachTimeout, so a window longer than DetachTimeout still dispatches +// its members with full budgets. +func TestBurstDispatchBudgetStartsWhenWindowCloses(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + // The window alone exceeds DetachTimeout: a budget that started at + // admission would already be exhausted when the member runs. + o.BurstWindow = 300 * time.Millisecond + o.DetachTimeout = 150 * time.Millisecond + }) + + recorder := &burstRecorder{} + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + time.Sleep(80 * time.Millisecond) + return recorder.handle(ctx, ev) + }) + + if status := postEvent(t, bot, "fake", mentionEvent("event-1", "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch event-1: status %d", status) + } + eventually(t, 5*time.Second, func() bool { + return outcomeCounts(observer)[chat.OutcomeHandled] == 1 + }, "member's execution budget was consumed by the collection window") +} + +// Finding: cooperative cancellation — a member that ignores cancellation until +// its DetachTimeout expires records a timeout outcome when it returns and does +// not starve the members behind it, which run with fresh budgets. +func TestBurstUncooperativeMemberTimesOutWithoutStarvingSuccessors(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + o.BurstWindow = 250 * time.Millisecond + o.DetachTimeout = 250 * time.Millisecond + }) + + recorder := &burstRecorder{} + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "event-1" { + // Cooperative-cancellation worst case: the handler only yields + // when its member context expires. + <-ctx.Done() + return ctx.Err() + } + return recorder.handle(ctx, ev) + }) + + for _, id := range []string{"event-1", "event-2"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + + eventually(t, 5*time.Second, func() bool { + counts := outcomeCounts(observer) + return counts[chat.OutcomeError] == 1 && counts[chat.OutcomeHandled] == 1 + }, "timed-out member and its successor did not both close") + if !strings.Contains(logs.String(), "chat deferred handler timed out") { + t.Fatalf("member timeout was not surfaced; logs:\n%s", logs.String()) + } + if got := recorder.snapshot(); !equalStrings(got, []string{"event-2"}) { + t.Fatalf("successor member did not run after the timed-out member: %v", got) + } +} + +// Finding: mid-batch lease loss — losing the Lock Lease cancels the running +// member with ErrPreempted and skips the remaining members rather than +// running them unserialized. +func TestBurstLeaseLossCancelsRunningMemberAndSkipsRemaining(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + o.BurstWindow = 250 * time.Millisecond + // A short TTL keeps the refresh cadence (TTL/2) fast so the loss is + // detected mid-member. + o.ThreadLockTTL = 120 * time.Millisecond + }) + + firstStarted := make(chan struct{}) + var preemptedErr atomic.Bool + recorder := &burstRecorder{} + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "event-1" { + close(firstStarted) + <-ctx.Done() + preemptedErr.Store(errors.Is(context.Cause(ctx), chat.ErrPreempted)) + return ctx.Err() + } + return recorder.handle(ctx, ev) + }) + + for _, id := range []string{"event-1", "event-2", "event-3"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + select { + case <-firstStarted: + case <-time.After(5 * time.Second): + t.Fatal("first member never started") + } + // The lease vanishes out from under the batch mid-member. + if expired, err := state.expireLock(context.Background(), "fake:v1:thread-1"); err != nil || !expired { + t.Fatalf("expire lease: expired=%v err=%v", expired, err) + } + + eventually(t, 5*time.Second, func() bool { + counts := outcomeCounts(observer) + return counts[chat.OutcomePreempted] == 1 && counts[chat.OutcomeIgnored] == 2 + }, "lease loss did not preempt the running member and skip the rest") + if !preemptedErr.Load() { + t.Fatal("running member's cancellation cause was not ErrPreempted") + } + if got := recorder.snapshot(); len(got) != 0 { + t.Fatalf("members ran after the lease was lost: %v", got) + } + if !strings.Contains(logs.String(), "chat burst batch abandoned") || !strings.Contains(logs.String(), "remaining=2") { + t.Fatalf("skipped members were not surfaced; logs:\n%s", logs.String()) + } +} + +// Finding: observable terminal outcomes — every admitted member of an aborted +// batch reaches exactly one terminal outcome; none is dropped from telemetry +// and un-started members are not misreported as preempted. +func TestBurstAbortedBatchRecordsTerminalOutcomeForEveryMember(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + o.BurstWindow = 250 * time.Millisecond + o.ThreadLockTTL = 120 * time.Millisecond + }) + + firstStarted := make(chan struct{}) + recorder := &burstRecorder{} + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "event-1" { + close(firstStarted) + <-ctx.Done() + return ctx.Err() + } + return recorder.handle(ctx, ev) + }) + + for _, id := range []string{"event-1", "event-2", "event-3"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + select { + case <-firstStarted: + case <-time.After(5 * time.Second): + t.Fatal("first member never started") + } + if expired, err := state.expireLock(context.Background(), "fake:v1:thread-1"); err != nil || !expired { + t.Fatalf("expire lease: expired=%v err=%v", expired, err) + } + eventually(t, 5*time.Second, func() bool { + return len(observer.terminalOutcomes()) == 3 + }, "aborted batch members were dropped from telemetry") + + // A healthy follow-up batch on the same scope still closes cleanly. + if status := postEvent(t, bot, "fake", mentionEvent("event-4", "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch event-4: status %d", status) + } + eventually(t, 5*time.Second, func() bool { + return len(observer.terminalOutcomes()) == 4 + }, "follow-up batch member did not close") + + counts := outcomeCounts(observer) + if counts[chat.OutcomePreempted] != 1 || counts[chat.OutcomeIgnored] != 2 || counts[chat.OutcomeHandled] != 1 { + t.Fatalf("terminal outcomes misreported: %v", counts) + } +} + +// Finding: handler-error member disposition — a member whose handler fails +// records the error and does not abort its batch while the lease is intact. +func TestBurstMemberHandlerErrorDoesNotAbortBatch(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + o.BurstWindow = 250 * time.Millisecond + }) + + recorder := &burstRecorder{} + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "event-1" { + return errors.New("boom") + } + return recorder.handle(ctx, ev) + }) + + for _, id := range []string{"event-1", "event-2"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + + eventually(t, 5*time.Second, func() bool { + counts := outcomeCounts(observer) + return counts[chat.OutcomeError] == 1 && counts[chat.OutcomeHandled] == 1 + }, "failing member aborted its batch") + if !observer.hasEvent(chat.ObsHandlerError) { + t.Fatal("handler error was not observed") + } + if got := recorder.snapshot(); !equalStrings(got, []string{"event-2"}) { + t.Fatalf("successor member did not run after the failing member: %v", got) + } +} + +// Finding: member-reference clearing — a disposed member's event payload is +// collectable while later members of the same batch still execute, so batch +// retention shrinks as the batch drains. +func TestBurstDisposedMemberReleasesEventForGC(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 250 * time.Millisecond + }) + + var weakEvent atomic.Pointer[weak.Pointer[chat.Event]] + secondStarted := make(chan struct{}) + gate := make(chan struct{}) + var release sync.Once + t.Cleanup(func() { release.Do(func() { close(gate) }) }) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + switch ev.Event.ID { + case "event-1": + ref := weak.Make(ev.Event) + weakEvent.Store(&ref) + case "event-2": + close(secondStarted) + <-gate + } + return nil + }) + + for _, id := range []string{"event-1", "event-2"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + select { + case <-secondStarted: + case <-time.After(5 * time.Second): + t.Fatal("second member never started") + } + + // The first member is disposed while the second still executes: its event + // payload must be unreachable. + eventually(t, 10*time.Second, func() bool { + runtime.GC() + ref := weakEvent.Load() + return ref != nil && ref.Value() == nil + }, "disposed member's event payload was still retained mid-batch") + release.Do(func() { close(gate) }) +} + +// Finding: admission slot retention through the batch tail — members free +// their slots at disposition, except the batch's final member, whose slot is +// held until the batch's lock cleanup (ReleaseLock) completes. +func TestBurstFinalMemberSlotHeldThroughLockRelease(t *testing.T) { + t.Parallel() + + state := newBlockingReleaseState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 10 * time.Second + o.MaxBurstBatch = 2 + o.MaxDetached = 2 + }) + var release sync.Once + t.Cleanup(func() { release.Do(func() { close(state.gate) }) }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + // Two members fill both admission slots and cap-seal one batch. + for _, id := range []string{"event-1", "event-2"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + select { + case <-state.started: + case <-time.After(5 * time.Second): + t.Fatal("batch lock release never started") + } + + // Both members completed but ReleaseLock is still in flight: the first + // member's slot is free, the final member's is not. + if status := postEvent(t, bot, "fake", mentionEvent("probe-1", "fake:v1:thread-1")); status != 200 { + t.Fatalf("first slot was not freed at member disposition: status %d", status) + } + status, body := postEventBody(t, bot, "fake", mentionEvent("probe-2", "fake:v1:thread-1")) + if status == 200 || !strings.Contains(body, "admission rejected") { + t.Fatalf("final member's slot was freed before lock cleanup completed: status %d body %q", status, body) + } + + release.Do(func() { close(state.gate) }) + // With the cleanup finished, the final slot frees: the next probe both + // admits and cap-seals the probe window so it dispatches. + eventually(t, 5*time.Second, func() bool { + status, _ := postEventBody(t, bot, "fake", mentionEvent("probe-3", "fake:v1:thread-1")) + return status == 200 + }, "final member's slot never freed after lock cleanup") + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) >= 4 + }, "probe batch never dispatched after the final slot freed") +} + +// ADR 0015 bounded retention: parked burst members occupy Admission Bound +// slots, so a window full of parked payloads saturates MaxDetached and new +// deliveries are honestly rejected until the batch disposes. +func TestBurstParkedMembersOccupyAdmissionSlots(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + o.BurstWindow = 400 * time.Millisecond + o.MaxDetached = 2 + }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + for _, id := range []string{"event-1", "event-2"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + // Both slots are held by parked members: the next delivery is rejected + // before acknowledgement and before dedupe marking. + status, body := postEventBody(t, bot, "fake", mentionEvent("event-3", "fake:v1:thread-1")) + if status == 200 || !strings.Contains(body, "admission rejected") { + t.Fatalf("parked members did not occupy admission slots: status %d body %q", status, body) + } + if !observer.hasEvent(chat.ObsAdmissionRejected) { + t.Fatal("admission rejection was not observed") + } + + // Once the batch disposes, capacity frees and the same event admits (it + // was never dedupe-marked). + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) == 2 + }, "parked batch never dispatched") + eventually(t, 5*time.Second, func() bool { + status, _ := postEventBody(t, bot, "fake", mentionEvent("event-3", "fake:v1:thread-1")) + return status == 200 + }, "capacity never freed after the batch disposed") +} + +// Finding: open-window shutdown drain — Runtime Shutdown disposes parked +// members promptly and observably instead of waiting out their window or +// dropping them silently. +func TestBurstOpenWindowDrainedByShutdown(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + o.BurstWindow = 10 * time.Second + }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + for _, id := range []string{"event-1", "event-2"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + + start := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := bot.Shutdown(ctx); err != nil { + t.Fatalf("shutdown with an open window: %v", err) + } + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Fatalf("shutdown waited out the collection window: %v", elapsed) + } + counts := outcomeCounts(observer) + if counts[chat.OutcomeIgnored] != 2 { + t.Fatalf("parked members were not observably disposed: %v", counts) + } + if !strings.Contains(logs.String(), "chat burst wait abandoned") { + t.Fatalf("abandoned members were not surfaced; logs:\n%s", logs.String()) + } + if got := chat.BurstScopeCount(bot); got != 0 { + t.Fatalf("burst scope state survived shutdown: %d entries", got) + } + if got := recorder.snapshot(); len(got) != 0 { + t.Fatalf("members ran during shutdown drain: %v", got) + } +} + +// Finding: admission-to-drain race — deliveries racing Runtime Shutdown are +// either honestly rejected or admitted and drained to a terminal outcome; no +// burst state survives shutdown. +func TestBurstDispatchRacingShutdownRejectedOrDrained(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + observer := &recordingObserver{} + bot := newBurstRuntime(t, state, adapter, &logs, observer, func(o *chat.RuntimeOptions) { + o.BurstWindow = 30 * time.Millisecond + }) + bot.OnNewMention(func(context.Context, *chat.MessageEvent) error { return nil }) + + const deliveries = 12 + statuses := make([]int, deliveries) + bodies := make([]string, deliveries) + var wg sync.WaitGroup + for i := 0; i < deliveries; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + id := "race-event-" + string(rune('a'+i)) + statuses[i], bodies[i] = postEventBody(t, bot, "fake", mentionEvent(id, "fake:v1:thread-race")) + }(i) + } + time.Sleep(10 * time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := bot.Shutdown(ctx); err != nil { + t.Fatalf("shutdown racing dispatch: %v", err) + } + wg.Wait() + + for i, status := range statuses { + if status != 200 && !strings.Contains(bodies[i], "admission rejected") { + t.Fatalf("delivery %d neither admitted nor honestly rejected: status %d body %q", i, status, bodies[i]) + } + } + // Every opened dispatch span closes: admitted deliveries drained to a + // terminal outcome, rejected ones closed as admission-rejected. + eventually(t, 5*time.Second, func() bool { + return len(observer.terminalOutcomes()) == deliveries + }, "a racing delivery was lost without a terminal outcome") + if got := chat.BurstScopeCount(bot); got != 0 { + t.Fatalf("burst scope state survived shutdown: %d entries", got) + } +} + +// Finding: idle-coordinator GC — a scope whose batches all disposed retains no +// coordinator map entry, so scope cardinality cannot leak memory. +func TestBurstIdleScopeCoordinatorRemoved(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 40 * time.Millisecond + }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + threads := []chat.ThreadID{"fake:v1:t1", "fake:v1:t2", "fake:v1:t3", "fake:v1:t4", "fake:v1:t5"} + for i, thread := range threads { + if status := postEvent(t, bot, "fake", mentionEvent("event-"+string(rune('a'+i)), thread)); status != 200 { + t.Fatalf("dispatch to %s: status %d", thread, status) + } + } + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) == len(threads) + }, "scope batches were not all dispatched") + eventually(t, 5*time.Second, func() bool { + return chat.BurstScopeCount(bot) == 0 + }, "idle scope coordinators were not garbage collected") +} diff --git a/burst_test.go b/burst_test.go new file mode 100644 index 0000000..68ca789 --- /dev/null +++ b/burst_test.go @@ -0,0 +1,219 @@ +package chat_test + +import ( + "context" + "log/slog" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/chat" +) + +// newBurstRuntime builds a burst runtime with a short collection window and a +// fast lock poll cadence, optional observer, and optional option mutations. +func newBurstRuntime(t *testing.T, state chat.State, adapter chat.Adapter, logs *syncBuffer, observer chat.Observer, mutate ...func(*chat.RuntimeOptions)) *chat.Chat { + t.Helper() + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + Concurrency: chat.ConcurrencyBurst, + BurstWindow: 60 * time.Millisecond, + Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, + // ThreadLockTTL/20 is the lock poll cadence, so a short TTL keeps + // batch lock acquisition prompt in tests. + ThreadLockTTL: 200 * time.Millisecond, + DetachTimeout: 5 * time.Second, + } + for _, m := range mutate { + m(&options) + } + opts := []chat.Option{ + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithLogger(slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + chat.WithRuntimeOptions(options), + } + if observer != nil { + opts = append(opts, chat.WithObserver(observer)) + } + bot, err := chat.New(context.Background(), opts...) + if err != nil { + t.Fatalf("new burst runtime: %v", err) + } + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := bot.Shutdown(ctx); err != nil { + t.Errorf("shutdown burst runtime: %v", err) + } + }) + return bot +} + +// burstRecorder tracks handled event IDs in handler completion order. +type burstRecorder struct { + mu sync.Mutex + handled []string +} + +func (r *burstRecorder) handle(_ context.Context, ev *chat.MessageEvent) error { + r.mu.Lock() + defer r.mu.Unlock() + r.handled = append(r.handled, ev.Event.ID) + return nil +} + +func (r *burstRecorder) snapshot() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.handled...) +} + +func TestBurstDispatchesCollectedBatchInJoinOrder(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil, func(o *chat.RuntimeOptions) { + o.BurstWindow = 250 * time.Millisecond + }) + + recorder := &burstRecorder{} + bot.OnNewMention(recorder.handle) + + // Three events arrive well inside one collection window: they dispatch as + // one batch, in join order, under a single Thread Lock hold. + for _, id := range []string{"event-1", "event-2", "event-3"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch %s: status %d", id, status) + } + } + + eventually(t, 5*time.Second, func() bool { + return len(recorder.snapshot()) == 3 + }, "batch members were not all handled") + if got := recorder.snapshot(); !equalStrings(got, []string{"event-1", "event-2", "event-3"}) { + t.Fatalf("batch ran out of join order: %v", got) + } + if !strings.Contains(logs.String(), "chat burst batch dispatch") || !strings.Contains(logs.String(), "size=3") { + t.Fatalf("expected one size-3 batch dispatch; logs:\n%s", logs.String()) + } +} + +func TestBurstWindowOpenedDuringDispatchRunsAfterCurrentBatch(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, nil) + + started := make(chan string, 8) + gate := make(chan struct{}) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + started <- ev.Event.ID + if ev.Event.ID == "event-1" { + <-gate + } + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("event-1", "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch event-1: status %d", status) + } + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("first batch never started") + } + + // The first batch is mid-dispatch (its member blocked): new arrivals open + // a successor window that dispatches strictly after the current batch. + if status := postEvent(t, bot, "fake", mentionEvent("event-2", "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch event-2: status %d", status) + } + if status := postEvent(t, bot, "fake", mentionEvent("event-3", "fake:v1:thread-1")); status != 200 { + t.Fatalf("dispatch event-3: status %d", status) + } + close(gate) + + eventually(t, 5*time.Second, func() bool { + mu.Lock() + defer mu.Unlock() + return len(handled) == 3 + }, "successor window members were not handled") + mu.Lock() + got := append([]string(nil), handled...) + mu.Unlock() + if got[0] != "event-1" { + t.Fatalf("successor batch overtook the dispatching batch: %v", got) + } + if got[1] == "event-1" || got[2] == "event-1" { + t.Fatalf("first batch member ran twice: %v", got) + } +} + +func TestBurstConstructionValidation(t *testing.T) { + t.Parallel() + + base := func() chat.RuntimeOptions { + return chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Minute, + Concurrency: chat.ConcurrencyBurst, + BurstWindow: 50 * time.Millisecond, + Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, + DetachTimeout: time.Second, + } + } + cases := []struct { + name string + mutate func(*chat.RuntimeOptions) + wantErr string + }{ + { + name: "zero burst window", + mutate: func(o *chat.RuntimeOptions) { o.BurstWindow = 0 }, + wantErr: "burst window must be positive", + }, + { + name: "negative burst window", + mutate: func(o *chat.RuntimeOptions) { o.BurstWindow = -time.Second }, + wantErr: "burst window must be positive", + }, + { + name: "negative max burst batch", + mutate: func(o *chat.RuntimeOptions) { o.MaxBurstBatch = -1 }, + wantErr: "max burst batch must not be negative", + }, + { + name: "synchronous dispatch", + mutate: func(o *chat.RuntimeOptions) { o.Dispatch = chat.DispatchSync }, + wantErr: "burst strategy requires deferred dispatch", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + options := base() + tc.mutate(&options) + _, err := chat.New(context.Background(), + chat.WithState(newFakeState()), + chat.WithAdapter(newFakeAdapter("fake")), + chat.WithRuntimeOptions(options), + ) + if err == nil || !strings.Contains(err.Error(), tc.wantErr) { + t.Fatalf("expected %q construction error, got %v", tc.wantErr, err) + } + }) + } +} diff --git a/docs/adr/0012-concurrency-strategy.md b/docs/adr/0012-concurrency-strategy.md index 08de0c0..25da4a2 100644 --- a/docs/adr/0012-concurrency-strategy.md +++ b/docs/adr/0012-concurrency-strategy.md @@ -2,7 +2,7 @@ ## Status -Accepted (implementation staged: `drop`, `queue`, `debounce`, and `concurrent` strategies plus the `LockScope` option ship in the runtime, with `debounce` requiring `DispatchDeferred`; the `burst` and force/steerability names remain reserved). +Accepted (implemented: the `drop`, `queue`, `debounce`, `concurrent`, and `burst` strategies plus the `LockScope` option ship in the runtime, with `debounce` and `burst` requiring `DispatchDeferred`; `burst` landed per ADR 0015's revival decision under the Admission Bound's invariants (issue #55). The force/steerability names remain reserved). Three statements of this ADR are superseded by ADR 0015 (deferred-dispatch admission bound; cross-instance coalescing rejected for now): diff --git a/docs/adr/0015-runtime-coordination.md b/docs/adr/0015-runtime-coordination.md index e29cd19..2c7bc0a 100644 --- a/docs/adr/0015-runtime-coordination.md +++ b/docs/adr/0015-runtime-coordination.md @@ -2,7 +2,7 @@ ## Status -Accepted (the admission bound is implemented; issue #44). This decides the deferred-dispatch admission bound (issue #44) and explicitly rejects, for now, the cross-instance coalescing State extension (issue #50) after a full design attempt — see the rejection section for the evidence and the reopening bar. Issue #50 stays open as gated future work. This ADR also gives the staged burst/preemption branch (PR #53) its verdict. +Accepted (the admission bound and the per-instance burst revival are implemented; issues #44 and #55). This decides the deferred-dispatch admission bound (issue #44) and explicitly rejects, for now, the cross-instance coalescing State extension (issue #50) after a full design attempt — see the rejection section for the evidence and the reopening bar. Issue #50 stays open as gated future work. This ADR also gives the staged burst/preemption branch (PR #53) its verdict. This is a decision-level document: it fixes decisions, invariants, and non-goals. Implementation mechanics — slot bookkeeping, timer and shutdown lifecycles, counter management — are deliberately not specified here; they are decided in the implementing PRs, where code and hardening tests can actually verify them against the invariants below. diff --git a/docs/how-to/deferred-dispatch.md b/docs/how-to/deferred-dispatch.md index fe41756..93e9091 100644 --- a/docs/how-to/deferred-dispatch.md +++ b/docs/how-to/deferred-dispatch.md @@ -63,11 +63,19 @@ bot, err := chat.New(ctx, follow-up is cancelled without running (it was already deduped, so it will not be redelivered). Size `DetachTimeout` to cover your longest handler *plus* the queue wait behind it. +- `Concurrency: chat.ConcurrencyBurst` batches instead of coalescing: events + for a thread collect during a fixed `BurstWindow`, then run as one batch in + join order under a single lock hold, each member with its own + `DetachTimeout` budget — no accepted event is dropped. `MaxBurstBatch` + optionally seals a full window early; batches dispatch in seal order. + Batching is per process, like queue coalescing. See the + `chat.ConcurrencyBurst` GoDoc for the full lifecycle contract. - `MaxDetached` (required under `DispatchDeferred`; `DefaultRuntimeOptions()` sets 1024) is the admission bound from [ADR 0015](../adr/0015-runtime-coordination.md): it caps admitted-but-incomplete deferred deliveries — running handlers, queued and - debounced waiters, concurrent slot-waiters — so an event flood cannot grow + debounced waiters, concurrent slot-waiters, parked burst batch members — so + an event flood cannot grow goroutines and retained payloads without limit. A delivery arriving at the cap is rejected with `chat.ErrAdmissionRejected` **before** the ack and **before** dedupe marking; the adapter maps that to a retry-inducing 503 for diff --git a/export_test.go b/export_test.go index 0798539..2c0dc42 100644 --- a/export_test.go +++ b/export_test.go @@ -10,3 +10,13 @@ func AdmissionTenantEntries(c *Chat) int { } return c.admission.tenantEntries() } + +// BurstScopeCount reports the number of live burst scope coordinators, for +// hardening tests proving an idle scope retains no map entry (ADR 0015 bounded +// retention). +func BurstScopeCount(c *Chat) int { + assert(c != nil, "BurstScopeCount called on nil runtime") + c.burstMu.Lock() + defer c.burstMu.Unlock() + return len(c.burstScopes) +} diff --git a/runtime.go b/runtime.go index 9989276..bd58501 100644 --- a/runtime.go +++ b/runtime.go @@ -46,12 +46,36 @@ const ( // MaxConcurrent. No Thread Lock is taken, so the caller accepts interleaved // replies and races on Thread Application State. ConcurrencyConcurrent + // ConcurrencyBurst collects routed events for a scope into a batch while a + // BurstWindow collection window is open, then dispatches the batch under a + // single Thread Lock hold, running every member in join order, each with + // its own DetachTimeout execution budget. The window is anchored at its + // first member's join and is never extended by later arrivals; a window + // reaching MaxBurstBatch seals immediately (the cap-reaching member is the + // batch's last member) and the next event opens a rolled window dispatched + // strictly after its predecessor. Batch shaping is delivery-preserving + // (ADR 0015): boundaries move, but an accepted member is never dropped. + // Requires deferred dispatch (a synchronous webhook cannot park an event + // past the platform's acknowledgement deadline). + // + // Parked members hold Admission Bound slots until their terminal + // disposition, so burst retention stays inside MaxDetached. A member whose + // handler returns an error does not abort its batch; losing the Lock + // Lease mid-batch cancels the running member (ErrPreempted) and skips the + // remaining members observably rather than running them unserialized. + // Like every deferred handler, member cancellation is cooperative: a + // handler that ignores its context blocks the members behind it. + // + // Like queue supersession and debounce coalescing, batching is per runtime + // instance (ADR 0015): events for one scope delivered to different + // instances sharing a State batch independently, serialized by the Thread + // Lock. Cross-instance coalescing is rejected for now behind that ADR's + // reopening bar. + ConcurrencyBurst ) -// The burst strategy and force/steerability names from ADR 0012 remain -// reserved. Per ADR 0015, burst revives as its own PR under the Admission -// Bound's invariants, while force/steerability is rejected pending that ADR's -// formal-design bar. +// The force/steerability names from ADR 0012 remain reserved: per ADR 0015 +// they are rejected pending that ADR's formal-design bar. // LockScope selects what key the Thread Lock guards. The opaque Thread ID is // unchanged; the scope only chooses the serialization key. @@ -96,12 +120,29 @@ type RuntimeOptions struct { // ConcurrencyConcurrent. It must be positive under that strategy and is // ignored otherwise. MaxConcurrent int + // BurstWindow is the ConcurrencyBurst collection window: routed events for + // a scope collect for this long — anchored at the window's first member, + // never extended by later arrivals — before dispatching as one batch. It + // must be positive under that strategy and is ignored otherwise. The + // window is collection time, not execution time: it does not consume the + // batch's lock-wait budget or any member's DetachTimeout. + BurstWindow time.Duration + // MaxBurstBatch caps how many members one burst batch may collect. A + // window reaching the cap seals immediately, with the cap-reaching member + // as the sealed batch's last member; the next event opens a rolled window + // dispatched strictly after its predecessor. The cap shapes batches — it + // never rejects or drops an accepted member (delivery-preserving shaping, + // ADR 0015) — and parked members remain bounded by MaxDetached regardless. + // Zero disables the cap; it must not be negative under the burst strategy + // and is ignored otherwise. + MaxBurstBatch int // MaxDetached is the deferred-dispatch Admission Bound (ADR 0015): a // per-instance cap on admitted-but-incomplete deferred deliveries. // Everything a delivery retains under DispatchDeferred counts against it — - // running detached tails, parked queue/debounce waiters, and concurrent - // slot-waiters — and capacity frees only when that retention ends. A - // delivery arriving at the cap is rejected with ErrAdmissionRejected + // running detached tails, parked queue/debounce waiters, concurrent + // slot-waiters, and parked burst batch members — and capacity frees only + // when that retention ends. A delivery arriving at the cap is rejected + // with ErrAdmissionRejected // before acknowledgement and before dedupe marking, so a platform retry is // never deduped away. It must be positive under DispatchDeferred and is // ignored under DispatchSync; DefaultRuntimeOptions sets 1024. @@ -202,6 +243,10 @@ type Chat struct { // ConcurrencyConcurrent; nil under every other strategy. concurrencySlots chan struct{} + // burstMu guards burstScopes, the per-scope burst collection state. + burstMu sync.Mutex + burstScopes map[string]*burstScope + // admission is the deferred-dispatch Admission Bound (ADR 0015); nil under // DispatchSync. admission *admissionGate @@ -255,6 +300,7 @@ func New(ctx context.Context, opts ...Option) (*Chat, error) { baseCtx: baseCtx, baseCancel: baseCancel, pending: map[string]*pendingWaiter{}, + burstScopes: map[string]*burstScope{}, } if cfg.options.Concurrency == ConcurrencyConcurrent { chat.concurrencySlots = make(chan struct{}, cfg.options.MaxConcurrent) @@ -314,6 +360,16 @@ func validateRuntimeOptions(options RuntimeOptions) error { if options.MaxConcurrent <= 0 { return errors.New("chat: max concurrent must be positive under the concurrent strategy") } + case ConcurrencyBurst: + if options.BurstWindow <= 0 { + return errors.New("chat: burst window must be positive under the burst strategy") + } + if options.MaxBurstBatch < 0 { + return errors.New("chat: max burst batch must not be negative under the burst strategy") + } + if options.Dispatch != DispatchDeferred { + return errors.New("chat: burst strategy requires deferred dispatch") + } default: return errors.New("chat: unsupported concurrency strategy") } @@ -568,6 +624,15 @@ func (c *Chat) dispatchDeferred(ctx context.Context, event *Event, seq uint64) e return err } work.releaseAdmission = release + if work.burst { + // A burst member parks in its scope's collection window instead of + // owning a detached tail; its admission slot travels with it and the + // scope's runner releases it at the member's terminal disposition, so + // a parked member counts against MaxDetached for as long as its + // payload and closure are retained. + c.joinBurstBatch(work) + return nil + } c.startDetachedTail(work) return nil } @@ -600,6 +665,11 @@ type preludeWork struct { // noLock is true under the concurrent strategy: no Thread Lock is taken and // the run is bounded by a MaxConcurrent slot instead. noLock bool + // burst is true under the burst strategy: the routed event joins its + // scope's collection window as a batch member instead of owning a detached + // tail; the scope's runner owns its span, its admission slot, and its + // terminal disposition. + burst bool // releaseAdmission frees the delivery's Admission Bound slot. It is set // only under DispatchDeferred and runs when the detached tail goroutine // returns — not when the handler returns — so stalled cleanup (lock @@ -675,8 +745,9 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64) (preludeWo } scope := c.lockScopeKey(event, ref) - // The prelude acquires the Thread Lock only under drop/queue; debounce - // always coordinates in the tail, and concurrent takes no lock at all. + // The prelude acquires the Thread Lock only under drop/queue; debounce and + // burst always coordinate in the tail (keeping ack prompt and free of lock + // contention), and concurrent takes no lock at all. var lease LockLease conflicted := false switch c.options.Concurrency { @@ -706,7 +777,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64) (preludeWo } conflicted = true } - case ConcurrencyDebounce, ConcurrencyConcurrent: + case ConcurrencyDebounce, ConcurrencyBurst, ConcurrencyConcurrent: } // releaseOnResolve releases the lease when the event resolves here; an event @@ -876,6 +947,15 @@ func (c *Chat) routedWork( case ConcurrencyConcurrent: work.noLock = true return work, false, nil + + case ConcurrencyBurst: + // The routed event becomes a burst member: dispatchDeferred parks it + // in its scope's collection window with its admission slot attached. + // Burst requires deferred dispatch (validated at construction), so + // dispatchSync never sees this flag. + assert(c.options.Dispatch == DispatchDeferred, "burst strategy requires deferred dispatch") + work.burst = true + return work, false, nil } // Unreachable: the strategy set is validated at construction.