diff --git a/CONTEXT.md b/CONTEXT.md index ef3846c..d1dfcf3 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -162,6 +162,10 @@ _Avoid_: Lock implementation detail A per-**Thread** coordination guard that prevents concurrent handler execution for the same conversation. _Avoid_: Event lock, adapter lock, handler lock +**Lock Scope**: +The **Runtime Options** choice of what key the **Thread Lock** guards: a single **Thread** (default) or the Thread's whole channel where the platform's model requires channel-wide serialization. +_Avoid_: Lock granularity flag, channel lock + **Lock Lease**: A token-owned **Thread Lock** record that can only be released or extended by its current owner. _Avoid_: Untokened lock, delete-only lock @@ -416,7 +420,8 @@ _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) and queue **Concurrency Strategy** values; burst, debounce, concurrent, lock-scope, and force/steerability remain reserved for future slices. +- The runtime implements the full upstream-aligned **Concurrency Strategy** set: drop (default), queue, debounce, burst, and concurrent, plus a **Lock Scope** option (thread default, channel opt-in) and a force/steerability hook that preempts an in-flight handler by force-releasing its **Lock Lease** through an optional **Runtime State** capability. +- Debounce and burst coalesce on a configured interval and require deferred **Dispatch Mode**; concurrent takes no **Thread Lock** and is bounded by a configured maximum; skipped (superseded) events are always observable, never silent. - 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. - Releasing or extending a **Lock Lease** must verify the ownership token so an expired holder cannot affect a newer holder. diff --git a/README.md b/README.md index 1f35922..c5c3ba1 100644 --- a/README.md +++ b/README.md @@ -518,11 +518,31 @@ chat.RuntimeOptions{ } ``` -Two concurrency strategies are implemented: `ConcurrencyDrop` (the default) -acknowledges and drops events that hit a locked thread, and `ConcurrencyQueue` -waits for the lock and runs only the most recent superseded follow-up, with -most-recent coalescing scoped per process (ADR 0012). Burst, debounce, force, -and concurrent strategies remain proposed in ADR 0012 and are not implemented. +The runtime implements the full upstream-aligned strategy set (ADR 0012): + +- `ConcurrencyDrop` (default): a lock conflict is acknowledged and dropped. +- `ConcurrencyQueue`: the newest follow-up waits for the in-flight handler; + superseded follow-ups are observable, never silent. +- `ConcurrencyDebounce`: each new event resets a `DebounceInterval` timer; only + the final event in a quiet period dispatches. Requires deferred dispatch. + Coalescing (like queue supersession) is per runtime instance; instances + sharing a state are serialized by the thread lock, not coalesced. +- `ConcurrencyBurst`: events collect for `DebounceInterval` on an idle scope, + then the whole batch dispatches — in the order events joined the window — + under one lock hold and a fresh `DetachTimeout` that starts when the window + closes. Requires deferred dispatch. +- `ConcurrencyConcurrent`: no thread lock at all; every event dispatches in its + own execution, bounded by `MaxConcurrent`. + +`LockScope` chooses what the lock guards: per thread (default) or per channel +(`LockScopeChannel`) for platforms whose model needs channel-wide ordering. + +`OnLockConflict` is the force/steerability hook: on a lock conflict it can +preempt the in-flight work. A local handler is cancelled with +`chat.ErrPreempted` and awaited, after which the new delivery acquires the +released lock; a lease held by another runtime instance (or orphaned) is +force released through the optional `LockForcer` state capability instead. It +requires deferred dispatch and the drop or queue strategy. Thread locks use token-owned lock leases. Release and extend operations must verify the token so an expired handler cannot release or extend another @@ -795,8 +815,6 @@ include: interaction response needs - no bundled metrics framework, exporters, or scrape endpoint (an optional no-op `Observer` seam is provided; OpenTelemetry stays out of the core import graph) -- no burst, debounce, force, or concurrent lock-conflict strategies (drop and - queue are implemented) - no built-in HTTP server or router integrations - no adapter marketplace/package conventions diff --git a/burst_test.go b/burst_test.go new file mode 100644 index 0000000..d1b217f --- /dev/null +++ b/burst_test.go @@ -0,0 +1,419 @@ +package chat_test + +import ( + "context" + "log/slog" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/chat" +) + +func newBurstRuntime(t *testing.T, state chat.State, adapter chat.Adapter, logs *syncBuffer, mutate ...func(*chat.RuntimeOptions)) *chat.Chat { + t.Helper() + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: 40 * time.Millisecond, + Concurrency: chat.ConcurrencyBurst, + DebounceInterval: 60 * time.Millisecond, + Dispatch: chat.DispatchDeferred, + DetachTimeout: 5 * time.Second, + } + for _, m := range mutate { + m(&options) + } + bot, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithLogger(slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + chat.WithRuntimeOptions(options), + ) + if err != nil { + t.Fatalf("new burst runtime: %v", err) + } + return bot +} + +func TestBurstDispatchesCollectedBatchInJoinOrder(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.DebounceInterval = 150 * time.Millisecond + }) + + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + return nil + }) + + // Three events join one 150ms collection window: the whole batch + // dispatches, in join order, once the window closes. + start := time.Now() + for _, id := range []string{"a", "b", "c"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("%s status = %d", id, status) + } + time.Sleep(5 * time.Millisecond) + } + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 3 { + break + } + select { + case <-deadline: + mu.Lock() + got := append([]string(nil), handled...) + mu.Unlock() + t.Fatalf("batch did not fully dispatch, handled = %v", got) + case <-time.After(5 * time.Millisecond): + } + } + if elapsed := time.Since(start); elapsed < 150*time.Millisecond { + t.Fatalf("batch dispatched %v after the first event, before the window closed", elapsed) + } + + time.Sleep(50 * time.Millisecond) + mu.Lock() + got := append([]string(nil), handled...) + mu.Unlock() + if !equalStrings(got, []string{"a", "b", "c"}) { + t.Fatalf("handled = %v, want the full batch in join order [a b c]", got) + } + + out := logs.String() + if !strings.Contains(out, "chat burst batch dispatch") || !strings.Contains(out, "size=3") { + t.Fatalf("batch dispatch (size=3) not surfaced via observation; logs:\n%s", out) + } + // Burst never skips or drops events. + if strings.Contains(out, "superseded") || strings.Contains(out, "chat lock conflict dropped") { + t.Fatalf("burst skipped or dropped a batch member; logs:\n%s", out) + } +} + +func TestBurstWindowOpenedDuringDispatchRunsAfterCurrentBatch(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.DebounceInterval = 30 * time.Millisecond + }) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("a", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("a status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first batch did not start dispatching") + } + + // A new window opens while the first batch is still dispatching; its batch + // must run only after the current one finishes. + if status := postEvent(t, bot, "fake", mentionEvent("b", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("b status = %d", status) + } + time.Sleep(80 * time.Millisecond) + mu.Lock() + if len(handled) != 1 { + got := append([]string(nil), handled...) + mu.Unlock() + t.Fatalf("successor window dispatched while the current batch was running, handled = %v", got) + } + mu.Unlock() + + close(releaseFirst) + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("successor batch did not dispatch after the current batch finished") + case <-time.After(5 * time.Millisecond): + } + } + + mu.Lock() + defer mu.Unlock() + if !equalStrings(handled, []string{"a", "b"}) { + t.Fatalf("handled = %v, want [a b] with windows serialized", handled) + } +} + +func TestBurstDispatchBudgetStartsWhenWindowCloses(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + // The window consumes most of one DetachTimeout; the batch must still get a + // full execution budget of its own once the window closes. + bot := newBurstRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.DebounceInterval = 400 * time.Millisecond + o.DetachTimeout = 500 * time.Millisecond + }) + + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + time.Sleep(60 * time.Millisecond) + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + return nil + }) + + for _, id := range []string{"a", "b", "c"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("%s status = %d", id, status) + } + } + + deadline := time.After(3 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 3 { + break + } + select { + case <-deadline: + mu.Lock() + got := append([]string(nil), handled...) + mu.Unlock() + t.Fatalf("collection time consumed the batch's execution budget, handled = %v", got) + case <-time.After(5 * time.Millisecond): + } + } + if strings.Contains(logs.String(), "chat burst batch abandoned") { + t.Fatalf("batch abandoned despite a fresh post-window budget; logs:\n%s", logs.String()) + } +} + +func TestBurstEveryMemberGetsItsOwnExecutionBudget(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + // Five 60ms handlers = 300ms total, above the 200ms DetachTimeout: with a + // shared batch deadline the tail members would be skipped, but each + // accepted member gets its own budget. + bot := newBurstRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.DebounceInterval = 50 * time.Millisecond + o.DetachTimeout = 200 * time.Millisecond + }) + + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + time.Sleep(60 * time.Millisecond) + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + return nil + }) + + ids := []string{"a", "b", "c", "d", "e"} + for _, id := range ids { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("%s status = %d", id, status) + } + } + + deadline := time.After(3 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= len(ids) { + break + } + select { + case <-deadline: + mu.Lock() + got := append([]string(nil), handled...) + mu.Unlock() + t.Fatalf("accepted batch members were skipped on a shared deadline, handled = %v", got) + case <-time.After(5 * time.Millisecond): + } + } + if strings.Contains(logs.String(), "chat burst batch abandoned") { + t.Fatalf("batch abandoned despite per-member budgets; logs:\n%s", logs.String()) + } +} + +func TestBurstAbandonedWaiterSurfacesObservation(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.ThreadLockTTL = time.Hour + o.DebounceInterval = 20 * time.Millisecond + o.DetachTimeout = 80 * time.Millisecond + }) + + var mu sync.Mutex + var calls int + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + calls++ + mu.Unlock() + return nil + }) + + // Hold the lock with an outside token that never releases. + if _, acquired, err := state.AcquireLock(context.Background(), "fake:v1:thread-1", time.Hour); err != nil || !acquired { + t.Fatalf("seed lock acquired=%v err=%v", acquired, err) + } + + if status := postEvent(t, bot, "fake", mentionEvent("abandoned", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("status = %d", status) + } + + deadline := time.After(2 * time.Second) + for !strings.Contains(logs.String(), "chat burst wait abandoned") { + select { + case <-deadline: + t.Fatalf("abandonment not surfaced via observation; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + + mu.Lock() + defer mu.Unlock() + if calls != 0 { + t.Fatalf("handler ran despite never acquiring the lock, calls = %d", calls) + } +} + +func TestBurstCollectingWindowDrainedByShutdown(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newBurstRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + // Only Shutdown can close the collection window. + o.DebounceInterval = time.Hour + o.DetachTimeout = 2 * time.Hour + }) + + var mu sync.Mutex + var calls int + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + calls++ + mu.Unlock() + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("collected", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("status = %d", status) + } + time.Sleep(40 * time.Millisecond) + + done := make(chan error, 1) + go func() { done <- bot.Shutdown(context.Background()) }() + select { + case err := <-done: + if err != nil { + t.Fatalf("shutdown error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("shutdown did not drain the collecting burst runner") + } + + if !strings.Contains(logs.String(), "chat burst wait abandoned") { + t.Fatalf("shutdown-cancelled window not surfaced as abandoned; logs:\n%s", logs.String()) + } + mu.Lock() + defer mu.Unlock() + if calls != 0 { + t.Fatalf("handler ran despite shutdown before the window closed, calls = %d", calls) + } +} + +func TestBurstConstructionValidation(t *testing.T) { + t.Parallel() + + newRuntime := func(mutate func(*chat.RuntimeOptions)) error { + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyBurst, + DebounceInterval: 50 * time.Millisecond, + Dispatch: chat.DispatchDeferred, + DetachTimeout: time.Second, + } + mutate(&options) + _, err := chat.New(context.Background(), + chat.WithState(newFakeState()), + chat.WithAdapter(newFakeAdapter("fake")), + chat.WithRuntimeOptions(options), + ) + return err + } + + if err := newRuntime(func(o *chat.RuntimeOptions) {}); err != nil { + t.Fatalf("valid burst options rejected: %v", err) + } + if err := newRuntime(func(o *chat.RuntimeOptions) { o.DebounceInterval = 0 }); err == nil { + t.Fatal("expected burst without an interval to fail") + } + if err := newRuntime(func(o *chat.RuntimeOptions) { + o.Dispatch = chat.DispatchSync + o.DetachTimeout = 0 + }); err == nil { + t.Fatal("expected burst under sync dispatch to fail") + } + // A detach timeout at or below the interval would abandon every window + // before it closes. + if err := newRuntime(func(o *chat.RuntimeOptions) { + o.DebounceInterval = time.Second + o.DetachTimeout = time.Second + }); err == nil { + t.Fatal("expected a detach timeout at or below the collection window to fail") + } +} diff --git a/concurrent_test.go b/concurrent_test.go new file mode 100644 index 0000000..c3998bf --- /dev/null +++ b/concurrent_test.go @@ -0,0 +1,285 @@ +package chat_test + +import ( + "context" + "log/slog" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/chat" +) + +// lockCountingState wraps fakeState to prove the concurrent strategy never +// touches the Thread Lock. +type lockCountingState struct { + *fakeState + countMu sync.Mutex + acquires int +} + +func (s *lockCountingState) AcquireLock(ctx context.Context, key string, ttl time.Duration) (chat.LockLease, bool, error) { + s.countMu.Lock() + s.acquires++ + s.countMu.Unlock() + return s.fakeState.AcquireLock(ctx, key, ttl) +} + +func (s *lockCountingState) acquireCalls() int { + s.countMu.Lock() + defer s.countMu.Unlock() + return s.acquires +} + +func newConcurrentRuntime(t *testing.T, state chat.State, adapter chat.Adapter, logs *syncBuffer, mutate ...func(*chat.RuntimeOptions)) *chat.Chat { + t.Helper() + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyConcurrent, + MaxConcurrent: 4, + Dispatch: chat.DispatchDeferred, + DetachTimeout: 5 * time.Second, + } + for _, m := range mutate { + m(&options) + } + bot, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithLogger(slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + chat.WithRuntimeOptions(options), + ) + if err != nil { + t.Fatalf("new concurrent runtime: %v", err) + } + return bot +} + +func TestConcurrentRunsSameThreadHandlersInParallelWithoutLock(t *testing.T) { + t.Parallel() + + state := &lockCountingState{fakeState: newFakeState()} + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newConcurrentRuntime(t, state, adapter, &logs) + + // The first handler only finishes once the second has started: proof that + // two handlers for the SAME Thread run concurrently (no serialization). + secondStarted := make(chan struct{}) + firstDone := make(chan struct{}) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + switch ev.Event.ID { + case "first": + select { + case <-secondStarted: + close(firstDone) + case <-ctx.Done(): + } + case "second": + close(secondStarted) + } + return nil + }) + + for _, id := range []string{"first", "second"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("%s status = %d", id, status) + } + } + + select { + case <-firstDone: + case <-time.After(2 * time.Second): + t.Fatal("handlers for the same thread did not run concurrently") + } + + if calls := state.acquireCalls(); calls != 0 { + t.Fatalf("concurrent strategy acquired the thread lock %d times, want 0", calls) + } + if strings.Contains(logs.String(), "chat lock conflict dropped") { + t.Fatalf("concurrent strategy surfaced a lock conflict; logs:\n%s", logs.String()) + } +} + +func TestConcurrentBoundedByMaxConcurrent(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newConcurrentRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.MaxConcurrent = 1 + }) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + // The second event is acked promptly but must wait for the single slot. + if status := postEvent(t, bot, "fake", mentionEvent("second", "fake:v1:thread-2")); status != http.StatusOK { + t.Fatalf("second status = %d", status) + } + time.Sleep(80 * time.Millisecond) + mu.Lock() + if len(handled) != 1 { + got := append([]string(nil), handled...) + mu.Unlock() + t.Fatalf("MaxConcurrent=1 did not bound execution, handled = %v", got) + } + mu.Unlock() + + close(releaseFirst) + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("second handler did not run after the slot freed") + case <-time.After(5 * time.Millisecond): + } + } +} + +func TestConcurrentSlotWaiterDrainedByShutdown(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newConcurrentRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.MaxConcurrent = 1 + o.DetachTimeout = time.Hour + }) + + firstStarted := make(chan struct{}) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "first" { + close(firstStarted) + // Occupy the only slot until the detached context is cancelled. + <-ctx.Done() + return ctx.Err() + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + if status := postEvent(t, bot, "fake", mentionEvent("waiter", "fake:v1:thread-2")); status != http.StatusOK { + t.Fatalf("waiter status = %d", status) + } + time.Sleep(40 * time.Millisecond) + + done := make(chan error, 1) + go func() { done <- bot.Shutdown(context.Background()) }() + select { + case err := <-done: + if err != nil { + t.Fatalf("shutdown error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("shutdown did not drain the slot waiter") + } + + if !strings.Contains(logs.String(), "chat concurrent slot wait abandoned") { + t.Fatalf("abandoned slot wait not surfaced via observation; logs:\n%s", logs.String()) + } +} + +func TestConcurrentSyncDispatchRunsWithoutLock(t *testing.T) { + t.Parallel() + + state := &lockCountingState{fakeState: newFakeState()} + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newConcurrentRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.Dispatch = chat.DispatchSync + o.DetachTimeout = 0 + }) + + handled := make(chan string, 1) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + handled <- ev.Event.ID + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("sync", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("status = %d", status) + } + select { + case id := <-handled: + if id != "sync" { + t.Fatalf("handled %q, want sync", id) + } + default: + t.Fatal("sync dispatch returned before the handler ran") + } + if calls := state.acquireCalls(); calls != 0 { + t.Fatalf("concurrent strategy acquired the thread lock %d times, want 0", calls) + } +} + +func TestConcurrentConstructionValidation(t *testing.T) { + t.Parallel() + + newRuntime := func(mutate func(*chat.RuntimeOptions)) error { + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyConcurrent, + MaxConcurrent: 2, + } + mutate(&options) + _, err := chat.New(context.Background(), + chat.WithState(newFakeState()), + chat.WithAdapter(newFakeAdapter("fake")), + chat.WithRuntimeOptions(options), + ) + return err + } + + if err := newRuntime(func(o *chat.RuntimeOptions) {}); err != nil { + t.Fatalf("valid concurrent options rejected: %v", err) + } + if err := newRuntime(func(o *chat.RuntimeOptions) { o.MaxConcurrent = 0 }); err == nil { + t.Fatal("expected concurrent without a max concurrent bound to fail") + } + if err := newRuntime(func(o *chat.RuntimeOptions) { o.MaxConcurrent = -1 }); err == nil { + t.Fatal("expected a negative max concurrent bound to fail") + } +} diff --git a/debounce_test.go b/debounce_test.go new file mode 100644 index 0000000..e82d839 --- /dev/null +++ b/debounce_test.go @@ -0,0 +1,463 @@ +package chat_test + +import ( + "context" + "log/slog" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/chat" +) + +func newDebounceRuntime(t *testing.T, state chat.State, adapter chat.Adapter, logs *syncBuffer, mutate ...func(*chat.RuntimeOptions)) *chat.Chat { + t.Helper() + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: 40 * time.Millisecond, + Concurrency: chat.ConcurrencyDebounce, + DebounceInterval: 50 * time.Millisecond, + Dispatch: chat.DispatchDeferred, + DetachTimeout: 5 * time.Second, + } + for _, m := range mutate { + m(&options) + } + bot, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithLogger(slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + chat.WithRuntimeOptions(options), + ) + if err != nil { + t.Fatalf("new debounce runtime: %v", err) + } + return bot +} + +func TestDebounceCoalescesRapidFollowUpsToFinalEvent(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newDebounceRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.DebounceInterval = 150 * time.Millisecond + }) + + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + return nil + }) + + // Three events arrive well inside one 150ms quiet period: only the final + // one may dispatch. + for _, id := range []string{"first", "second", "third"} { + if status := postEvent(t, bot, "fake", mentionEvent(id, "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("%s status = %d", id, status) + } + time.Sleep(5 * time.Millisecond) + } + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 1 { + break + } + select { + case <-deadline: + t.Fatal("debounce winner did not dispatch after the quiet period") + case <-time.After(5 * time.Millisecond): + } + } + + // Give any (incorrect) superseded dispatch time to surface. + time.Sleep(100 * time.Millisecond) + mu.Lock() + got := append([]string(nil), handled...) + mu.Unlock() + if len(got) != 1 || got[0] != "third" { + t.Fatalf("handled = %v, want only the final event [third]", got) + } + + out := logs.String() + if !strings.Contains(out, "chat debounce superseded") { + t.Fatalf("superseded events not surfaced via observation; logs:\n%s", out) + } + if !strings.Contains(out, "superseded_by=third") { + t.Fatalf("supersession did not carry superseded_by=third; logs:\n%s", out) + } + // A debounce coalesce is never a Lock Conflict drop. + if strings.Contains(out, "chat lock conflict dropped") { + t.Fatalf("debounce surfaced a lock-conflict drop; logs:\n%s", out) + } +} + +func TestDebounceAcksPromptlyAndDispatchesAfterQuietPeriod(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newDebounceRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.DebounceInterval = 400 * time.Millisecond + }) + + handled := make(chan string, 1) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + handled <- ev.Event.ID + return nil + }) + + start := time.Now() + if status := postEvent(t, bot, "fake", mentionEvent("lone", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("status = %d", status) + } + if ackLatency := time.Since(start); ackLatency > 200*time.Millisecond { + t.Fatalf("ack blocked on the quiet period (%v); debounce must be ack-then-work", ackLatency) + } + + select { + case id := <-handled: + if id != "lone" { + t.Fatalf("handled %q, want lone", id) + } + if elapsed := time.Since(start); elapsed < 400*time.Millisecond { + t.Fatalf("handler ran %v after post, before the quiet period elapsed", elapsed) + } + case <-time.After(2 * time.Second): + t.Fatal("debounced event never dispatched") + } +} + +func TestDebounceWinnerWaitsForInflightLockRelease(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newDebounceRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.ThreadLockTTL = time.Hour + o.DebounceInterval = 20 * time.Millisecond + }) + + handled := make(chan string, 1) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + handled <- ev.Event.ID + return nil + }) + + // Hold the scope's lock with an outside token so the quiet-period winner + // must wait for release. + lease, acquired, err := state.AcquireLock(context.Background(), "fake:v1:thread-1", time.Hour) + if err != nil || !acquired { + t.Fatalf("seed lock acquired=%v err=%v", acquired, err) + } + + if status := postEvent(t, bot, "fake", mentionEvent("waiter", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("status = %d", status) + } + + // Well past the quiet period the handler must still be blocked on the lock. + time.Sleep(100 * time.Millisecond) + select { + case id := <-handled: + t.Fatalf("handler %q ran while the lock was held", id) + default: + } + + if released, err := state.ReleaseLock(context.Background(), lease); err != nil || !released { + t.Fatalf("release seed lock released=%v err=%v", released, err) + } + + select { + case id := <-handled: + if id != "waiter" { + t.Fatalf("handled %q, want waiter", id) + } + case <-time.After(2 * time.Second): + t.Fatal("debounce winner did not run after the lock released") + } +} + +func TestDebounceAbandonedWaiterSurfacesObservation(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newDebounceRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.ThreadLockTTL = time.Hour + o.DebounceInterval = 20 * time.Millisecond + // The detached waiter abandons at DetachTimeout while the outside lock + // never releases. + o.DetachTimeout = 80 * time.Millisecond + }) + + var mu sync.Mutex + var calls int + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + calls++ + mu.Unlock() + return nil + }) + + if _, acquired, err := state.AcquireLock(context.Background(), "fake:v1:thread-1", time.Hour); err != nil || !acquired { + t.Fatalf("seed lock acquired=%v err=%v", acquired, err) + } + + if status := postEvent(t, bot, "fake", mentionEvent("abandoned", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("status = %d", status) + } + + deadline := time.After(2 * time.Second) + for !strings.Contains(logs.String(), "chat debounce wait abandoned") { + select { + case <-deadline: + t.Fatalf("abandonment not surfaced via observation; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + + mu.Lock() + defer mu.Unlock() + if calls != 0 { + t.Fatalf("handler ran despite never acquiring the lock, calls = %d", calls) + } +} + +func TestDebounceSleepingWaiterDrainedByShutdown(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newDebounceRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + // Only Shutdown can end the quiet-period sleep. + o.DebounceInterval = time.Hour + o.DetachTimeout = 2 * time.Hour + }) + + var mu sync.Mutex + var calls int + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + calls++ + mu.Unlock() + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("sleeper", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("status = %d", status) + } + time.Sleep(40 * time.Millisecond) + + done := make(chan error, 1) + go func() { done <- bot.Shutdown(context.Background()) }() + select { + case err := <-done: + if err != nil { + t.Fatalf("shutdown error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("shutdown did not drain the sleeping debounce waiter") + } + + if !strings.Contains(logs.String(), "chat debounce wait abandoned") { + t.Fatalf("shutdown-cancelled waiter not surfaced as abandoned; logs:\n%s", logs.String()) + } + mu.Lock() + defer mu.Unlock() + if calls != 0 { + t.Fatalf("handler ran despite shutdown before the quiet period, calls = %d", calls) + } +} + +// acquireGatedState wraps fakeState so a test can hold a tail's AcquireLock +// call open and register a newer event while the acquire is in flight. +type acquireGatedState struct { + *fakeState + mu sync.Mutex + gate chan struct{} + entered chan struct{} + armed bool +} + +func newAcquireGatedState() *acquireGatedState { + return &acquireGatedState{ + fakeState: newFakeState(), + gate: make(chan struct{}), + entered: make(chan struct{}), + armed: true, + } +} + +func (s *acquireGatedState) AcquireLock(ctx context.Context, key string, ttl time.Duration) (chat.LockLease, bool, error) { + s.mu.Lock() + first := s.armed + s.armed = false + s.mu.Unlock() + if first { + close(s.entered) + <-s.gate + } + return s.fakeState.AcquireLock(ctx, key, ttl) +} + +func TestDebounceWaiterDisplacedDuringAcquireDoesNotDispatch(t *testing.T) { + t.Parallel() + + state := newAcquireGatedState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newDebounceRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.DebounceInterval = 20 * time.Millisecond + }) + + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("older", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("older status = %d", status) + } + + // Hold older's post-quiet-period AcquireLock open, then register the newer + // event while the acquire is in flight. + select { + case <-state.entered: + case <-time.After(2 * time.Second): + t.Fatal("older waiter never reached AcquireLock") + } + if status := postEvent(t, bot, "fake", mentionEvent("newer", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("newer status = %d", status) + } + close(state.gate) + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 1 { + break + } + select { + case <-deadline: + t.Fatal("debounce winner did not dispatch") + case <-time.After(5 * time.Millisecond): + } + } + time.Sleep(60 * time.Millisecond) + + mu.Lock() + defer mu.Unlock() + if !equalStrings(handled, []string{"newer"}) { + t.Fatalf("handled = %v, want only [newer]: a waiter displaced mid-acquire must not dispatch", handled) + } +} + +func TestDebounceConstructionValidation(t *testing.T) { + t.Parallel() + + newRuntime := func(mutate func(*chat.RuntimeOptions)) error { + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyDebounce, + DebounceInterval: 50 * time.Millisecond, + Dispatch: chat.DispatchDeferred, + DetachTimeout: time.Second, + } + mutate(&options) + _, err := chat.New(context.Background(), + chat.WithState(newFakeState()), + chat.WithAdapter(newFakeAdapter("fake")), + chat.WithRuntimeOptions(options), + ) + return err + } + + if err := newRuntime(func(o *chat.RuntimeOptions) {}); err != nil { + t.Fatalf("valid debounce options rejected: %v", err) + } + if err := newRuntime(func(o *chat.RuntimeOptions) { o.DebounceInterval = 0 }); err == nil { + t.Fatal("expected debounce without an interval to fail") + } + if err := newRuntime(func(o *chat.RuntimeOptions) { + o.Dispatch = chat.DispatchSync + o.DetachTimeout = 0 + }); err == nil { + t.Fatal("expected debounce under sync dispatch to fail") + } + // A detach timeout at or below the interval would abandon every event + // before its quiet period elapses. + if err := newRuntime(func(o *chat.RuntimeOptions) { + o.DebounceInterval = time.Second + o.DetachTimeout = time.Second + }); err == nil { + t.Fatal("expected a detach timeout at or below the debounce interval to fail") + } +} + +func TestDebounceSupersededWaiterExitsBeforeItsInterval(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newDebounceRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + // The interval is far longer than the test: only the displacement signal + // can explain a prompt exit. + o.DebounceInterval = time.Hour + o.DetachTimeout = 2 * time.Hour + }) + + var mu sync.Mutex + var calls int + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + calls++ + mu.Unlock() + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("older", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("older status = %d", status) + } + if status := postEvent(t, bot, "fake", mentionEvent("newer", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("newer status = %d", status) + } + + // The superseded waiter must release its goroutine and event promptly, not + // park through the hour-long interval. + deadline := time.After(2 * time.Second) + for !strings.Contains(logs.String(), "chat debounce waiter superseded") { + select { + case <-deadline: + t.Fatalf("superseded waiter did not exit promptly; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + + mu.Lock() + defer mu.Unlock() + if calls != 0 { + t.Fatalf("handler ran before any quiet period elapsed, calls = %d", calls) + } +} diff --git a/dispatch_deferred_test.go b/dispatch_deferred_test.go index 532a5c8..1ab3649 100644 --- a/dispatch_deferred_test.go +++ b/dispatch_deferred_test.go @@ -529,7 +529,7 @@ func TestRuntimeConstructionValidatesDispatchAndConcurrency(t *testing.T) { t.Fatalf("queue concurrency should be accepted: %v", err) } - // Unimplemented strategies (burst/debounce/concurrent) remain rejected. + // Unknown strategy values remain rejected. if err := newRuntime(chat.RuntimeOptions{ DedupeTTL: time.Hour, ThreadLockTTL: time.Hour, @@ -757,6 +757,77 @@ func TestDeferredDispatchObservationRecords(t *testing.T) { } } +// TestDeferredHandlerCancelledOnLeaseLossWithoutHook proves that every +// deferred lock holder stops on lease loss, even when the runtime has no +// OnLockConflict hook of its own: a lease force released elsewhere (or +// expired) means mutual exclusion is gone, and the handler observes +// chat.ErrPreempted as its cancellation cause. +func TestDeferredHandlerCancelledOnLeaseLossWithoutHook(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithLogger(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + chat.WithRuntimeOptions(chat.RuntimeOptions{ + DedupeTTL: time.Hour, + // A short TTL keeps the refresh cadence (TTL/2) fast. + ThreadLockTTL: 100 * time.Millisecond, + Concurrency: chat.ConcurrencyDrop, + Dispatch: chat.DispatchDeferred, + DetachTimeout: 5 * time.Second, + }), + ) + if err != nil { + t.Fatalf("new runtime: %v", err) + } + + started := make(chan struct{}) + cause := make(chan error, 1) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + close(started) + <-ctx.Done() + cause <- context.Cause(ctx) + return ctx.Err() + }) + + if status := postEvent(t, bot, "fake", mentionEvent("holder", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("status = %d", status) + } + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("handler did not start") + } + + // Another runtime instance force releases the lease out from under the + // handler; the next refresh must observe the loss and cancel it. + if released, err := state.ForceReleaseLock(context.Background(), "fake:v1:thread-1"); err != nil || !released { + t.Fatalf("external force release released=%v err=%v", released, err) + } + + select { + case got := <-cause: + if !errors.Is(got, chat.ErrPreempted) { + t.Fatalf("cancellation cause = %v, want chat.ErrPreempted", got) + } + case <-time.After(2 * time.Second): + t.Fatal("handler was not cancelled after its lease was force released") + } + + deadline := time.After(2 * time.Second) + for !strings.Contains(logs.String(), "chat handler preempted") { + select { + case <-deadline: + t.Fatalf("lease-loss cancellation not surfaced; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } +} + // syncBuffer is a goroutine-safe bytes.Buffer for capturing logs from detached // tails. type syncBuffer struct { diff --git a/docs/adr/0012-concurrency-strategy.md b/docs/adr/0012-concurrency-strategy.md index 900a471..79a2b58 100644 --- a/docs/adr/0012-concurrency-strategy.md +++ b/docs/adr/0012-concurrency-strategy.md @@ -2,7 +2,7 @@ ## Status -Accepted (implementation staged: the `queue` strategy is implemented as the `DispatchDeferred` companion; `burst`, `debounce`, `concurrent`, `lockScope`, and force/steerability remain proposed and unbuilt). +Accepted (implemented: `drop`, `queue`, `debounce`, `burst`, and `concurrent` strategies, the `LockScope` option, and force/steerability via `RuntimeOptions.OnLockConflict` + the `LockForcer` **Runtime State** capability all ship in the runtime; `debounce`/`burst` and the steerability hook require `DispatchDeferred`). ## Context diff --git a/docs/explanation.md b/docs/explanation.md index 7f236fc..957afe3 100644 --- a/docs/explanation.md +++ b/docs/explanation.md @@ -37,7 +37,7 @@ design deliberately diverges from Vercel Chat SDK. | [0009](adr/0009-message-history.md) | Message history stays application-owned; optional storage-free `HistoryReader` | Accepted | | [0010](adr/0010-observability.md) | Optional `Observer` seam; no OpenTelemetry in core | Accepted | | [0011](adr/0011-resumable-streaming.md) | Resumable streaming deferred from core, not foreclosed | Proposed | -| [0012](adr/0012-concurrency-strategy.md) | Concurrency strategy expansion (`queue` implemented; `burst`/`debounce`/`concurrent` staged) | Accepted (staged) | +| [0012](adr/0012-concurrency-strategy.md) | Concurrency strategy expansion (`drop`/`queue`/`debounce`/`burst`/`concurrent`, lock scope, force/steerability) | Accepted (implemented) | | [0013](adr/0013-linear-generic-comments.md) | Linear generic issue/comment participation | Accepted | | [0014](adr/0014-nats-state-adapter.md) | NATS JetStream state adapter | Accepted | diff --git a/docs/how-to/linear-agent-sessions.md b/docs/how-to/linear-agent-sessions.md index 79a731c..66dfd8e 100644 --- a/docs/how-to/linear-agent-sessions.md +++ b/docs/how-to/linear-agent-sessions.md @@ -136,11 +136,13 @@ if raw, ok := linear.RawMessageFrom(ev.Message); ok && raw.StopRequested() { ``` This check only runs when the stop event reaches your handler, and events on -one thread are serialized by the thread lock — a stop arriving while a -handler is still running cannot preempt it (`ConcurrencyDrop` discards it on -conflict; `ConcurrencyQueue` delivers it only after the in-flight handler -returns). There is no pre-lock hook, so **Linear's Stop control cannot cancel -in-flight work through this adapter today**. What you can do: structure long +one thread are serialized by the thread lock — by default a stop arriving +while a handler is still running does not preempt it (`ConcurrencyDrop` +discards it on conflict; `ConcurrencyQueue` delivers it only after the +in-flight handler returns). Under deferred dispatch you can opt into +preemption with the `RuntimeOptions.OnLockConflict` hook (ADR 0012): return +true for a stop event and the in-flight handler's context is cancelled with +`chat.ErrPreempted`. Without that hook, structure long sessions as short handler turns (each turn checks `StopRequested` on the event that started it before doing more work), or receive the stop signal out-of-band through your own channel (for example, your own Linear webhook diff --git a/errors.go b/errors.go index 2c697aa..8b36938 100644 --- a/errors.go +++ b/errors.go @@ -4,6 +4,12 @@ import "errors" var ( ErrUnsupportedCapability = errors.New("chat: unsupported adapter capability") + // ErrPreempted is the cancellation cause a stopped handler observes via + // context.Cause: a new delivery preempted it through the OnLockConflict + // steerability hook, or its Lock Lease was lost (force released by another + // runtime instance, or expired) so mutual exclusion could no longer be + // guaranteed. + ErrPreempted = errors.New("chat: handler preempted") ) func assert(ok bool, message string) { diff --git a/internal/statetest/state.go b/internal/statetest/state.go index b666b5d..9a04251 100644 --- a/internal/statetest/state.go +++ b/internal/statetest/state.go @@ -159,6 +159,71 @@ func RunStateConformance(t *testing.T, newState func(*testing.T) Harness) { } }) + t.Run("force release", func(t *testing.T) { + t.Parallel() + harness := newState(t) + state := harness.State + forcer, ok := state.(chat.LockForcer) + if !ok { + t.Fatal("state does not implement chat.LockForcer") + } + + released, err := forcer.ForceReleaseLock(context.Background(), "thread-force") + if err != nil { + t.Fatalf("force release unheld lock: %v", err) + } + if released { + t.Fatal("force release of an unheld lock should report false") + } + + lease, acquired, err := state.AcquireLock(context.Background(), "thread-force", time.Minute) + if err != nil { + t.Fatalf("acquire lock: %v", err) + } + if !acquired { + t.Fatal("first lock acquire should succeed") + } + + released, err = forcer.ForceReleaseLock(context.Background(), "thread-force") + if err != nil { + t.Fatalf("force release held lock: %v", err) + } + if !released { + t.Fatal("force release of a held lock should report true") + } + + // The invalidated lease must not extend the key or steal a newer lock. + extended, err := state.ExtendLock(context.Background(), lease, time.Minute) + if err != nil { + t.Fatalf("extend force-released lease: %v", err) + } + if extended { + t.Fatal("force-released lease should not extend") + } + + fresh, acquired, err := state.AcquireLock(context.Background(), "thread-force", time.Minute) + if err != nil { + t.Fatalf("acquire after force release: %v", err) + } + if !acquired { + t.Fatal("acquire should succeed after force release") + } + released, err = state.ReleaseLock(context.Background(), lease) + if err != nil { + t.Fatalf("release force-released lease: %v", err) + } + if released { + t.Fatal("force-released lease should not release the fresh lock") + } + released, err = state.ReleaseLock(context.Background(), fresh) + if err != nil { + t.Fatalf("release fresh lock: %v", err) + } + if !released { + t.Fatal("fresh owner should release its lock") + } + }) + t.Run("context cancellation", func(t *testing.T) { t.Parallel() harness := newState(t) @@ -195,6 +260,11 @@ func RunStateConformance(t *testing.T, newState func(*testing.T) Harness) { if _, _, err := state.AcquireLock(ctx, lockKey, time.Minute); err == nil { t.Fatal("expected cancelled context to stop lock mutation") } + if forcer, ok := state.(chat.LockForcer); ok { + if _, err := forcer.ForceReleaseLock(ctx, lockKey); err == nil { + t.Fatal("expected cancelled context to stop force release") + } + } _, acquired, err := state.AcquireLock(bg, lockKey, time.Minute) if err != nil { t.Fatalf("acquire lock after cancelled acquire: %v", err) diff --git a/lockscope_test.go b/lockscope_test.go new file mode 100644 index 0000000..abdafb2 --- /dev/null +++ b/lockscope_test.go @@ -0,0 +1,324 @@ +package chat_test + +import ( + "context" + "log/slog" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/chat" +) + +// channelAdapter wraps fakeAdapter so thread ids of the form +// "fake:v1::" report a channel on their ThreadRef; shorter ids +// and ids carrying the runtime's synthesized channel-key prefix (used to prove +// fallback keys cannot collide) report no channel. +type channelAdapter struct { + *fakeAdapter +} + +func (a *channelAdapter) ValidateThreadID(id chat.ThreadID) (chat.ThreadRef, error) { + ref, err := a.fakeAdapter.ValidateThreadID(id) + if err != nil { + return ref, err + } + ref.Channel = "" + if parts := strings.Split(string(id), ":"); len(parts) >= 4 && !strings.HasPrefix(string(id), "channel-scope/") { + ref.Channel = parts[2] + } + return ref, nil +} + +func newLockScopeRuntime(t *testing.T, state chat.State, adapter chat.Adapter, logs *syncBuffer, mutate ...func(*chat.RuntimeOptions)) *chat.Chat { + t.Helper() + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyDrop, + LockScope: chat.LockScopeChannel, + Dispatch: chat.DispatchDeferred, + DetachTimeout: 5 * time.Second, + } + for _, m := range mutate { + m(&options) + } + bot, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithLogger(slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + chat.WithRuntimeOptions(options), + ) + if err != nil { + t.Fatalf("new lock scope runtime: %v", err) + } + return bot +} + +func TestChannelScopeSerializesThreadsSharingAChannel(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := &channelAdapter{fakeAdapter: newFakeAdapter("fake")} + var logs syncBuffer + bot := newLockScopeRuntime(t, state, adapter, &logs) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + defer close(releaseFirst) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:chan-a:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + // A DIFFERENT thread in the SAME channel is a Lock Conflict under channel + // scope and drops. + if status := postEvent(t, bot, "fake", mentionEvent("same-channel", "fake:v1:chan-a:thread-2")); status != http.StatusOK { + t.Fatalf("same-channel status = %d", status) + } + deadline := time.After(2 * time.Second) + for !strings.Contains(logs.String(), "chat lock conflict dropped") { + select { + case <-deadline: + t.Fatalf("channel-scope conflict not surfaced; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + + // A thread in ANOTHER channel does not conflict and runs. + if status := postEvent(t, bot, "fake", mentionEvent("other-channel", "fake:v1:chan-b:thread-3")); status != http.StatusOK { + t.Fatalf("other-channel status = %d", status) + } + deadline = time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("other-channel event did not run despite a distinct channel scope") + case <-time.After(5 * time.Millisecond): + } + } + + mu.Lock() + defer mu.Unlock() + if !equalStrings(handled, []string{"first", "other-channel"}) { + t.Fatalf("handled = %v, want [first other-channel] (same-channel dropped)", handled) + } +} + +func TestThreadScopeDefaultDoesNotSerializeAcrossThreads(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := &channelAdapter{fakeAdapter: newFakeAdapter("fake")} + var logs syncBuffer + // Identical setup, but with the default thread scope: two threads in one + // channel never conflict. + bot := newLockScopeRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.LockScope = chat.LockScopeThread + }) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + defer close(releaseFirst) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:chan-a:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + if status := postEvent(t, bot, "fake", mentionEvent("sibling", "fake:v1:chan-a:thread-2")); status != http.StatusOK { + t.Fatalf("sibling status = %d", status) + } + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("sibling thread blocked under the default thread scope") + case <-time.After(5 * time.Millisecond): + } + } + if strings.Contains(logs.String(), "chat lock conflict dropped") { + t.Fatalf("thread scope surfaced a cross-thread conflict; logs:\n%s", logs.String()) + } +} + +func TestChannelScopeFallsBackToThreadWhenChannelEmpty(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := &channelAdapter{fakeAdapter: newFakeAdapter("fake")} + var logs syncBuffer + bot := newLockScopeRuntime(t, state, adapter, &logs) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + defer close(releaseFirst) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + } + return nil + }) + + // Short thread ids report no channel: each falls back to its own + // per-thread key rather than sharing one adapter-wide lock. + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + if status := postEvent(t, bot, "fake", mentionEvent("second", "fake:v1:thread-2")); status != http.StatusOK { + t.Fatalf("second status = %d", status) + } + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("channelless threads shared one lock; fallback to thread scope failed") + case <-time.After(5 * time.Millisecond): + } + } +} + +func TestChannelScopeFallbackKeyCannotCollideWithChannelKey(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := &channelAdapter{fakeAdapter: newFakeAdapter("fake")} + var logs syncBuffer + bot := newLockScopeRuntime(t, state, adapter, &logs) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + defer close(releaseFirst) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + } + return nil + }) + + // The first event locks the synthesized key for channel chan-a. + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:chan-a:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + // A channelless thread whose opaque ID is exactly the synthesized channel + // key must not be treated as contention on chan-a. + collider := mentionEvent("collider", chat.ThreadID("channel-scope/4:fake/6:tenant/6:chan-a")) + if status := postEvent(t, bot, "fake", collider); status != http.StatusOK { + t.Fatalf("collider status = %d", status) + } + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatalf("crafted thread id collided with the channel key; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + if strings.Contains(logs.String(), "chat lock conflict dropped") { + t.Fatalf("fallback key collided with a channel key; logs:\n%s", logs.String()) + } +} + +func TestLockScopeConstructionValidation(t *testing.T) { + t.Parallel() + + _, err := chat.New(context.Background(), + chat.WithState(newFakeState()), + chat.WithAdapter(newFakeAdapter("fake")), + chat.WithRuntimeOptions(chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + LockScope: chat.LockScope(99), + }), + ) + if err == nil { + t.Fatal("expected an unknown lock scope to fail construction") + } +} diff --git a/observer.go b/observer.go index 7e84049..23d3e51 100644 --- a/observer.go +++ b/observer.go @@ -32,6 +32,11 @@ const ( // observes platform rate limiting (ADR 0005). Adapter-owned, like // ObsAdapterCall. ObsRateLimit ObservationName = "rate_limit" + // ObsLockPreempted is emitted when a new delivery preempts the scope's + // in-flight work via the OnLockConflict steerability hook: a local handler + // is cancelled and awaited, or a remote/orphaned Lock Lease is force + // released. + ObsLockPreempted ObservationName = "lock_preempted" ) // DispatchOutcome is the closed set of terminal outcomes for one Runtime @@ -44,6 +49,9 @@ const ( OutcomeDroppedLockConflict DispatchOutcome = "dropped-lock-conflict" OutcomeDuplicate DispatchOutcome = "duplicate" OutcomeError DispatchOutcome = "error" + // OutcomePreempted is the terminal outcome of a handler that stopped because + // a new delivery preempted it via the OnLockConflict steerability hook. + OutcomePreempted DispatchOutcome = "preempted" ) // Attribute key constants form the documented, stable, low-cardinality set. diff --git a/preempt_test.go b/preempt_test.go new file mode 100644 index 0000000..76aaad2 --- /dev/null +++ b/preempt_test.go @@ -0,0 +1,731 @@ +package chat_test + +import ( + "context" + "errors" + "log/slog" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/chat" +) + +func newPreemptRuntime(t *testing.T, state chat.State, adapter chat.Adapter, logs *syncBuffer, hook chat.LockConflictHook, mutate ...func(*chat.RuntimeOptions)) *chat.Chat { + t.Helper() + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyDrop, + Dispatch: chat.DispatchDeferred, + DetachTimeout: 5 * time.Second, + OnLockConflict: hook, + } + for _, m := range mutate { + m(&options) + } + bot, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithLogger(slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + chat.WithRuntimeOptions(options), + ) + if err != nil { + t.Fatalf("new preempt runtime: %v", err) + } + return bot +} + +func TestPreemptCancelsInflightHandlerAndDispatchesNewEvent(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + return true + }) + + firstStarted := make(chan struct{}) + firstCause := make(chan error, 1) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + if ev.Event.ID == "first" { + close(firstStarted) + // A preemptible handler observes cancellation and stops. + <-ctx.Done() + firstCause <- context.Cause(ctx) + return ctx.Err() + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + // The follow-up preempts: the in-flight handler is cancelled with + // ErrPreempted and the new event dispatches under a fresh lease. + if status := postEvent(t, bot, "fake", mentionEvent("preemptor", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("preemptor status = %d", status) + } + + select { + case cause := <-firstCause: + if !errors.Is(cause, chat.ErrPreempted) { + t.Fatalf("first handler cancellation cause = %v, want chat.ErrPreempted", cause) + } + case <-time.After(2 * time.Second): + t.Fatal("in-flight handler was not cancelled by the preempting delivery") + } + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("preempting event did not dispatch") + case <-time.After(5 * time.Millisecond): + } + } + mu.Lock() + if !equalStrings(handled, []string{"first", "preemptor"}) { + got := append([]string(nil), handled...) + mu.Unlock() + t.Fatalf("handled = %v, want [first preemptor]", got) + } + mu.Unlock() + + deadline = time.After(2 * time.Second) + for { + out := logs.String() + if strings.Contains(out, "chat lock preempted") && strings.Contains(out, "chat handler preempted") { + break + } + select { + case <-deadline: + t.Fatalf("preemption not surfaced via observation; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + + // The victim's failed lock release is benign (its lease was force + // released), never a WARN. + time.Sleep(50 * time.Millisecond) + if strings.Contains(logs.String(), `level=WARN msg="chat thread lock was not released"`) { + t.Fatalf("preempted handler's release surfaced as WARN; logs:\n%s", logs.String()) + } +} + +func TestPreemptWaitsForLocalVictimBeforeDispatching(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + return true + }) + + firstStarted := make(chan struct{}) + var mu sync.Mutex + var sequence []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "first" { + close(firstStarted) + <-ctx.Done() + // Cancellation-aware cleanup still runs; the preemptor must not + // overlap it. + time.Sleep(75 * time.Millisecond) + mu.Lock() + sequence = append(sequence, "victim-cleanup-done") + mu.Unlock() + return ctx.Err() + } + mu.Lock() + sequence = append(sequence, "preemptor-start") + mu.Unlock() + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + if status := postEvent(t, bot, "fake", mentionEvent("preemptor", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("preemptor status = %d", status) + } + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(sequence) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + mu.Lock() + got := append([]string(nil), sequence...) + mu.Unlock() + t.Fatalf("preemption did not complete, sequence = %v", got) + case <-time.After(5 * time.Millisecond): + } + } + + mu.Lock() + defer mu.Unlock() + if !equalStrings(sequence, []string{"victim-cleanup-done", "preemptor-start"}) { + t.Fatalf("sequence = %v, want the local victim to finish before the preemptor dispatches", sequence) + } +} + +func TestPreemptForceReleasesRemoteLeaseWhenNoLocalVictim(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + return true + }) + + handled := make(chan string, 1) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + handled <- ev.Event.ID + return nil + }) + + // A lease held outside this runtime (another instance, or orphaned): there + // is no local victim to cancel, so preemption must force-release it. + if _, acquired, err := state.AcquireLock(context.Background(), "fake:v1:thread-1", time.Hour); err != nil || !acquired { + t.Fatalf("seed lock acquired=%v err=%v", acquired, err) + } + + if status := postEvent(t, bot, "fake", mentionEvent("preemptor", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("preemptor status = %d", status) + } + + select { + case id := <-handled: + if id != "preemptor" { + t.Fatalf("handled %q, want preemptor", id) + } + case <-time.After(2 * time.Second): + t.Fatalf("preemptor did not dispatch after force release; logs:\n%s", logs.String()) + } + + out := logs.String() + if !strings.Contains(out, "forced=true") || !strings.Contains(out, "chat lock preempted") { + t.Fatalf("remote force release not surfaced; logs:\n%s", out) + } +} + +func TestPreemptSeesLocalOwnerAcquiredDuringPrelude(t *testing.T) { + t.Parallel() + + state := newFakeState() + routingStarted := make(chan struct{}) + continueRouting := make(chan struct{}) + state.isThreadSubscribedStarted = routingStarted + state.continueIsThreadSubscribed = continueRouting + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + return true + }) + + handled := make(chan string, 2) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + handled <- ev.Event.ID + return nil + }) + + // The first event acquires its lease in the prelude and then parks in + // routing: the window between acquisition and its detached tail must + // already carry a local ownership reservation. + firstDone := make(chan postEventResult, 1) + go func() { firstDone <- postEventResultFor(bot, "fake", mentionEvent("first", "fake:v1:thread-1")) }() + select { + case <-routingStarted: + case <-time.After(2 * time.Second): + t.Fatal("first event did not reach routing") + } + + // The preemptor arrives inside that window: it must wait for the local + // owner, never force-release the freshly acquired lease. + if status := postEvent(t, bot, "fake", mentionEvent("preemptor", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("preemptor status = %d", status) + } + time.Sleep(80 * time.Millisecond) + select { + case id := <-handled: + t.Fatalf("handler %q ran while the owner was still in its prelude", id) + default: + } + if strings.Contains(logs.String(), "forced=true") { + t.Fatalf("preemptor force released a fresh local lease; logs:\n%s", logs.String()) + } + + close(continueRouting) + res := <-firstDone + if res.err != nil || res.status != http.StatusOK { + t.Fatalf("first result status=%d err=%v", res.status, res.err) + } + + select { + case id := <-handled: + if id != "preemptor" { + t.Fatalf("handled %q, want preemptor (first was preempted before start)", id) + } + case <-time.After(2 * time.Second): + t.Fatalf("preemptor did not dispatch; logs:\n%s", logs.String()) + } + // The preempted-before-start owner must never run its handler. + time.Sleep(50 * time.Millisecond) + select { + case id := <-handled: + t.Fatalf("preempted-before-start handler %q still ran", id) + default: + } + if !strings.Contains(logs.String(), "started=false") { + t.Fatalf("pre-start preemption not surfaced; logs:\n%s", logs.String()) + } +} + +func TestPreemptSupersededPreemptorDoesNotForceRelease(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + return true + }) + + firstStarted := make(chan struct{}) + firstCancelled := make(chan struct{}) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + if ev.Event.ID == "first" { + close(firstStarted) + <-ctx.Done() + close(firstCancelled) + // Long cancellation-aware cleanup: both preemptors park on this + // victim, and the superseding one arrives while it sleeps. + time.Sleep(200 * time.Millisecond) + return ctx.Err() + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + // preempt-a cancels the victim, then parks waiting for it to finish. + if status := postEvent(t, bot, "fake", mentionEvent("preempt-a", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("preempt-a status = %d", status) + } + select { + case <-firstCancelled: + case <-time.After(2 * time.Second): + t.Fatal("victim was not cancelled") + } + // preempt-b supersedes preempt-a while the victim is still cleaning up. + if status := postEvent(t, bot, "fake", mentionEvent("preempt-b", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("preempt-b status = %d", status) + } + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("superseding preemptor did not dispatch") + case <-time.After(5 * time.Millisecond): + } + } + time.Sleep(50 * time.Millisecond) + + mu.Lock() + got := append([]string(nil), handled...) + mu.Unlock() + if !equalStrings(got, []string{"first", "preempt-b"}) { + t.Fatalf("handled = %v, want [first preempt-b] (superseded preemptor never dispatches)", got) + } + + out := logs.String() + if !strings.Contains(out, "chat preempt superseded") { + t.Fatalf("preemptor supersession not surfaced; logs:\n%s", out) + } + // The displaced preemptor must not have force released on behalf of an + // event that never dispatched: only preempt-b may act. + if strings.Contains(out, `msg="chat lock preempted" adapter=fake event_id=preempt-a`) { + t.Fatalf("superseded preemptor force released the lock; logs:\n%s", out) + } +} + +func TestPreemptHookDecliningFallsBackToDrop(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + var hookMu sync.Mutex + hookCalls := 0 + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + hookMu.Lock() + hookCalls++ + hookMu.Unlock() + return false + }) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + var mu sync.Mutex + var handled []string + firstInterrupted := false + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + if ctx.Err() != nil { + mu.Lock() + firstInterrupted = true + mu.Unlock() + } + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + if status := postEvent(t, bot, "fake", mentionEvent("declined", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("declined status = %d", status) + } + + deadline := time.After(2 * time.Second) + for !strings.Contains(logs.String(), "chat lock conflict dropped") { + select { + case <-deadline: + t.Fatalf("declined conflict was not dropped; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + hookMu.Lock() + if hookCalls != 1 { + hookMu.Unlock() + t.Fatalf("hook consulted %d times, want 1", hookCalls) + } + hookMu.Unlock() + + close(releaseFirst) + time.Sleep(50 * time.Millisecond) + + mu.Lock() + defer mu.Unlock() + if len(handled) != 1 || handled[0] != "first" { + t.Fatalf("handled = %v, want only [first] with the follow-up dropped", handled) + } + if firstInterrupted { + t.Fatal("declined preemption cancelled the in-flight handler") + } + if strings.Contains(logs.String(), "chat lock preempted") { + t.Fatalf("declined hook still force released the lock; logs:\n%s", logs.String()) + } +} + +func TestPreemptHookDecliningFallsBackToQueue(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + return false + }, func(o *chat.RuntimeOptions) { + o.Concurrency = chat.ConcurrencyQueue + o.ThreadLockTTL = 40 * time.Millisecond + }) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + if status := postEvent(t, bot, "fake", mentionEvent("queued", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("queued status = %d", status) + } + time.Sleep(30 * time.Millisecond) + close(releaseFirst) + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("declined preemption did not fall back to the queue strategy") + case <-time.After(5 * time.Millisecond): + } + } + mu.Lock() + defer mu.Unlock() + if !equalStrings(handled, []string{"first", "queued"}) { + t.Fatalf("handled = %v, want [first queued]", handled) + } +} + +func TestPreemptHookPanicIsRecoveredAndFallsBack(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + panic("hook exploded") + }) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + first := len(handled) == 0 + handled = append(handled, ev.Event.ID) + mu.Unlock() + if first { + close(firstStarted) + <-releaseFirst + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + // A panicking hook is recovered and treated as declining: the conflict + // drops and the runtime stays healthy. + if status := postEvent(t, bot, "fake", mentionEvent("panicked", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("panicked status = %d", status) + } + deadline := time.After(2 * time.Second) + for { + out := logs.String() + if strings.Contains(out, "chat lock conflict hook panicked") && strings.Contains(out, "chat lock conflict dropped") { + break + } + select { + case <-deadline: + t.Fatalf("panicking hook not recovered and dropped; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + + close(releaseFirst) + time.Sleep(50 * time.Millisecond) + + // The runtime must still dispatch after the panic. + if status := postEvent(t, bot, "fake", mentionEvent("after", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("after status = %d", status) + } + deadline = time.After(2 * time.Second) + for { + mu.Lock() + n := len(handled) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("runtime did not dispatch after a hook panic") + case <-time.After(5 * time.Millisecond): + } + } +} + +func TestPreemptHookNotConsultedForSelfEvents(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + var hookMu sync.Mutex + hookCalls := 0 + bot := newPreemptRuntime(t, state, adapter, &logs, func(ctx context.Context, ev *chat.Event) bool { + hookMu.Lock() + hookCalls++ + hookMu.Unlock() + return true + }) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + defer close(releaseFirst) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + close(firstStarted) + select { + case <-releaseFirst: + case <-ctx.Done(): + } + return nil + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + // A self message during the conflict resolves as ignored: it must never + // preempt the in-flight handler. + self := mentionEvent("self", "fake:v1:thread-1") + self.Message.Author = adapter.BotActor() + if status := postEvent(t, bot, "fake", self); status != http.StatusOK { + t.Fatalf("self status = %d", status) + } + + deadline := time.After(2 * time.Second) + for !strings.Contains(logs.String(), "chat ignored self message") { + select { + case <-deadline: + t.Fatalf("self message not ignored; logs:\n%s", logs.String()) + case <-time.After(5 * time.Millisecond): + } + } + hookMu.Lock() + defer hookMu.Unlock() + if hookCalls != 0 { + t.Fatalf("hook consulted %d times for a self event, want 0", hookCalls) + } + if strings.Contains(logs.String(), "chat lock preempted") { + t.Fatalf("self event force released the lock; logs:\n%s", logs.String()) + } +} + +func TestPreemptConstructionValidation(t *testing.T) { + t.Parallel() + + hook := func(ctx context.Context, ev *chat.Event) bool { return true } + newRuntime := func(state chat.State, mutate func(*chat.RuntimeOptions)) error { + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyDrop, + Dispatch: chat.DispatchDeferred, + DetachTimeout: time.Second, + OnLockConflict: hook, + } + mutate(&options) + _, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(newFakeAdapter("fake")), + chat.WithRuntimeOptions(options), + ) + return err + } + + if err := newRuntime(newFakeState(), func(o *chat.RuntimeOptions) {}); err != nil { + t.Fatalf("valid preemption options rejected: %v", err) + } + if err := newRuntime(newFakeState(), func(o *chat.RuntimeOptions) { + o.Dispatch = chat.DispatchSync + o.DetachTimeout = 0 + }); err == nil { + t.Fatal("expected the hook under sync dispatch to fail") + } + if err := newRuntime(newFakeState(), func(o *chat.RuntimeOptions) { + o.Concurrency = chat.ConcurrencyConcurrent + o.MaxConcurrent = 2 + }); err == nil { + t.Fatal("expected the hook under the concurrent strategy to fail") + } + // A State without the LockForcer capability cannot support preemption. + noForce := struct{ chat.State }{State: newFakeState()} + if err := newRuntime(noForce, func(o *chat.RuntimeOptions) {}); err == nil { + t.Fatal("expected the hook without a LockForcer state to fail") + } +} diff --git a/runtime.go b/runtime.go index 3821ad2..db8efe2 100644 --- a/runtime.go +++ b/runtime.go @@ -20,8 +20,61 @@ const ( // ConcurrencyQueue waits for the in-flight handler, then dispatches the most // recent superseded event for the scope. ConcurrencyQueue + // ConcurrencyDebounce coalesces rapid follow-ups: each new routed event for a + // scope supersedes the previous waiter and only the final event in a + // DebounceInterval quiet period dispatches. Superseded events are surfaced as + // skipped through Runtime Observation, never silently. Requires deferred + // dispatch (a synchronous webhook cannot park an event past the platform's + // acknowledgement deadline). + // + // Like queue supersession, coalescing is per runtime instance: events for + // one scope delivered to different instances sharing a State are not + // superseded across instances (each instance dispatches its own final + // event, serialized by the Thread Lock). Cross-instance coalescing needs + // the wait/coalesce State-contract extension anticipated by ADR 0012. + ConcurrencyDebounce + // ConcurrencyBurst collects routed events for a scope while a + // DebounceInterval window is open, then dispatches the whole batch — in the + // order the events joined the window — under a single Thread Lock hold. No + // event is skipped: every batch member runs under its own DetachTimeout + // execution budget (collection time and earlier members never consume it), + // with the batch as a whole bounded by Shutdown and by Lock Lease loss. + // Join order follows each event's admission through the dispatch prelude; + // ordering of concurrent webhook deliveries is platform-dependent and not + // re-established by the runtime. Windows are per runtime instance (like + // debounce coalescing), serialized across instances by the Thread Lock. + // Requires deferred dispatch, like ConcurrencyDebounce. + ConcurrencyBurst + // ConcurrencyConcurrent is the explicit opt-out of per-scope serialization: + // every routed event dispatches immediately in its own execution, bounded by + // MaxConcurrent. No Thread Lock is taken, so the caller accepts interleaved + // replies and races on Thread Application State. + ConcurrencyConcurrent ) +// LockScope selects what key the Thread Lock guards. The opaque Thread ID is +// unchanged; the scope only chooses the serialization key. +type LockScope int + +const ( + // LockScopeThread serializes handlers per Thread. This is the default. + LockScopeThread LockScope = iota + // LockScopeChannel widens serialization from a single Thread to its whole + // channel, for platforms whose model requires channel-wide ordering. A + // Thread whose adapter reports no channel falls back to per-Thread locking + // rather than sharing one adapter-wide key. + LockScopeChannel +) + +// LockConflictHook is the force/steerability hook consulted on a Lock Conflict +// for a routed, accepted event. Returning true preempts the in-flight handler: +// the runtime cancels the local Detached Work Context with ErrPreempted, force +// releases the current Lock Lease through the state's LockForcer capability, +// and dispatches the new event under a fresh lease. Returning false falls back +// to the configured Concurrency Strategy. The hook must be fast and must not +// block dispatch; a panicking hook is recovered and treated as false. +type LockConflictHook func(context.Context, *Event) bool + // DispatchMode selects whether the routed handler runs before or after the // adapter acknowledges the platform. type DispatchMode int @@ -41,6 +94,23 @@ type RuntimeOptions struct { Concurrency ConcurrencyStrategy Dispatch DispatchMode DetachTimeout time.Duration + // LockScope selects the Thread Lock key: per Thread (default) or per + // channel. + LockScope LockScope + // DebounceInterval is the quiet period (ConcurrencyDebounce) or the + // collection window (ConcurrencyBurst). It must be positive under those + // strategies and is ignored otherwise. + DebounceInterval time.Duration + // MaxConcurrent bounds simultaneous handler executions under + // ConcurrencyConcurrent. It must be positive under that strategy and is + // ignored otherwise. + MaxConcurrent int + // OnLockConflict, when set, lets a new delivery preempt an in-flight handler + // on a Lock Conflict instead of being dropped or queued. It requires + // deferred dispatch (the Detached Work Context is what makes the preempted + // handler cancellable), the drop or queue strategy, and a State that + // implements LockForcer. + OnLockConflict LockConflictHook } func DefaultRuntimeOptions() RuntimeOptions { @@ -50,6 +120,7 @@ func DefaultRuntimeOptions() RuntimeOptions { Concurrency: ConcurrencyDrop, Dispatch: DispatchSync, DetachTimeout: 0, + LockScope: LockScopeThread, } } @@ -109,9 +180,22 @@ type Chat struct { // inflight tracks detached tails so Shutdown drains them before state shutdown. inflight sync.WaitGroup - // queueMu guards pending, the per-scope most-recent queued event. + // queueMu guards pending, the per-scope most-recent pending waiter. queueMu sync.Mutex - pending map[string]*Event + pending map[string]*pendingWaiter + + // burstMu guards burstScopes, the per-scope burst collection windows. + burstMu sync.Mutex + burstScopes map[string]*burstScope + + // inflightMu guards inflightCancels, the per-scope local lease ownership + // reservations (acquiring or holding), used by preemption. + inflightMu sync.Mutex + inflightCancels map[string]map[*inflightCancel]struct{} + + // concurrencySlots bounds simultaneous handler executions under + // ConcurrencyConcurrent; nil under every other strategy. + concurrencySlots chan struct{} shutdownMu sync.Mutex shutdown bool @@ -150,6 +234,11 @@ func New(ctx context.Context, opts ...Option) (*Chat, error) { if err := validateRuntimeOptions(cfg.options); err != nil { return nil, err } + if cfg.options.OnLockConflict != nil { + if _, ok := cfg.state.(LockForcer); !ok { + return nil, errors.New("chat: on-lock-conflict hook requires a State implementing LockForcer") + } + } baseCtx, baseCancel := context.WithCancel(context.Background()) chat := &Chat{ @@ -161,7 +250,12 @@ func New(ctx context.Context, opts ...Option) (*Chat, error) { eventAcceptances: map[string]*eventAcceptance{}, baseCtx: baseCtx, baseCancel: baseCancel, - pending: map[string]*Event{}, + pending: map[string]*pendingWaiter{}, + burstScopes: map[string]*burstScope{}, + inflightCancels: map[string]map[*inflightCancel]struct{}{}, + } + if cfg.options.Concurrency == ConcurrencyConcurrent { + chat.concurrencySlots = make(chan struct{}, cfg.options.MaxConcurrent) } for _, adapter := range cfg.adapters { if adapter == nil { @@ -198,9 +292,41 @@ func validateRuntimeOptions(options RuntimeOptions) error { } switch options.Concurrency { case ConcurrencyDrop, ConcurrencyQueue: + case ConcurrencyDebounce, ConcurrencyBurst: + if options.DebounceInterval <= 0 { + return errors.New("chat: debounce interval must be positive under the debounce and burst strategies") + } + if options.Dispatch != DispatchDeferred { + return errors.New("chat: debounce and burst strategies require deferred dispatch") + } + // The interval wait runs inside the DetachTimeout-bounded Detached Work + // Context: a timeout at or below the interval would abandon every + // accepted event before its quiet period or collection window closes. + if options.DetachTimeout <= options.DebounceInterval { + return errors.New("chat: detach timeout must exceed the debounce interval under the debounce and burst strategies") + } + case ConcurrencyConcurrent: + if options.MaxConcurrent <= 0 { + return errors.New("chat: max concurrent must be positive under the concurrent strategy") + } default: return errors.New("chat: unsupported concurrency strategy") } + switch options.LockScope { + case LockScopeThread, LockScopeChannel: + default: + return errors.New("chat: unsupported lock scope") + } + if options.OnLockConflict != nil { + if options.Dispatch != DispatchDeferred { + return errors.New("chat: on-lock-conflict hook requires deferred dispatch") + } + switch options.Concurrency { + case ConcurrencyDrop, ConcurrencyQueue: + default: + return errors.New("chat: on-lock-conflict hook requires the drop or queue strategy") + } + } switch options.Dispatch { case DispatchSync: case DispatchDeferred: @@ -354,21 +480,32 @@ func (c *Chat) dispatch(ctx context.Context, event *Event) error { // dispatchSync runs the prelude and the routed handler inline under the request // context, releasing the Thread Lock when the handler returns. Under the queue // strategy a Lock Conflict waits inline (bounded by ctx) for the in-flight -// handler before the routed handler runs. +// handler before the routed handler runs. Under the concurrent strategy the +// request waits inline for a MaxConcurrent slot instead of a lock. func (c *Chat) dispatchSync(ctx context.Context, event *Event) error { work, resolved, err := c.prelude(ctx, event) if err != nil || resolved { return err } - if work.needsLock { - lease, outcome := c.queueForLock(ctx, work.scope, event) - if outcome != acquireHeld { + if work.noLock { + if !c.acquireConcurrencySlot(ctx, event) { c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) return nil } - work.lease = lease + defer c.releaseConcurrencySlot() + } else { + if work.needsLock { + // The steerability hook requires deferred dispatch, so a sync wait + // never carries an ownership reservation. + lease, _, outcome := c.queueForLock(ctx, work.scope, event, work.waitLabel) + if outcome != acquireHeld { + c.safeEnd(work.span, waitOutcome(outcome), RouteAttr(work.route)) + return nil + } + work.lease = lease + } + defer c.releaseLock(ctx, work.lease, event.ThreadID) } - defer c.releaseLock(ctx, work.lease, event.ThreadID) if err := work.run(ctx); err != nil { c.logger.Error("chat handler failed", "error", err, "adapter", event.Adapter, "event_id", event.ID, "route", work.route) c.safeEvent(ctx, ObsHandlerError, AdapterAttr(event.Adapter), RouteAttr(work.route)) @@ -402,10 +539,34 @@ type preludeWork struct { run func(context.Context) error route string scope string - // needsLock is true when the prelude registered the event as pending under the - // queue strategy without holding the lock; the tail must wait for and acquire - // it before running the handler. + // needsLock is true when the prelude did not acquire the Thread Lock (a + // queued/preempting Lock Conflict, or the debounce strategy); the tail must + // wait for and acquire it before running the handler. needsLock bool + // waitLabel names the coordination mode ("queue", "debounce", or "preempt") + // in wait observations so a log line names the strategy that produced it. + waitLabel string + // displaced closes when a newer event supersedes this pending waiter; the + // debounce quiet-period wait exits promptly on it instead of parking through + // its full interval. + displaced <-chan struct{} + // debounce is true when the tail must hold the event through the + // DebounceInterval quiet period before waiting for the lock. + debounce bool + // preempt is true when the OnLockConflict hook elected to preempt the + // in-flight handler: the tail cancels it locally and force releases the + // current Lock Lease before waiting for a fresh one. + preempt bool + // noLock is true under the concurrent strategy: no Thread Lock is taken and + // the run is bounded by a MaxConcurrent slot instead. + noLock bool + // inflight is the scope ownership reservation made when the lease was + // acquired (preemptible runtimes only); the tail arms it with the running + // handler's cancel and retires it after release. + inflight *inflightCancel + // burstRunner marks a scope-level burst runner tail rather than an + // event-level dispatch; only event and scope are set. + burstRunner bool // span is opened in the prelude and closed by the tail, so deferred dispatch // measures Ack-Then-Work latency to handler completion. span DispatchSpan @@ -465,40 +626,80 @@ func (c *Chat) prelude(ctx context.Context, event *Event) (preludeWork, bool, er return preludeWork{}, true, err } - scope := string(event.ThreadID) - lease, acquired, err := c.state.AcquireLock(ctx, scope, c.options.ThreadLockTTL) - if err != nil { - err := finish(fmt.Errorf("chat: acquire thread lock: %w", err)) + // The Thread ID is validated before the Thread Lock so the lock scope key + // (which may be channel-wide) comes from the adapter-validated ThreadRef; an + // event with an invalid Thread ID never touches the lock. + ref, validateErr := adapter.ValidateThreadID(event.ThreadID) + if validateErr != nil { + err := finish(fmt.Errorf("chat: validate event thread id: %w", validateErr)) c.safeEnd(span, OutcomeError) return preludeWork{}, true, err } - queued := false - if !acquired { - if c.options.Concurrency != ConcurrencyQueue { - accepted, err := acceptEvent() - if err != nil { - c.safeEnd(span, OutcomeError) - return preludeWork{}, true, err + scope := c.lockScopeKey(event, ref) + + // The prelude acquires the Thread Lock only under drop/queue; debounce and + // burst always coordinate in the tail, and concurrent takes no lock at all. + var lease LockLease + var reservation *inflightCancel + conflicted := false + switch c.options.Concurrency { + case ConcurrencyDrop, ConcurrencyQueue: + if c.options.OnLockConflict != nil { + // Local ownership is reserved BEFORE the acquire so a lease can never + // exist locally without a visible reservation: a preemptor arriving at + // any point after acquisition finds the local owner and waits instead + // of force-releasing the fresh lease. + reservation = c.reserveInflight(scope) + } + acquiredLease, acquired, err := c.state.AcquireLock(ctx, scope, c.options.ThreadLockTTL) + if err != nil { + if reservation != nil { + c.retireInflight(scope, reservation) } - if !accepted { - c.safeEnd(span, OutcomeDuplicate) + err := finish(fmt.Errorf("chat: acquire thread lock: %w", err)) + c.safeEnd(span, OutcomeError) + return preludeWork{}, true, err + } + lease = acquiredLease + if acquired && reservation != nil { + reservation.markHeld() + } + if !acquired { + if reservation != nil { + c.retireInflight(scope, reservation) + reservation = nil + } + if c.options.Concurrency == ConcurrencyDrop && c.options.OnLockConflict == nil { + accepted, err := acceptEvent() + if err != nil { + c.safeEnd(span, OutcomeError) + return preludeWork{}, true, err + } + if !accepted { + c.safeEnd(span, OutcomeDuplicate) + return preludeWork{}, true, nil + } + c.logger.Info("chat lock conflict dropped", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) + c.safeEvent(ctx, ObsLockConflict, AdapterAttr(event.Adapter)) + c.safeEnd(span, OutcomeDroppedLockConflict) return preludeWork{}, true, nil } - c.logger.Info("chat lock conflict dropped", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) - c.safeEvent(ctx, ObsLockConflict, AdapterAttr(event.Adapter)) - c.safeEnd(span, OutcomeDroppedLockConflict) - return preludeWork{}, true, nil + conflicted = true } - queued = true + case ConcurrencyDebounce, ConcurrencyBurst, ConcurrencyConcurrent: } - // releaseOnResolve releases the lease when the event resolves here; a queued - // event holds no lease yet. + // releaseOnResolve releases the lease (and retires its ownership + // reservation) when the event resolves here; an event whose strategy defers + // lock coordination to the tail holds no lease yet. releaseOnResolve := func() { - if queued { + if lease == (LockLease{}) { return } c.releaseLock(ctx, lease, event.ThreadID) + if reservation != nil { + c.retireInflight(scope, reservation) + } } // resolveError releases the lease and closes the span for a failed prelude. resolveError := func(err error) (preludeWork, bool, error) { @@ -525,10 +726,6 @@ func (c *Chat) prelude(ctx context.Context, event *Event) (preludeWork, bool, er return preludeWork{}, true, nil } - ref, err := adapter.ValidateThreadID(event.ThreadID) - if err != nil { - return resolveError(fmt.Errorf("chat: validate event thread id: %w", err)) - } thread := c.newThread(adapter, ref) bot := adapter.BotActor() @@ -547,7 +744,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event) (preludeWork, bool, er return resolveIgnored("no-command-handler", "chat ignored command with no handler", slog.LevelInfo) } cmdEvent := &CommandEvent{Event: event, Thread: thread, Command: event.Command} - return c.routedWork(span, event, lease, scope, queued, "command", func(ctx context.Context) error { + return c.routedWork(ctx, span, event, lease, reservation, scope, conflicted, "command", func(ctx context.Context) error { return handler(ctx, cmdEvent) }, acceptEvent, releaseOnResolve) @@ -562,7 +759,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event) (preludeWork, bool, er return resolveIgnored("no-interaction-handler", "chat ignored interaction with no handler", slog.LevelInfo) } intEvent := &InteractionEvent{Event: event, Thread: thread, Interaction: event.Interaction} - return c.routedWork(span, event, lease, scope, queued, "interaction", func(ctx context.Context) error { + return c.routedWork(ctx, span, event, lease, reservation, scope, conflicted, "interaction", func(ctx context.Context) error { return handler(ctx, intEvent) }, acceptEvent, releaseOnResolve) } @@ -583,20 +780,23 @@ func (c *Chat) prelude(ctx context.Context, event *Event) (preludeWork, bool, er } msgEvent := &MessageEvent{Event: event, Thread: thread, Message: event.Message} - return c.routedWork(span, event, lease, scope, queued, route, func(ctx context.Context) error { + return c.routedWork(ctx, span, event, lease, reservation, scope, conflicted, route, func(ctx context.Context) error { return handler(ctx, msgEvent) }, acceptEvent, releaseOnResolve) } -// routedWork accepts a routed event (deduping), registers it as a queue waiter -// when it is waiting for the lock, and builds the handler-agnostic preludeWork. A -// duplicate or a dedupe error resolves here instead. +// routedWork accepts a routed event (deduping), applies the Concurrency +// Strategy's coordination decision (queue/preempt/debounce/burst/concurrent), +// and builds the handler-agnostic preludeWork. A duplicate, a dedupe error, or +// a dropped Lock Conflict resolves here instead. func (c *Chat) routedWork( + ctx context.Context, span DispatchSpan, event *Event, lease LockLease, + reservation *inflightCancel, scope string, - queued bool, + conflicted bool, route string, run func(context.Context) error, acceptEvent func() (bool, error), @@ -614,27 +814,89 @@ func (c *Chat) routedWork( return preludeWork{}, true, nil } - if queued { - c.registerPending(scope, event) + work := preludeWork{ + event: event, + lease: lease, + run: run, + route: route, + scope: scope, + span: span, + inflight: reservation, + } + + switch c.options.Concurrency { + case ConcurrencyDrop, ConcurrencyQueue: + if !conflicted { + return work, false, nil + } + // A Lock Conflict on a routed, accepted event: the steerability hook may + // preempt the in-flight handler; otherwise the strategy decides. + work.needsLock = true + if c.options.OnLockConflict != nil && c.safeLockConflictHook(ctx, event) { + work.preempt = true + work.waitLabel = "preempt" + work.displaced = c.registerPending(scope, event, work.waitLabel) + return work, false, nil + } + if c.options.Concurrency == ConcurrencyQueue { + work.waitLabel = "queue" + work.displaced = c.registerPending(scope, event, work.waitLabel) + return work, false, nil + } + // Drop fallback: the hook declined, so the conflict is acknowledged and + // dropped exactly like the plain drop strategy. + c.logger.Info("chat lock conflict dropped", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) + c.safeEvent(ctx, ObsLockConflict, AdapterAttr(event.Adapter)) + c.safeEnd(span, OutcomeDroppedLockConflict) + return preludeWork{}, true, nil + + case ConcurrencyDebounce: + work.needsLock = true + work.debounce = true + work.waitLabel = "debounce" + work.displaced = c.registerPending(scope, event, work.waitLabel) + return work, false, nil + + case ConcurrencyBurst: + // The event joins its scope's collection window; the runner tail (started + // by at most one joining event per window chain) owns every joined span. + if c.joinBurstBatch(scope, work) { + return preludeWork{event: event, scope: scope, burstRunner: true}, false, nil + } + return preludeWork{}, true, nil + + case ConcurrencyConcurrent: + work.noLock = true + return work, false, nil } - return preludeWork{ - event: event, - lease: lease, - run: run, - route: route, - scope: scope, - needsLock: queued, - span: span, - }, false, nil + // Unreachable: the strategy set is validated at construction. + releaseOnResolve() + c.safeEnd(span, OutcomeError) + return preludeWork{}, true, fmt.Errorf("chat: unsupported concurrency strategy %d", c.options.Concurrency) +} + +// safeLockConflictHook invokes the steerability hook best-effort: a panicking +// hook is recovered and treated as "follow the configured strategy". +func (c *Chat) safeLockConflictHook(ctx context.Context, event *Event) (preempt bool) { + defer func() { + if r := recover(); r != nil { + c.logger.Warn("chat lock conflict hook panicked", "recovered", r, "adapter", event.Adapter, "event_id", event.ID) + preempt = false + } + }() + return c.options.OnLockConflict(ctx, event) } // startDetachedTail runs the routed handler on the detached work context after // ack: the Thread Lock is held across the tail, refreshed via ExtendLock, and // released on exit. The context is derived from baseCtx, bounded by // DetachTimeout, and cancelled by Shutdown. When needsLock, the tail first waits -// for the lock under the queue strategy; a superseded or abandoned waiter exits -// without running the handler. +// for the lock (after the debounce quiet period and/or preemption when the work +// asks for them); a superseded or abandoned waiter exits without running the +// handler. Burst runner tails coordinate their scope's batch instead of a +// single event, and concurrent tails wait for a MaxConcurrent slot rather than +// a lock. func (c *Chat) startDetachedTail(work preludeWork) { tailCtx, tailCancel := context.WithTimeout(c.baseCtx, c.options.DetachTimeout) c.inflight.Add(1) @@ -642,47 +904,398 @@ func (c *Chat) startDetachedTail(work preludeWork) { defer c.inflight.Done() defer tailCancel() + if work.burstRunner { + c.runBurstScope(tailCtx, work.scope) + return + } + + if work.noLock { + if !c.acquireConcurrencySlot(tailCtx, work.event) { + c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) + return + } + defer c.releaseConcurrencySlot() + c.logger.Info("chat deferred dispatch started", "adapter", work.event.Adapter, "event_id", work.event.ID, "route", work.route) + err := work.run(tailCtx) + c.endHandlerRun(tailCtx, work.event, work.route, work.span, err) + return + } + + if work.debounce && !c.waitDebounceQuietPeriod(tailCtx, work) { + return + } + + if work.preempt { + c.preemptScope(tailCtx, work) + } + if work.needsLock { - lease, outcome := c.queueForLock(tailCtx, work.scope, work.event) + lease, reservation, outcome := c.queueForLock(tailCtx, work.scope, work.event, work.waitLabel) if outcome != acquireHeld { - // A superseded or abandoned waiter never runs the handler. - c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) + // A superseded or abandoned waiter never runs the handler; a state + // failure is an error outcome, not a silent skip. + c.safeEnd(work.span, waitOutcome(outcome), RouteAttr(work.route)) return } work.lease = lease + work.inflight = reservation } c.logger.Info("chat deferred dispatch started", "adapter", work.event.Adapter, "event_id", work.event.ID, "route", work.route) + c.runLockedTail(tailCtx, work) + }() +} - stopRefresh, leaseLost := c.startLockRefresh(tailCtx, work.lease, work.event.ThreadID) - err := work.run(tailCtx) - stopRefresh() +// waitOutcome maps a non-held lock-wait result to its terminal DispatchOutcome: +// a backend failure surfaces as an error; supersession and abandonment close as +// ignored. +func waitOutcome(outcome acquireOutcome) DispatchOutcome { + if outcome == acquireFailed { + return OutcomeError + } + return OutcomeIgnored +} - if err != nil { - if ctxErr := tailCtx.Err(); ctxErr != nil { - c.logger.Error("chat deferred handler timed out", "error", err, "adapter", work.event.Adapter, "event_id", work.event.ID, "route", work.route) - } else { - c.logger.Error("chat handler failed", "error", err, "adapter", work.event.Adapter, "event_id", work.event.ID, "route", work.route, "mode", "deferred") - } - c.safeEvent(tailCtx, ObsHandlerError, AdapterAttr(work.event.Adapter), RouteAttr(work.route)) - c.safeEnd(work.span, OutcomeError, RouteAttr(work.route)) +// runLockedTail runs a lock-holding deferred handler: the Lock Lease refresh +// loop runs alongside, the handler's cancel is armed on the scope's ownership +// reservation when the steerability hook is configured, and the lease is +// released on exit. The reservation is retired only after the lease release so +// a waiting preemptor observes a scope whose lock is genuinely free. +func (c *Chat) runLockedTail(tailCtx context.Context, work preludeWork) { + // Every deferred lock holder is cancellable on lease loss (with cause + // ErrPreempted), regardless of the local hook configuration: a lease force + // released by another runtime instance — or expired — means mutual + // exclusion is already gone, so the handler must stop rather than run + // alongside the lease's next holder. + runCtx, cancel := context.WithCancelCause(tailCtx) + defer cancel(nil) + if c.options.OnLockConflict != nil { + assert(work.inflight != nil, "preemptible lock-holding work must carry an ownership reservation") + if !work.inflight.arm(cancel) { + // Preempted between acquisition and start: the handler never runs. + c.logger.Info("chat handler preempted", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "route", work.route, "started", false) + c.safeEnd(work.span, OutcomePreempted, RouteAttr(work.route)) + c.releaseTailLock(tailCtx, work.lease, work.event.ThreadID, true) + c.retireInflight(work.scope, work.inflight) + return + } + } + + stopRefresh, leaseLost := c.startLockRefresh(tailCtx, work.lease, work.event.ThreadID, cancel) + err := work.run(runCtx) + stopRefresh() + + // The preemption outcome follows the cancellation cause, not the handler's + // return convention: a handler that observes ctx.Done, shuts down cleanly, + // and returns nil was still preempted (or lost its lease). + preempted := errors.Is(context.Cause(runCtx), ErrPreempted) + if preempted && tailCtx.Err() == nil { + c.logger.Info("chat handler preempted", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "route", work.route, "started", true) + c.safeEnd(work.span, OutcomePreempted, RouteAttr(work.route)) + } else { + c.endHandlerRun(tailCtx, work.event, work.route, work.span, err) + } + + // A failed release is expected (lease gone) when the context was cancelled, + // the refresh loop already lost the lease, or the lease was force released + // by a preempting delivery, so downgrade it from WARN. + benignRelease := tailCtx.Err() != nil || leaseLost() || preempted + c.releaseTailLock(tailCtx, work.lease, work.event.ThreadID, benignRelease) + if work.inflight != nil { + c.retireInflight(work.scope, work.inflight) + } +} + +// endHandlerRun closes one deferred handler run with the shared terminal +// logging, observation, and span outcome. +func (c *Chat) endHandlerRun(tailCtx context.Context, event *Event, route string, span DispatchSpan, err error) { + if err != nil { + if ctxErr := tailCtx.Err(); ctxErr != nil { + c.logger.Error("chat deferred handler timed out", "error", err, "adapter", event.Adapter, "event_id", event.ID, "route", route) } else { - c.safeEnd(work.span, OutcomeHandled, RouteAttr(work.route)) + c.logger.Error("chat handler failed", "error", err, "adapter", event.Adapter, "event_id", event.ID, "route", route, "mode", "deferred") } + c.safeEvent(tailCtx, ObsHandlerError, AdapterAttr(event.Adapter), RouteAttr(route)) + c.safeEnd(span, OutcomeError, RouteAttr(route)) + return + } + c.safeEnd(span, OutcomeHandled, RouteAttr(route)) +} - // A failed release is expected (lease gone) when the context was cancelled - // or the refresh loop already lost the lease, so downgrade it from WARN. - benignRelease := tailCtx.Err() != nil || leaseLost() - c.releaseTailLock(tailCtx, work.lease, work.event.ThreadID, benignRelease) - }() +// waitDebounceQuietPeriod holds the routed event through its DebounceInterval +// quiet period. A superseded waiter exits promptly on its displacement signal +// (releasing its goroutine and event immediately instead of parking through the +// full interval), and an abandoned wait exits when the Detached Work Context +// ends. It reports false when the handler must not run (the span is closed +// here). +func (c *Chat) waitDebounceQuietPeriod(ctx context.Context, work preludeWork) bool { + timer := time.NewTimer(c.options.DebounceInterval) + defer timer.Stop() + select { + case <-ctx.Done(): + c.clearPending(work.scope, work.event) + c.logger.Info("chat debounce wait abandoned", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "error", ctx.Err()) + c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) + return false + case <-work.displaced: + // registerPending already emitted the superseded_by record. + c.logger.Debug("chat debounce waiter superseded", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID) + c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) + return false + case <-timer.C: + return true + } +} + +// preemptScope preempts the scope's in-flight handler on behalf of a new +// delivery. A local victim is cancelled with ErrPreempted and awaited: it +// releases its own lease, so the preemptor then acquires normally and the +// force capability is never invoked — a lease acquired by a fresh handler +// after the victim's release is never destroyed. Only when no local victim is +// registered (the conflicting lease belongs to another runtime instance or is +// orphaned) is the lease force released through LockForcer so this delivery +// can acquire a fresh one. The token-owned lease invariant is preserved: the +// lease is explicitly invalidated, never deleted by presenting another +// holder's token. +// +// Waiting (bounded by the Detached Work Context) for the local victim keeps +// drop/queue per-scope serialization intact for handlers on this instance. A +// remote holder can only be fenced by the lease: it observes the force release +// on its next refresh, which cancels its handler, so cross-instance overlap is +// bounded by one refresh interval. The same fencing bounds the instruction- +// scale window in which a fresh holder could acquire between the pending +// re-check and the force call. +func (c *Chat) preemptScope(ctx context.Context, work preludeWork) { + // A waiter superseded between routing and this tail must not destroy the + // in-flight handler on behalf of an event that will never dispatch; the + // newest waiter performs its own preemption if its hook elected one. + for { + // A waiter superseded (or abandoned) at any point yields silently: the + // newest waiter owns the preemption. + if ctx.Err() != nil { + return + } + // Validation and destructive cancellation are atomic under the pending + // registry lock: a stale preemptor (displaced by a newer registration) + // can never cancel the active holder — or poison a newer waiter's + // reservation — on behalf of an event that will not dispatch. + victims, stillPending := c.preemptLocalIfPending(work.scope, work.event) + if !stillPending { + return + } + if len(victims) == 0 { + break + } + anyHeld := false + for _, victim := range victims { + select { + case <-victim.done: + case <-ctx.Done(): + // The abandoned preemptor exits through the lock wait; the victim + // keeps its lease. + return + } + if victim.wasHeld() { + anyHeld = true + } + } + if anyHeld { + // A local owner was preempted and has released; this delivery + // acquires through the normal lock wait — never by force. + if !c.isPending(work.scope, work.event) { + return + } + c.logger.Info("chat lock preempted", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "forced", false) + c.safeEvent(ctx, ObsLockPreempted, AdapterAttr(work.event.Adapter)) + return + } + // Only non-holding acquirers came and went; look again for a holder + // before concluding the conflicting lease is remote. + } + // No local reservation: re-validate ownership, then force-invalidate the + // remote or orphaned lease. + if !c.isPending(work.scope, work.event) { + return + } + forcer, ok := c.state.(LockForcer) + assert(ok, "preempt requires a LockForcer state (validated at construction)") + // The force release stays bounded by the Detached Work Context so a stalled + // state backend cannot park this tail past DetachTimeout. + released, err := forcer.ForceReleaseLock(ctx, work.scope) + if err != nil { + // The lock wait that follows still polls; a stale lease expires at TTL. + c.logger.Error("chat force release thread lock failed", "error", err, "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID) + return + } + c.logger.Info("chat lock preempted", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "forced", true, "released", released) + c.safeEvent(ctx, ObsLockPreempted, AdapterAttr(work.event.Adapter)) +} + +// inflightCancel is one local lease ownership reservation for a scope, +// registered BEFORE AcquireLock is attempted (preemptible runtimes only) so a +// lease can never exist locally without a visible reservation: a preemptor can +// never mistake a freshly acquired (or still-acquiring) local lease for a +// remote one. held records whether the reservation's acquisition succeeded. +// done closes once the reservation is retired (its lease, if any, released), +// so a preemptor can wait for local completion. +type inflightCancel struct { + mu sync.Mutex + // cancel is nil until the handler starts running (arm); a reservation + // preempted before arming prevents the handler from starting at all. + cancel context.CancelCauseFunc + preempted bool + held bool + done chan struct{} +} + +// markHeld records that the reservation's AcquireLock succeeded. +func (e *inflightCancel) markHeld() { + e.mu.Lock() + e.held = true + e.mu.Unlock() +} + +// wasHeld reports whether the reservation ever held the lease; stable once +// done has closed. +func (e *inflightCancel) wasHeld() bool { + e.mu.Lock() + defer e.mu.Unlock() + return e.held +} + +// preempt marks the reservation preempted, cancelling the running handler when +// one is armed. +func (e *inflightCancel) preempt() { + e.mu.Lock() + e.preempted = true + cancel := e.cancel + e.mu.Unlock() + if cancel != nil { + cancel(ErrPreempted) + } +} + +// arm installs the running handler's cancel. It reports false when the +// reservation was preempted before the handler started; the handler must not +// run. +func (e *inflightCancel) arm(cancel context.CancelCauseFunc) bool { + e.mu.Lock() + defer e.mu.Unlock() + if e.preempted { + return false + } + e.cancel = cancel + return true +} + +// reserveInflight registers a local ownership reservation for the scope. It +// must be called BEFORE every AcquireLock attempt on a preemptible runtime +// (so acquisition and registration cannot be reordered by scheduling), marked +// held on success, and retired (retireInflight) once the reservation ends — +// after the lease release when one was held. +func (c *Chat) reserveInflight(scope string) *inflightCancel { + entry := &inflightCancel{done: make(chan struct{})} + c.inflightMu.Lock() + set := c.inflightCancels[scope] + if set == nil { + set = map[*inflightCancel]struct{}{} + c.inflightCancels[scope] = set + } + set[entry] = struct{}{} + c.inflightMu.Unlock() + return entry +} + +// retireInflight removes the reservation and signals waiting preemptors. Call +// it only after the reserved lease (if any) has been released. +func (c *Chat) retireInflight(scope string, entry *inflightCancel) { + c.inflightMu.Lock() + if set := c.inflightCancels[scope]; set != nil { + delete(set, entry) + if len(set) == 0 { + delete(c.inflightCancels, scope) + } + } + c.inflightMu.Unlock() + close(entry.done) +} + +// preemptLocalIfPending atomically re-validates that event still owns the +// scope's pending slot and, only then, preempts every current local +// reservation (holders and in-flight acquirers alike), returning the preempted +// entries so the preemptor can wait for their completion. The pending registry +// lock is held across the marking so a newer registration cannot interleave +// between validation and cancellation. Lock order is queueMu > inflightMu > +// entry.mu, nested nowhere else in reverse. +func (c *Chat) preemptLocalIfPending(scope string, event *Event) (victims []*inflightCancel, stillPending bool) { + c.queueMu.Lock() + defer c.queueMu.Unlock() + waiter := c.pending[scope] + if waiter == nil || waiter.event != event { + return nil, false + } + c.inflightMu.Lock() + set := c.inflightCancels[scope] + victims = make([]*inflightCancel, 0, len(set)) + for entry := range set { + victims = append(victims, entry) + } + c.inflightMu.Unlock() + for _, entry := range victims { + entry.preempt() + } + return victims, true +} + +// acquireConcurrencySlot blocks until a MaxConcurrent slot frees, bounded by +// ctx. It reports false when the wait was abandoned. +func (c *Chat) acquireConcurrencySlot(ctx context.Context, event *Event) bool { + assert(c.concurrencySlots != nil, "concurrency slots are only used under the concurrent strategy") + select { + case c.concurrencySlots <- struct{}{}: + return true + case <-ctx.Done(): + c.logger.Info("chat concurrent slot wait abandoned", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID, "error", ctx.Err()) + return false + } +} + +func (c *Chat) releaseConcurrencySlot() { + <-c.concurrencySlots +} + +// lockScopeKey chooses what key the Thread Lock guards. Thread scope (the +// default) keys by the opaque Thread ID, unchanged from prior behavior. Channel +// scope widens serialization to the Thread's channel; a Thread whose adapter +// reports no channel falls back to per-Thread locking rather than sharing one +// adapter-wide key. Under channel scope every key carries a namespace prefix +// and length-prefixed fields, so channel keys are injective and a fallback +// Thread ID can never collide with a synthesized channel key. +func (c *Chat) lockScopeKey(event *Event, ref ThreadRef) string { + if c.options.LockScope != LockScopeChannel { + return string(event.ThreadID) + } + if ref.Channel == "" { + return fmt.Sprintf("thread-scope/%d:%s", len(event.ThreadID), event.ThreadID) + } + return fmt.Sprintf("channel-scope/%d:%s/%d:%s/%d:%s", + len(event.Adapter), event.Adapter, + len(ref.Tenant), ref.Tenant, + len(ref.Channel), ref.Channel, + ) } // startLockRefresh extends the Lock Lease on a cadence below ThreadLockTTL so it // does not expire while the detached handler runs. Extend runs under // context.WithoutCancel so a refresh in flight at cancellation still completes. // stop halts the loop and blocks until it exits; leaseLost reports whether the -// loop saw the lease already gone, and is safe to read once stop has returned. -func (c *Chat) startLockRefresh(ctx context.Context, lease LockLease, threadID ThreadID) (stop func(), leaseLost func() bool) { +// loop could no longer vouch for the lease, and is safe to read once stop has +// returned. A lost lease — vanished (force released by a preempting delivery +// on any runtime instance, or expired) or unmaintainable (a failed refresh +// means it expires at TTL mid-handler) — also cancels the handler with +// ErrPreempted (onLeaseLost): mutual exclusion is gone or going, so the +// handler must stop rather than run alongside the lease's next holder. +func (c *Chat) startLockRefresh(ctx context.Context, lease LockLease, threadID ThreadID, onLeaseLost context.CancelCauseFunc) (stop func(), leaseLost func() bool) { interval := c.options.ThreadLockTTL / 2 if interval <= 0 { interval = c.options.ThreadLockTTL @@ -703,12 +1316,22 @@ func (c *Chat) startLockRefresh(ctx context.Context, lease LockLease, threadID T case <-ticker.C: extended, err := c.state.ExtendLock(context.WithoutCancel(ctx), lease, c.options.ThreadLockTTL) if err != nil { + // A failed refresh means the lease can no longer be guaranteed: + // it will expire at TTL while the handler is still running, so + // the handler must stop rather than outlive its serialization. c.logger.Error("chat extend thread lock failed", "error", err, "thread_id", threadID) + lost.Store(true) + if onLeaseLost != nil { + onLeaseLost(ErrPreempted) + } return } if !extended { c.logger.Warn("chat thread lock lease lost", "thread_id", threadID) lost.Store(true) + if onLeaseLost != nil { + onLeaseLost(ErrPreempted) + } return } c.logger.Debug("chat thread lock refreshed", "thread_id", threadID) @@ -738,68 +1361,292 @@ const ( acquireFailed ) +// pendingWaiter is one scope's most-recent pending event plus a displacement +// signal: displaced closes when a newer event supersedes this waiter, so a +// parked (debounce) waiter exits promptly instead of holding its event through +// the full interval. +type pendingWaiter struct { + event *Event + displaced chan struct{} +} + // registerPending records event as the most-recent pending event for its scope, -// superseding any earlier waiter. Recording supersession here, where the -// displacing event's id is known, lets the record carry superseded_by. -func (c *Chat) registerPending(scope string, event *Event) { +// superseding (and signalling) any earlier waiter. Recording supersession here, +// where the displacing event's id is known, lets the record carry +// superseded_by. label names the coordination mode ("queue", "debounce", or +// "preempt") so the observation names the strategy that displaced the waiter. +// The returned channel closes when this registration is itself superseded. +func (c *Chat) registerPending(scope string, event *Event, label string) <-chan struct{} { c.queueMu.Lock() defer c.queueMu.Unlock() if prev := c.pending[scope]; prev != nil { - c.logger.Info("chat queue superseded", "adapter", prev.Adapter, "event_id", prev.ID, "thread_id", prev.ThreadID, "superseded_by", event.ID) + c.logger.Info("chat "+label+" superseded", "adapter", prev.event.Adapter, "event_id", prev.event.ID, "thread_id", prev.event.ThreadID, "superseded_by", event.ID) + close(prev.displaced) } - c.pending[scope] = event + waiter := &pendingWaiter{event: event, displaced: make(chan struct{})} + c.pending[scope] = waiter + return waiter.displaced +} + +// isPending reports whether event is still its scope's most-recent pending +// waiter. +func (c *Chat) isPending(scope string, event *Event) bool { + c.queueMu.Lock() + defer c.queueMu.Unlock() + waiter := c.pending[scope] + return waiter != nil && waiter.event == event } // queueForLock waits for the Thread Lock on behalf of an event already // registered as a pending waiter for its scope, polling AcquireLock until the // in-flight handler releases. It returns acquireSuperseded if a newer follow-up // displaced this waiter, acquireAbandoned if ctx ended, acquireFailed on an -// AcquireLock error, or the lease (acquireHeld) once acquired. -func (c *Chat) queueForLock(ctx context.Context, scope string, event *Event) (LockLease, acquireOutcome) { - ticker := time.NewTicker(c.queuePollInterval()) - defer ticker.Stop() - abandon := func() (LockLease, acquireOutcome) { +// AcquireLock error, or the lease (acquireHeld) once acquired. label names the +// coordination mode in the wait observations. +func (c *Chat) queueForLock(ctx context.Context, scope string, event *Event, label string) (LockLease, *inflightCancel, acquireOutcome) { + // The wait itself is reserved (preemptible runtimes) BEFORE any acquire + // attempt, so a lease can never exist locally without a visible + // reservation. A reservation preempted while still waiting is displaced by + // the preemptor's pending registration and exits superseded. + var reservation *inflightCancel + if c.options.OnLockConflict != nil { + reservation = c.reserveInflight(scope) + } + retire := func() { + if reservation != nil { + c.retireInflight(scope, reservation) + reservation = nil + } + } + superseded := func() bool { + return !c.isPending(scope, event) + } + lease, outcome, err := c.pollForLock(ctx, scope, superseded) + switch outcome { + case acquireHeld: + if reservation != nil { + reservation.markHeld() + } + // Re-check ownership after the acquire: a newer event registering while + // AcquireLock was in flight must win, or a superseded debounce/queue + // waiter would dispatch alongside the final one. Once the holder passes + // this check it is committed; later arrivals wait on the lock. + if !c.isPending(scope, event) { + c.releaseLock(ctx, lease, event.ThreadID) + retire() + c.logger.Debug("chat "+label+" waiter superseded", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) + return LockLease{}, nil, acquireSuperseded + } + c.clearPending(scope, event) + case acquireSuperseded: + // registerPending already emitted the superseded_by record; this waiter + // just stops. + retire() + c.logger.Debug("chat "+label+" waiter superseded", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) + case acquireAbandoned: + retire() c.clearPending(scope, event) - c.logger.Info("chat queue wait abandoned", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID, "error", ctx.Err()) - return LockLease{}, acquireAbandoned + c.logger.Info("chat "+label+" wait abandoned", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID, "error", ctx.Err()) + case acquireFailed: + retire() + c.clearPending(scope, event) + c.logger.Error("chat "+label+" acquire lock failed", "error", err, "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) } + return lease, reservation, outcome +} + +// pollForLock polls AcquireLock until it is held, the optional superseded check +// trips, ctx ends, or the state fails. Waiters poll because State does not +// signal release. +func (c *Chat) pollForLock(ctx context.Context, scope string, superseded func() bool) (LockLease, acquireOutcome, error) { + ticker := time.NewTicker(c.queuePollInterval()) + defer ticker.Stop() for { - c.queueMu.Lock() - superseded := c.pending[scope] != event - c.queueMu.Unlock() - if superseded { - // registerPending already emitted the superseded_by record; this waiter - // just stops. - c.logger.Debug("chat queue waiter superseded", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) - return LockLease{}, acquireSuperseded + if superseded != nil && superseded() { + return LockLease{}, acquireSuperseded, nil } lease, acquired, err := c.state.AcquireLock(ctx, scope, c.options.ThreadLockTTL) if err != nil { if ctx.Err() != nil { - return abandon() + return LockLease{}, acquireAbandoned, ctx.Err() } - c.clearPending(scope, event) - c.logger.Error("chat queue acquire lock failed", "error", err, "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) - return LockLease{}, acquireFailed + return LockLease{}, acquireFailed, err } if acquired { - c.clearPending(scope, event) - return lease, acquireHeld + return lease, acquireHeld, nil } select { case <-ctx.Done(): - return abandon() + return LockLease{}, acquireAbandoned, ctx.Err() case <-ticker.C: } } } +// burstScope is one scope's burst collection state: the currently open window +// (if any) and whether a runner tail owns the scope. runnerActive stays true +// across a runner handoff so arrivals never double-start a runner, and a window +// opened while a batch dispatches is handed to a fresh successor runner so +// windows never race each other for the Thread Lock. +type burstScope struct { + open bool + openedAt time.Time + batch []preludeWork + runnerActive bool +} + +// joinBurstBatch appends the routed work to its scope's collection window, +// opening a new window when none is open. It reports whether the caller must +// start the scope's runner tail (no runner is active for the scope). +func (c *Chat) joinBurstBatch(scope string, work preludeWork) (startRunner bool) { + c.burstMu.Lock() + defer c.burstMu.Unlock() + b := c.burstScopes[scope] + if b == nil { + b = &burstScope{} + c.burstScopes[scope] = b + } + if !b.open { + b.open = true + b.openedAt = time.Now() + } + b.batch = append(b.batch, work) + if b.runnerActive { + return false + } + b.runnerActive = true + return true +} + +// runBurstScope is the detached burst runner for one collection window: it +// sleeps until the window closes, takes the batch, acquires the scope's Thread +// Lock once, and runs every collected handler in join order under the single +// hold. It owns every joined work's span. +func (c *Chat) runBurstScope(ctx context.Context, scope string) { + c.burstMu.Lock() + b := c.burstScopes[scope] + assert(b != nil && b.open, "burst runner started without an open window") + deadline := b.openedAt.Add(c.options.DebounceInterval) + c.burstMu.Unlock() + + timer := time.NewTimer(time.Until(deadline)) + defer timer.Stop() + select { + case <-ctx.Done(): + c.abandonBurstWindow(ctx, scope) + return + case <-timer.C: + } + + c.burstMu.Lock() + batch := b.batch + b.batch = nil + b.open = false + c.burstMu.Unlock() + assert(len(batch) > 0, "burst window closed with no collected events") + + // The lock wait gets a fresh DetachTimeout starting when the window closes, + // so time spent collecting can never consume it. Derived from baseCtx so + // Shutdown still cancels it. + waitCtx, waitCancel := context.WithTimeout(c.baseCtx, c.options.DetachTimeout) + defer waitCancel() + + first := batch[0].event + lease, outcome, err := c.pollForLock(waitCtx, scope, nil) + if outcome != acquireHeld { + if outcome == acquireFailed { + c.logger.Error("chat burst acquire lock failed", "error", err, "adapter", first.Adapter, "thread_id", first.ThreadID, "size", len(batch)) + } else { + c.logger.Info("chat burst wait abandoned", "adapter", first.Adapter, "thread_id", first.ThreadID, "size", len(batch), "error", waitCtx.Err()) + } + for _, work := range batch { + c.safeEnd(work.span, waitOutcome(outcome), RouteAttr(work.route)) + } + c.finishBurstScope(scope) + return + } + + // Every accepted batch member gets its own DetachTimeout execution budget, + // so a member is never skipped merely because earlier members consumed a + // shared deadline; the batch as a whole is bounded by Shutdown (batchCtx) + // and by lease loss, which cancels the remaining members. + batchCtx, batchCancel := context.WithCancelCause(c.baseCtx) + defer batchCancel(nil) + + c.logger.Info("chat burst batch dispatch", "adapter", first.Adapter, "thread_id", first.ThreadID, "size", len(batch)) + stopRefresh, leaseLost := c.startLockRefresh(batchCtx, lease, first.ThreadID, batchCancel) + for i, work := range batch { + if batchCtx.Err() != nil { + c.logger.Info("chat burst batch abandoned", "adapter", first.Adapter, "thread_id", first.ThreadID, "remaining", len(batch)-i, "error", context.Cause(batchCtx)) + for _, rest := range batch[i:] { + c.safeEnd(rest.span, OutcomeIgnored, RouteAttr(rest.route)) + } + break + } + 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) + memberErr := work.run(memberCtx) + // Like runLockedTail, the preemption outcome follows the cancellation + // cause: a member stopped by lease loss is preempted, whatever it + // returned. + 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, "started", true) + c.safeEnd(work.span, OutcomePreempted, RouteAttr(work.route)) + } else { + c.endHandlerRun(memberCtx, work.event, work.route, work.span, memberErr) + } + memberCancel() + } + stopRefresh() + benignRelease := batchCtx.Err() != nil || leaseLost() + c.releaseTailLock(batchCtx, lease, first.ThreadID, benignRelease) + c.finishBurstScope(scope) +} + +// abandonBurstWindow surfaces an abandoned (cancelled or timed-out) collection +// window: every collected work's span closes as ignored, and the scope is +// finished so a window opened afterwards still gets a runner. +func (c *Chat) abandonBurstWindow(ctx context.Context, scope string) { + c.burstMu.Lock() + b := c.burstScopes[scope] + batch := b.batch + b.batch = nil + b.open = false + c.burstMu.Unlock() + if len(batch) > 0 { + first := batch[0].event + c.logger.Info("chat burst wait abandoned", "adapter", first.Adapter, "thread_id", first.ThreadID, "size", len(batch), "error", ctx.Err()) + } + for _, work := range batch { + c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) + } + c.finishBurstScope(scope) +} + +// finishBurstScope retires the scope's runner: a window opened while the batch +// dispatched is handed to a fresh successor runner (with its own DetachTimeout); +// otherwise the scope goes idle. runnerActive stays true across the handoff so +// an arrival can never double-start a runner, and the successor is started +// before this runner's inflight slot is released so Shutdown still drains it. +func (c *Chat) finishBurstScope(scope string) { + c.burstMu.Lock() + b := c.burstScopes[scope] + if b.open { + c.burstMu.Unlock() + c.startDetachedTail(preludeWork{scope: scope, burstRunner: true}) + return + } + b.runnerActive = false + delete(c.burstScopes, scope) + c.burstMu.Unlock() +} + func (c *Chat) clearPending(scope string, event *Event) { c.queueMu.Lock() defer c.queueMu.Unlock() - if c.pending[scope] == event { + if waiter := c.pending[scope]; waiter != nil && waiter.event == event { delete(c.pending, scope) } } diff --git a/runtime_test.go b/runtime_test.go index 723b023..56f68cc 100644 --- a/runtime_test.go +++ b/runtime_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" "sync" @@ -778,6 +779,7 @@ type fakeState struct { subscribed map[chat.ThreadID]bool seen map[string]bool locked map[string]chat.LockLease + lockSeq int acquireLockErr error isThreadSubscribedErr error isThreadSubscribedStarted chan struct{} @@ -863,7 +865,10 @@ func (s *fakeState) AcquireLock(ctx context.Context, key string, ttl time.Durati if _, ok := s.locked[key]; ok { return chat.LockLease{}, false, nil } - lease := chat.LockLease{Key: key, Token: key + "-token"} + // Tokens are unique per acquisition: a stale lease must never match a newer + // lock for the same key (the token-owned Lock Lease invariant). + s.lockSeq++ + lease := chat.LockLease{Key: key, Token: fmt.Sprintf("%s-token-%d", key, s.lockSeq)} s.locked[key] = lease return lease, true, nil } @@ -878,6 +883,19 @@ func (s *fakeState) ExtendLock(ctx context.Context, lease chat.LockLease, ttl ti return ok && held.Token == lease.Token, nil } +func (s *fakeState) ForceReleaseLock(ctx context.Context, key string) (bool, error) { + if err := ctx.Err(); err != nil { + return false, err + } + s.mu.Lock() + defer s.mu.Unlock() + if _, ok := s.locked[key]; !ok { + return false, nil + } + delete(s.locked, key) + return true, nil +} + func (s *fakeState) ReleaseLock(ctx context.Context, lease chat.LockLease) (bool, error) { if err := ctx.Err(); err != nil { return false, err diff --git a/state.go b/state.go index d40ac30..22004a6 100644 --- a/state.go +++ b/state.go @@ -20,3 +20,15 @@ type LockLease struct { Key string Token string } + +// LockForcer is the Optional Capability for force-releasing a Thread Lock (the +// ADR 0012 force/steerability path). ForceReleaseLock invalidates the current +// Lock Lease for key regardless of owner so a preempting delivery can acquire a +// fresh lease; it reports whether a lease was invalidated. The token-owned +// lease invariant is preserved: the previous holder's ExtendLock and +// ReleaseLock fail cleanly instead of touching the preemptor's newer lock. +// States that support preemption implement it; the runtime requires it only +// when RuntimeOptions.OnLockConflict is configured. +type LockForcer interface { + ForceReleaseLock(ctx context.Context, key string) (bool, error) +} diff --git a/state/memory/memory.go b/state/memory/memory.go index 263ebca..825ced7 100644 --- a/state/memory/memory.go +++ b/state/memory/memory.go @@ -26,6 +26,11 @@ type lockRecord struct { expiry time.Time } +var ( + _ chat.State = (*State)(nil) + _ chat.LockForcer = (*State)(nil) +) + func New() *State { return &State{ subscribed: map[chat.ThreadID]bool{}, @@ -141,6 +146,26 @@ func (s *State) ExtendLock(ctx context.Context, lease chat.LockLease, ttl time.D return true, nil } +// ForceReleaseLock invalidates the current lock for key regardless of owner +// (chat.LockForcer): the previous holder's lease token no longer matches +// anything, so its ExtendLock and ReleaseLock fail cleanly. +func (s *State) ForceReleaseLock(ctx context.Context, key string) (bool, error) { + if key == "" { + return false, errors.New("memory state: lock key is required") + } + if err := s.lockOperation(ctx); err != nil { + return false, err + } + defer s.mu.Unlock() + + s.pruneExpiredLocks(s.now()) + if _, ok := s.locks[key]; !ok { + return false, nil + } + delete(s.locks, key) + return true, nil +} + func (s *State) ReleaseLock(ctx context.Context, lease chat.LockLease) (bool, error) { if lease.Key == "" || lease.Token == "" { return false, errors.New("memory state: lock lease is required") diff --git a/state/nats/nats.go b/state/nats/nats.go index c23321a..70a5d44 100644 --- a/state/nats/nats.go +++ b/state/nats/nats.go @@ -35,7 +35,10 @@ type State struct { once sync.Once } -var _ chat.State = (*State)(nil) +var ( + _ chat.State = (*State)(nil) + _ chat.LockForcer = (*State)(nil) +) func New(ctx context.Context, opts Options) (*State, error) { if opts.Conn == nil { @@ -221,6 +224,54 @@ func (s *State) ExtendLock(ctx context.Context, lease chat.LockLease, ttl time.D return true, nil } +// ForceReleaseLock invalidates the current lock for key regardless of owner +// (chat.LockForcer). The delete is gated by the observed revision, so only the +// lease observed in this call is invalidated: a lease whose token changes +// hands between the get and the delete survives (reported as false), matching +// the single-statement atomicity of the Redis and Postgres implementations. A +// revision that moved while the token stayed the same is an ordinary renewal +// (ExtendLock) by the same holder, not a handover, so the delete is retried +// against the renewed revision. The previous holder's ExtendLock and +// ReleaseLock fail cleanly on the vanished entry. +func (s *State) ForceReleaseLock(ctx context.Context, key string) (bool, error) { + if key == "" { + return false, errors.New("nats state: lock key is required") + } + encoded := encodeKey(key) + entry, err := s.lock.Get(ctx, encoded) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("nats state: force release lock get: %w", err) + } + token := string(entry.Value()) + for { + err := s.lock.Delete(ctx, encoded, jetstream.LastRevision(entry.Revision())) + if err == nil { + return true, nil + } + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + if !isWrongLastSequence(err) { + return false, fmt.Errorf("nats state: force release lock: %w", err) + } + // The entry moved. Re-read: a same-token renewal retries against the + // new revision; a different token means the lease changed hands. + entry, err = s.lock.Get(ctx, encoded) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("nats state: force release lock get: %w", err) + } + if string(entry.Value()) != token { + return false, nil + } + } +} + func (s *State) ReleaseLock(ctx context.Context, lease chat.LockLease) (bool, error) { if lease.Key == "" || lease.Token == "" { return false, errors.New("nats state: lock lease is required") diff --git a/state/postgres/postgres.go b/state/postgres/postgres.go index fd8115a..28c2a1d 100644 --- a/state/postgres/postgres.go +++ b/state/postgres/postgres.go @@ -25,7 +25,10 @@ type State struct { once sync.Once } -var _ chat.State = (*State)(nil) +var ( + _ chat.State = (*State)(nil) + _ chat.LockForcer = (*State)(nil) +) func New(ctx context.Context, opts Options) (*State, error) { if opts.Pool == nil { @@ -167,6 +170,23 @@ func (s *State) ExtendLock(ctx context.Context, lease chat.LockLease, ttl time.D return tag.RowsAffected() == 1, nil } +// ForceReleaseLock invalidates the current lock for key regardless of owner +// (chat.LockForcer): the previous holder's lease token no longer matches +// anything, so its ExtendLock and ReleaseLock fail cleanly. +func (s *State) ForceReleaseLock(ctx context.Context, key string) (bool, error) { + if key == "" { + return false, errors.New("postgres state: lock key is required") + } + tag, err := s.pool.Exec(ctx, ` + DELETE FROM chat_runtime_locks + WHERE namespace = $1 AND lock_key = $2 AND expires_at > now() + `, s.namespace, key) + if err != nil { + return false, err + } + return tag.RowsAffected() == 1, nil +} + func (s *State) ReleaseLock(ctx context.Context, lease chat.LockLease) (bool, error) { if lease.Key == "" || lease.Token == "" { return false, errors.New("postgres state: lock lease is required") diff --git a/state/redis/redis.go b/state/redis/redis.go index f36f36c..48ebc6a 100644 --- a/state/redis/redis.go +++ b/state/redis/redis.go @@ -26,6 +26,11 @@ type State struct { once sync.Once } +var ( + _ chat.State = (*State)(nil) + _ chat.LockForcer = (*State)(nil) +) + func New(ctx context.Context, opts Options) (*State, error) { if opts.Client == nil { return nil, errors.New("redis state: client is required") @@ -114,6 +119,20 @@ func (s *State) ExtendLock(ctx context.Context, lease chat.LockLease, ttl time.D return result == 1, nil } +// ForceReleaseLock invalidates the current lock for key regardless of owner +// (chat.LockForcer): the previous holder's lease token no longer matches +// anything, so its ExtendLock and ReleaseLock fail cleanly. +func (s *State) ForceReleaseLock(ctx context.Context, key string) (bool, error) { + if key == "" { + return false, errors.New("redis state: lock key is required") + } + deleted, err := s.client.Del(ctx, s.key("lock", key)).Result() + if err != nil { + return false, err + } + return deleted == 1, nil +} + func (s *State) ReleaseLock(ctx context.Context, lease chat.LockLease) (bool, error) { if lease.Key == "" || lease.Token == "" { return false, errors.New("redis state: lock lease is required")