From 5199976541e977e89133558e782f6227f16dfc32 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 27 Aug 2026 21:41:27 +0000 Subject: [PATCH 1/2] feat(runtime): deferred-dispatch admission bound per ADR 0015 (MaxDetached / MaxDetachedPerTenant) Closes #44 --- CONTEXT.md | 4 + adapters/linear/admission_test.go | 88 ++++ adapters/linear/deferred_test.go | 1 + adapters/linear/linear.go | 8 + adapters/slack/admission_test.go | 113 +++++ adapters/slack/interactive.go | 26 + adapters/slack/slack.go | 8 + admission.go | 95 ++++ admission_test.go | 584 ++++++++++++++++++++++ command_interaction_hardening_test.go | 1 + concurrent_test.go | 1 + debounce_test.go | 2 + dispatch_deferred_test.go | 5 + dispatch_edge_test.go | 1 + docs/adr/0012-concurrency-strategy.md | 8 +- docs/adr/0015-runtime-coordination.md | 2 +- docs/how-to/deferred-dispatch.md | 13 + errors.go | 10 + examples/linear-agent-hello-world/main.go | 4 + export_test.go | 12 + lockscope_test.go | 1 + observer.go | 9 + observer_hardening_test.go | 1 + queue_edge_test.go | 2 + runtime.go | 95 +++- 25 files changed, 1084 insertions(+), 10 deletions(-) create mode 100644 adapters/linear/admission_test.go create mode 100644 adapters/slack/admission_test.go create mode 100644 admission.go create mode 100644 admission_test.go create mode 100644 export_test.go diff --git a/CONTEXT.md b/CONTEXT.md index ac00058..c2f572a 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -194,6 +194,10 @@ _Avoid_: Synchronous handler, fire-and-forget goroutine A runtime-derived context that outlives the inbound request so a deferred handler keeps working after acknowledgement; bounded by a detach timeout and cancelled by **Runtime Shutdown**. _Avoid_: Request context, context.Background() +**Admission Bound**: +The per-instance cap on admitted-but-incomplete deferred deliveries (`MaxDetached`, optionally `MaxDetachedPerTenant` per installation): a delivery arriving at the cap is rejected before acknowledgement and before dedupe marking with a typed signal the adapter maps to a shape-aware overload response. +_Avoid_: Rate limit, drop policy, queue depth + **Accepted Event**: A verified and normalized inbound event that the runtime has taken responsibility for, regardless of handler success. _Avoid_: Successful handler, retriable event diff --git a/adapters/linear/admission_test.go b/adapters/linear/admission_test.go new file mode 100644 index 0000000..7d77e50 --- /dev/null +++ b/adapters/linear/admission_test.go @@ -0,0 +1,88 @@ +package linear_test + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/coder/chat" + "github.com/coder/chat/adapters/linear" + "github.com/coder/chat/state/memory" +) + +// TestAdmissionRejectionAnswersRetryStatus proves the Linear mapping end to +// end: Linear redelivers webhooks on non-2xx, so a delivery the runtime's +// Admission Bound rejects (ADR 0015) is answered with a retry-inducing 503 — +// never acknowledged as handled. +func TestAdmissionRejectionAnswersRetryStatus(t *testing.T) { + t.Parallel() + + api := newLinearAPIServer(t, 3600) + now := time.UnixMilli(1_700_000_000_000) + + adapter, err := linear.New(context.Background(), linear.Options{ + WebhookSecret: "whsec", + ClientCredentials: linear.ClientCredentials{ClientID: "client", ClientSecret: "secret"}, + APIBaseURL: api.URL, + Client: api.Client(), + Now: func() time.Time { return now }, + }) + if err != nil { + t.Fatalf("new adapter: %v", err) + } + bot, err := chat.New(context.Background(), + chat.WithState(memory.New()), + chat.WithAdapter(adapter), + chat.WithRuntimeOptions(chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Dispatch: chat.DispatchDeferred, + MaxDetached: 1, + DetachTimeout: 5 * time.Second, + }), + ) + if err != nil { + t.Fatalf("new runtime: %v", err) + } + defer func() { _ = bot.Shutdown(context.Background()) }() + + var wg sync.WaitGroup + wg.Add(1) + release := make(chan struct{}) + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + defer wg.Done() + <-release + return nil + }) + + handler, err := bot.Webhook("linear") + if err != nil { + t.Fatalf("webhook: %v", err) + } + serve := func(body string) *httptest.ResponseRecorder { + bodyBytes := []byte(body) + req := httptest.NewRequest(http.MethodPost, "/linear", bytes.NewReader(bodyBytes)) + signLinearRequest(req, "whsec", bodyBytes) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + return rec + } + + // First delivery saturates MaxDetached=1. + if rec := serve(createdPayload(now, "C1", "hello", "U1", "User One", "APP1")); rec.Code != http.StatusOK { + t.Fatalf("first delivery status = %d, body = %s", rec.Code, rec.Body.String()) + } + // Second delivery hits the Admission Bound: retry-inducing 503, so + // Linear's redelivery covers the event. + rec := serve(createdPayloadForOrganization(now, "ORG1", "C2", "hello again", "U1", "User One", "APP1")) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("saturated delivery status = %d, want 503 (body = %s)", rec.Code, rec.Body.String()) + } + + close(release) + wg.Wait() +} diff --git a/adapters/linear/deferred_test.go b/adapters/linear/deferred_test.go index 9495f5a..7d693ef 100644 --- a/adapters/linear/deferred_test.go +++ b/adapters/linear/deferred_test.go @@ -38,6 +38,7 @@ func TestDeferredDispatchPostsFromDetachedTail(t *testing.T) { DedupeTTL: time.Hour, ThreadLockTTL: time.Hour, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, }), ) diff --git a/adapters/linear/linear.go b/adapters/linear/linear.go index 4b4e1f7..c6bbe5e 100644 --- a/adapters/linear/linear.go +++ b/adapters/linear/linear.go @@ -287,6 +287,14 @@ func (a *Adapter) Webhook(dispatch chat.DispatchFunc) http.Handler { } if ok { if err := dispatch(r.Context(), event); err != nil { + if errors.Is(err, chat.ErrAdmissionRejected) { + // Linear redelivers webhooks on non-2xx: a retry-inducing + // status lets the platform's own retry cover the event, + // and because the delivery was never dedupe-marked that + // retry is not deduped away. + http.Error(w, "linear runtime at capacity", http.StatusServiceUnavailable) + return + } http.Error(w, err.Error(), http.StatusInternalServerError) return } diff --git a/adapters/slack/admission_test.go b/adapters/slack/admission_test.go new file mode 100644 index 0000000..05ea137 --- /dev/null +++ b/adapters/slack/admission_test.go @@ -0,0 +1,113 @@ +package slack_test + +import ( + "context" + "fmt" + "net/http" + "net/url" + "strings" + "testing" + "time" + + "github.com/coder/chat" + "github.com/coder/chat/adapters/slack" +) + +// newAdmissionTestAdapter builds a single-tenant adapter with static identity +// so webhook shapes reach dispatch without any Slack API traffic. +func newAdmissionTestAdapter(t *testing.T, now time.Time) *slack.Adapter { + t.Helper() + adapter, err := slack.New(context.Background(), slack.Options{ + SigningSecret: "secret", + BotToken: "xoxb-test", + TeamID: "T1", + BotUserID: "UBOT", + BotID: "BBOT", + Now: func() time.Time { return now }, + }) + if err != nil { + t.Fatalf("new slack adapter: %v", err) + } + return adapter +} + +// rejectingDispatch simulates the runtime's Admission Bound rejection (ADR +// 0015), wrapped the way dispatch wraps its sentinel. +func rejectingDispatch(context.Context, *chat.Event) error { + return fmt.Errorf("dispatch: %w", chat.ErrAdmissionRejected) +} + +// TestEventAdmissionRejectionAnswersRetryStatus pins the Events API mapping: +// Slack redelivers event callbacks, so an admission rejection answers a +// retry-inducing 503 instead of acknowledging work the runtime refused. +func TestEventAdmissionRejectionAnswersRetryStatus(t *testing.T) { + t.Parallel() + + now := time.Unix(1_700_000_000, 0) + handler := newAdmissionTestAdapter(t, now).Webhook(rejectingDispatch) + + rec := serveSignedSlackWebhook(t, handler, now, `{ + "type":"event_callback", + "team_id":"T1", + "event_id":"Ev1", + "event":{ + "type":"app_mention", + "channel":"C1", + "user":"U1", + "text":"<@UBOT> hi", + "ts":"111.000" + } + }`, "", "") + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("event admission rejection status = %d, want 503 (body = %s)", rec.Code, rec.Body.String()) + } +} + +// TestCommandAdmissionRejectionAnswersBusySignal pins the slash-command +// mapping: Slack does not redeliver commands, so the shape's ack body carries +// a truthful, user-visible busy message instead of a retry-inducing status. +func TestCommandAdmissionRejectionAnswersBusySignal(t *testing.T) { + t.Parallel() + + now := time.Unix(1_700_000_000, 0) + handler := newAdmissionTestAdapter(t, now).Webhook(rejectingDispatch) + + form := url.Values{} + form.Set("command", "/deploy") + form.Set("team_id", "T1") + form.Set("channel_id", "C1") + form.Set("user_id", "U1") + form.Set("trigger_id", "trig") + rec := serveSignedSlackForm(t, handler, now, form) + if rec.Code != http.StatusOK { + t.Fatalf("command admission rejection status = %d, want 200 with a busy body (body = %s)", rec.Code, rec.Body.String()) + } + body := rec.Body.String() + if !strings.Contains(body, "at capacity") || !strings.Contains(body, "was not run") { + t.Fatalf("command admission rejection body = %q, want a truthful busy signal", body) + } +} + +// TestInteractionAdmissionRejectionAnswersRetryStatus pins the block_actions +// mapping: the ack body is not rendered to the user, so the truthful signal is +// a non-2xx ack Slack surfaces as a visible failure on the component. +func TestInteractionAdmissionRejectionAnswersRetryStatus(t *testing.T) { + t.Parallel() + + now := time.Unix(1_700_000_000, 0) + handler := newAdmissionTestAdapter(t, now).Webhook(rejectingDispatch) + + rec := serveSignedSlackInteractivity(t, handler, now, `{ + "type":"block_actions", + "team":{"id":"T1"}, + "user":{"id":"U1"}, + "channel":{"id":"C1"}, + "container":{"channel_id":"C1","message_ts":"333.000"}, + "trigger_id":"trigger-999", + "response_url":"https://hooks.slack.com/actions/T1/999", + "actions":[{"action_id":"approve","block_id":"b1","value":"yes","type":"button"}] + }`) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("interaction admission rejection status = %d, want 503 (body = %s)", rec.Code, rec.Body.String()) + } +} diff --git a/adapters/slack/interactive.go b/adapters/slack/interactive.go index 82dd6f3..e173aaf 100644 --- a/adapters/slack/interactive.go +++ b/adapters/slack/interactive.go @@ -64,12 +64,29 @@ func (a *Adapter) handleCommandForm(w http.ResponseWriter, r *http.Request, disp return } if err := dispatch(r.Context(), event); err != nil { + if errors.Is(err, chat.ErrAdmissionRejected) { + // Slack does not redeliver slash commands, so a retry-inducing + // status would lose the invocation with only a generic Slack error + // shown. The shape's acknowledgement contract renders the ack body + // as an ephemeral message to the invoking user, so answer 200 with + // a truthful busy signal instead — never a silent failure, never a + // fake success. + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(commandBusyMessage)) + return + } http.Error(w, err.Error(), http.StatusInternalServerError) return } w.WriteHeader(http.StatusOK) } +// commandBusyMessage is the truthful busy signal a slash command invoker sees +// when the runtime's Admission Bound rejects the invocation (ADR 0015). It +// renders as the command's ephemeral response. +const commandBusyMessage = "The bot is at capacity right now. Your command was not run — please try again in a moment." + // normalizeCommand decodes and validates a slash command into a Command Event. A // missing required field or a cross-tenant team is a 400, like a malformed event. func (a *Adapter) normalizeCommand(r *http.Request, form url.Values) (*chat.Event, error) { @@ -226,6 +243,15 @@ func (a *Adapter) handleInteractionForm(w http.ResponseWriter, r *http.Request, return } if err := dispatch(r.Context(), event); err != nil { + if errors.Is(err, chat.ErrAdmissionRejected) { + // Slack does not redeliver block_actions, and the shape's ack body + // is not rendered to the user; a non-2xx ack is the truthful + // signal the shape supports — Slack surfaces the failed delivery + // as a visible warning on the activated component, and the + // rejection stays observable through Runtime Observation. + http.Error(w, "slack runtime at capacity", http.StatusServiceUnavailable) + return + } http.Error(w, err.Error(), http.StatusInternalServerError) return } diff --git a/adapters/slack/slack.go b/adapters/slack/slack.go index 4846c5a..f678f00 100644 --- a/adapters/slack/slack.go +++ b/adapters/slack/slack.go @@ -273,6 +273,14 @@ func (a *Adapter) Webhook(dispatch chat.DispatchFunc) http.Handler { } if ok { if err := dispatch(r.Context(), event); err != nil { + if errors.Is(err, chat.ErrAdmissionRejected) { + // Events API callbacks are platform-redelivered: a + // retry-inducing status lets Slack's own retry cover + // the event, and because the delivery was never + // dedupe-marked that retry is not deduped away. + http.Error(w, "slack runtime at capacity", http.StatusServiceUnavailable) + return + } http.Error(w, err.Error(), http.StatusInternalServerError) return } diff --git a/admission.go b/admission.go new file mode 100644 index 0000000..e85e461 --- /dev/null +++ b/admission.go @@ -0,0 +1,95 @@ +package chat + +import "sync" + +// tenantKey is the ADR 0006 installation identity the per-tenant admission +// ceiling is keyed on. An empty tenant is a valid key: a single-tenant +// adapter's deliveries are capped under (adapter, "") rather than escaping +// the ceiling or colliding with another adapter's empty tenant. +type tenantKey struct { + adapter string + tenant string +} + +// admissionGate is the deferred-dispatch admission bound (ADR 0015): a +// per-instance cap on admitted-but-incomplete deferred deliveries, with an +// optional per-installation ceiling through the same rejection path. It is +// constructed only under DispatchDeferred and closed by Runtime Shutdown so +// a shutting-down instance sheds new deliveries to the platform's retry +// instead of admitting work it is about to cancel. +type admissionGate struct { + mu sync.Mutex + // capacity is MaxDetached; perTenant is MaxDetachedPerTenant (0 = disabled). + capacity int + perTenant int + inUse int + // tenants counts held slots per installation; entries are deleted when a + // tenant's count reaches zero so an idle tenant retains no map entry. + tenants map[tenantKey]int + closed bool +} + +func newAdmissionGate(capacity, perTenant int) *admissionGate { + assert(capacity > 0, "admission gate capacity must be positive") + assert(perTenant >= 0, "admission per-tenant ceiling must not be negative") + gate := &admissionGate{capacity: capacity, perTenant: perTenant} + if perTenant > 0 { + gate.tenants = map[tenantKey]int{} + } + return gate +} + +// admit reserves one admission slot for a deferred delivery. It reports false +// when the gate is closed (Runtime Shutdown), the instance is at capacity, or +// the delivery's installation is at its per-tenant ceiling. The returned +// release is idempotent and must be called exactly when the delivery stops +// retaining work: when its prelude resolves or fails, or when its detached +// tail goroutine returns. +func (g *admissionGate) admit(adapter, tenant string) (release func(), ok bool) { + g.mu.Lock() + defer g.mu.Unlock() + if g.closed || g.inUse >= g.capacity { + return nil, false + } + key := tenantKey{adapter: adapter, tenant: tenant} + if g.perTenant > 0 && g.tenants[key] >= g.perTenant { + return nil, false + } + g.inUse++ + if g.perTenant > 0 { + g.tenants[key]++ + } + return sync.OnceFunc(func() { g.release(key) }), true +} + +func (g *admissionGate) release(key tenantKey) { + g.mu.Lock() + defer g.mu.Unlock() + assert(g.inUse > 0, "admission release without a held slot") + g.inUse-- + if g.perTenant > 0 { + count := g.tenants[key] + assert(count > 0, "admission release without a held tenant slot") + if count == 1 { + delete(g.tenants, key) + } else { + g.tenants[key] = count - 1 + } + } +} + +// close permanently rejects new admissions. Held slots are unaffected: their +// releases still run so accounting stays exact while in-flight tails drain. +func (g *admissionGate) close() { + g.mu.Lock() + defer g.mu.Unlock() + g.closed = true +} + +// tenantEntries reports the number of live per-tenant counter entries, for +// tests proving released tenants are garbage collected. +func (g *admissionGate) tenantEntries() int { + g.mu.Lock() + defer g.mu.Unlock() + return len(g.tenants) +} diff --git a/admission_test.go b/admission_test.go new file mode 100644 index 0000000..413af8b --- /dev/null +++ b/admission_test.go @@ -0,0 +1,584 @@ +package chat_test + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/chat" +) + +// newAdmissionRuntime builds a deferred runtime with an admission cap small +// enough to saturate deterministically, an observer, and optional option +// mutations. +func newAdmissionRuntime(t *testing.T, state chat.State, adapter chat.Adapter, observer chat.Observer, opts ...func(*chat.RuntimeOptions)) *chat.Chat { + t.Helper() + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyDrop, + Dispatch: chat.DispatchDeferred, + MaxDetached: 1, + DetachTimeout: 5 * time.Second, + } + for _, opt := range opts { + opt(&options) + } + bot, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithLogger(slog.New(slog.NewTextHandler(newSyncBuffer(), nil))), + chat.WithObserver(observer), + chat.WithRuntimeOptions(options), + ) + if err != nil { + t.Fatalf("new admission runtime: %v", err) + } + return bot +} + +// tenantMentionEvent is mentionEvent with an explicit Platform Tenant. +func tenantMentionEvent(id string, threadID chat.ThreadID, tenant string) chat.Event { + ev := mentionEvent(id, threadID) + ev.Tenant = tenant + return ev +} + +// postEventBody posts one event through the fake adapter webhook and returns +// the status plus response body, so tests can assert the admission sentinel +// surfaced to the adapter. +func postEventBody(t *testing.T, bot *chat.Chat, adapter string, ev chat.Event) (int, string) { + t.Helper() + handler, err := bot.Webhook(adapter) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(ev) + if err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodPost, "/webhook", bytes.NewReader(body)) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + return rec.Code, rec.Body.String() +} + +// eventually polls cond until it holds or the deadline passes. +func eventually(t *testing.T, timeout time.Duration, cond func() bool, msg string) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal(msg) +} + +// handlerRecorder tracks handled event IDs and blocks handlers on a shared +// gate until released. +type handlerRecorder struct { + mu sync.Mutex + handled []string + gate chan struct{} + started chan string +} + +func newHandlerRecorder() *handlerRecorder { + return &handlerRecorder{gate: make(chan struct{}), started: make(chan string, 64)} +} + +func (h *handlerRecorder) handle(_ context.Context, ev *chat.MessageEvent) error { + h.started <- ev.Event.ID + <-h.gate + h.mu.Lock() + defer h.mu.Unlock() + h.handled = append(h.handled, ev.Event.ID) + return nil +} + +func (h *handlerRecorder) release() { close(h.gate) } + +func (h *handlerRecorder) handledIDs() []string { + h.mu.Lock() + defer h.mu.Unlock() + return append([]string(nil), h.handled...) +} + +func (h *handlerRecorder) waitStarted(t *testing.T, want string) { + t.Helper() + select { + case got := <-h.started: + if got != want { + t.Fatalf("handler started for %q, want %q", got, want) + } + case <-time.After(time.Second): + t.Fatalf("handler for %q did not start", want) + } +} + +// TestAdmissionRejectsAtMaxDetachedWithoutDedupeMark saturates MaxDetached=1 +// and verifies the rejection contract: the webhook sees the typed sentinel +// (never a 2xx), the observation and terminal outcome are emitted with adapter +// and tenant, and — because rejection happens before dedupe marking — the same +// Event Identity is handled normally once capacity frees. +func TestAdmissionRejectsAtMaxDetachedWithoutDedupeMark(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer) + handlers := newHandlerRecorder() + bot.OnNewMention(handlers.handle) + + if status := postEvent(t, bot, "fake", mentionEvent("event-1", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first event status = %d", status) + } + handlers.waitStarted(t, "event-1") + + status, body := postEventBody(t, bot, "fake", mentionEvent("event-2", "fake:v1:thread-2")) + if status == http.StatusOK { + t.Fatalf("saturated dispatch acknowledged 2xx, body = %q", body) + } + if !strings.Contains(body, "admission rejected") { + t.Fatalf("rejection body = %q, want the admission sentinel", body) + } + if !observer.hasEvent(chat.ObsAdmissionRejected) { + t.Fatalf("observations = %v, want admission_rejected", observer.eventNames()) + } + assertAdmissionRejectedAttrs(t, observer, "fake", "tenant") + if !hasOutcome(observer, chat.OutcomeAdmissionRejected) { + t.Fatalf("outcomes = %v, want admission-rejected", observer.terminalOutcomes()) + } + // The rejected delivery was never marked in Event Identity dedupe: once + // capacity frees, redelivering the same event id runs the handler instead + // of resolving as a duplicate. + handlers.release() + eventually(t, 5*time.Second, func() bool { + status, _ := postEventBody(t, bot, "fake", mentionEvent("event-2", "fake:v1:thread-2")) + return status == http.StatusOK + }, "capacity did not free after handler completion") + eventually(t, 5*time.Second, func() bool { + for _, id := range handlers.handledIDs() { + if id == "event-2" { + return true + } + } + return false + }, "redelivered rejected event was not handled (dedupe-marked before rejection?)") +} + +// TestAdmissionSlotReleasedOnErroredAndResolvedPreludes floods a MaxDetached=1 +// runtime with preludes that fail (state error) or resolve without detached +// work (ignored events) and proves none of them leak admission capacity. +func TestAdmissionSlotReleasedOnErroredAndResolvedPreludes(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer) + handlers := newHandlerRecorder() + bot.OnNewMention(handlers.handle) + + // Errored preludes: AcquireLock fails, dispatch returns the error, and the + // admission slot must come back every time. + state.mu.Lock() + state.acquireLockErr = errors.New("acquire lock boom") + state.mu.Unlock() + for i := range 3 { + status, body := postEventBody(t, bot, "fake", mentionEvent("errored-"+string(rune('a'+i)), "fake:v1:thread-err")) + if status == http.StatusOK { + t.Fatalf("errored prelude %d acknowledged 2xx", i) + } + if strings.Contains(body, "admission rejected") { + t.Fatalf("errored prelude %d hit the admission cap: slot leaked (body = %q)", i, body) + } + } + state.mu.Lock() + state.acquireLockErr = nil + state.mu.Unlock() + + // Resolved preludes: ignored events (no payload handler routes them) + // resolve synchronously and must release their slots too. + for i := range 3 { + ev := chat.Event{ID: "ignored-" + string(rune('a'+i)), Adapter: "fake", Tenant: "tenant", ThreadID: "fake:v1:thread-ign"} + if status := postEvent(t, bot, "fake", ev); status != http.StatusOK { + t.Fatalf("ignored event %d status = %d", i, status) + } + } + + // With every slot returned, a routed event is admitted. + if status := postEvent(t, bot, "fake", mentionEvent("event-ok", "fake:v1:thread-ok")); status != http.StatusOK { + t.Fatalf("post-flood event status = %d", status) + } + handlers.waitStarted(t, "event-ok") + handlers.release() +} + +// stalledReleaseState blocks ReleaseLock until unblocked, simulating stalled +// tail cleanup after the handler already returned. +type stalledReleaseState struct { + *fakeState + releaseStarted chan struct{} + releaseGate chan struct{} + startOnce sync.Once +} + +func newStalledReleaseState() *stalledReleaseState { + return &stalledReleaseState{ + fakeState: newFakeState(), + releaseStarted: make(chan struct{}), + releaseGate: make(chan struct{}), + } +} + +func (s *stalledReleaseState) ReleaseLock(ctx context.Context, lease chat.LockLease) (bool, error) { + s.startOnce.Do(func() { close(s.releaseStarted) }) + <-s.releaseGate + return s.fakeState.ReleaseLock(ctx, lease) +} + +// TestAdmissionSlotHeldUntilTailGoroutineReturns pins the bounded-retention +// invariant: a handler that returned but whose tail goroutine is stalled in +// cleanup (a blocked lock release) still occupies its admission slot. +func TestAdmissionSlotHeldUntilTailGoroutineReturns(t *testing.T) { + t.Parallel() + + state := newStalledReleaseState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer) + handlers := newHandlerRecorder() + handlers.release() // handlers return immediately; only cleanup stalls + bot.OnNewMention(handlers.handle) + + if status := postEvent(t, bot, "fake", mentionEvent("event-1", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first event status = %d", status) + } + select { + case <-state.releaseStarted: + case <-time.After(5 * time.Second): + t.Fatal("tail cleanup did not reach ReleaseLock") + } + + // The handler has returned, but the tail goroutine is parked in cleanup: + // the slot must still be held. + status, body := postEventBody(t, bot, "fake", mentionEvent("event-2", "fake:v1:thread-2")) + if status == http.StatusOK || !strings.Contains(body, "admission rejected") { + t.Fatalf("stalled-cleanup dispatch = %d %q, want admission rejection", status, body) + } + + close(state.releaseGate) + eventually(t, 5*time.Second, func() bool { + status, _ := postEventBody(t, bot, "fake", mentionEvent("event-3", "fake:v1:thread-3")) + return status == http.StatusOK + }, "slot did not free after the tail goroutine returned") +} + +// TestAdmissionClosedOnShutdown pins the shutdown contract: once Runtime +// Shutdown begins, new deferred deliveries are rejected with the admission +// sentinel (observable, platform retry covers them) instead of being admitted +// into a runtime that is cancelling its work. +func TestAdmissionClosedOnShutdown(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer, func(o *chat.RuntimeOptions) { + o.MaxDetached = 8 + }) + + if err := bot.Shutdown(context.Background()); err != nil { + t.Fatalf("shutdown: %v", err) + } + + status, body := postEventBody(t, bot, "fake", mentionEvent("late-event", "fake:v1:thread-1")) + if status == http.StatusOK || !strings.Contains(body, "admission rejected") { + t.Fatalf("post-shutdown dispatch = %d %q, want admission rejection", status, body) + } + if !observer.hasEvent(chat.ObsAdmissionRejected) { + t.Fatalf("observations = %v, want admission_rejected", observer.eventNames()) + } + if !hasOutcome(observer, chat.OutcomeAdmissionRejected) { + t.Fatalf("outcomes = %v, want admission-rejected", observer.terminalOutcomes()) + } +} + +// TestAdmissionOptionValidation pins constructor validation: MaxDetached must +// be positive and MaxDetachedPerTenant non-negative under DispatchDeferred, +// while DispatchSync ignores both (the DetachTimeout precedent). +func TestAdmissionOptionValidation(t *testing.T) { + t.Parallel() + + if got := chat.DefaultRuntimeOptions().MaxDetached; got != 1024 { + t.Fatalf("default MaxDetached = %d, want 1024", got) + } + + newWith := func(options chat.RuntimeOptions) error { + _, err := chat.New(context.Background(), + chat.WithState(newFakeState()), + chat.WithAdapter(newFakeAdapter("fake")), + chat.WithRuntimeOptions(options), + ) + return err + } + deferred := func(mutate func(*chat.RuntimeOptions)) chat.RuntimeOptions { + options := chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Dispatch: chat.DispatchDeferred, + DetachTimeout: time.Second, + MaxDetached: 1, + } + mutate(&options) + return options + } + + for name, tt := range map[string]struct { + options chat.RuntimeOptions + wantErr string + }{ + "zero max detached under deferred": { + options: deferred(func(o *chat.RuntimeOptions) { o.MaxDetached = 0 }), + wantErr: "max detached must be positive", + }, + "negative max detached under deferred": { + options: deferred(func(o *chat.RuntimeOptions) { o.MaxDetached = -1 }), + wantErr: "max detached must be positive", + }, + "negative per-tenant ceiling under deferred": { + options: deferred(func(o *chat.RuntimeOptions) { o.MaxDetachedPerTenant = -1 }), + wantErr: "max detached per tenant must not be negative", + }, + "valid per-tenant ceiling under deferred": { + options: deferred(func(o *chat.RuntimeOptions) { o.MaxDetachedPerTenant = 4 }), + }, + "sync ignores both": { + options: chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + MaxDetached: 0, + MaxDetachedPerTenant: -1, + }, + }, + } { + t.Run(name, func(t *testing.T) { + err := newWith(tt.options) + if tt.wantErr == "" { + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("error = %v, want %q", err, tt.wantErr) + } + }) + } +} + +// TestAdmissionPerTenantCeilingKeysOnInstallation proves MaxDetachedPerTenant +// caps one (adapter, tenant) installation without starving others, and that an +// empty tenant is a countable key rather than an escape from the ceiling. +func TestAdmissionPerTenantCeilingKeysOnInstallation(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer, func(o *chat.RuntimeOptions) { + o.MaxDetached = 10 + o.MaxDetachedPerTenant = 1 + }) + handlers := newHandlerRecorder() + bot.OnNewMention(handlers.handle) + + if status := postEvent(t, bot, "fake", tenantMentionEvent("empty-1", "fake:v1:thread-1", "")); status != http.StatusOK { + t.Fatalf("first empty-tenant event status = %d", status) + } + handlers.waitStarted(t, "empty-1") + + status, body := postEventBody(t, bot, "fake", tenantMentionEvent("empty-2", "fake:v1:thread-2", "")) + if status == http.StatusOK || !strings.Contains(body, "admission rejected") { + t.Fatalf("empty-tenant ceiling dispatch = %d %q, want admission rejection", status, body) + } + assertAdmissionRejectedAttrs(t, observer, "fake", "") + + // A different installation still has headroom. + if status := postEvent(t, bot, "fake", tenantMentionEvent("acme-1", "fake:v1:thread-3", "acme")); status != http.StatusOK { + t.Fatalf("other-tenant event status = %d", status) + } + handlers.waitStarted(t, "acme-1") + + handlers.release() + eventually(t, 5*time.Second, func() bool { + status, _ := postEventBody(t, bot, "fake", tenantMentionEvent("empty-3", "fake:v1:thread-4", "")) + return status == http.StatusOK + }, "per-tenant capacity did not free after handler completion") +} + +// TestAdmissionPerTenantCounterGC proves the per-tenant counters are garbage +// collected: once a tenant's deliveries all release capacity, no map entry +// remains for it. +func TestAdmissionPerTenantCounterGC(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer, func(o *chat.RuntimeOptions) { + o.MaxDetached = 10 + o.MaxDetachedPerTenant = 2 + }) + handlers := newHandlerRecorder() + bot.OnNewMention(handlers.handle) + + for _, ev := range []chat.Event{ + tenantMentionEvent("t1-a", "fake:v1:thread-1", "tenant-1"), + tenantMentionEvent("t1-b", "fake:v1:thread-2", "tenant-1"), + tenantMentionEvent("t2-a", "fake:v1:thread-3", "tenant-2"), + } { + if status := postEvent(t, bot, "fake", ev); status != http.StatusOK { + t.Fatalf("event %s status = %d", ev.ID, status) + } + } + for range 3 { + <-handlers.started + } + if entries := chat.AdmissionTenantEntries(bot); entries != 2 { + t.Fatalf("in-flight tenant entries = %d, want 2", entries) + } + + handlers.release() + eventually(t, 5*time.Second, func() bool { + return chat.AdmissionTenantEntries(bot) == 0 + }, "tenant counter entries were not garbage collected after release") +} + +// TestAdmissionCountsParkedQueueWaiters pins the queue row of the ADR 0015 +// retention table: a parked waiter holds an admission slot just like a running +// tail. +func TestAdmissionCountsParkedQueueWaiters(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer, func(o *chat.RuntimeOptions) { + o.Concurrency = chat.ConcurrencyQueue + o.MaxDetached = 2 + }) + handlers := newHandlerRecorder() + bot.OnNewMention(handlers.handle) + + // Slot 1: running tail on thread-1. Slot 2: parked waiter behind it. + if status := postEvent(t, bot, "fake", mentionEvent("event-1", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first event status = %d", status) + } + handlers.waitStarted(t, "event-1") + if status := postEvent(t, bot, "fake", mentionEvent("event-2", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("queued event status = %d", status) + } + + // The parked waiter counts: a third delivery for a completely idle thread + // is rejected. + status, body := postEventBody(t, bot, "fake", mentionEvent("event-3", "fake:v1:thread-2")) + if status == http.StatusOK || !strings.Contains(body, "admission rejected") { + t.Fatalf("dispatch with parked waiter = %d %q, want admission rejection", status, body) + } + + handlers.release() + eventually(t, 5*time.Second, func() bool { + for _, id := range handlers.handledIDs() { + if id == "event-2" { + return true + } + } + return false + }, "queued waiter did not run after the in-flight handler finished") +} + +// TestAdmissionCountsConcurrentSlotWaiters pins the concurrent row of the ADR +// 0015 retention table: a tail waiting for a MaxConcurrent slot holds an +// admission slot. +func TestAdmissionCountsConcurrentSlotWaiters(t *testing.T) { + t.Parallel() + + state := newFakeState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer, func(o *chat.RuntimeOptions) { + o.Concurrency = chat.ConcurrencyConcurrent + o.MaxConcurrent = 1 + o.MaxDetached = 2 + }) + handlers := newHandlerRecorder() + bot.OnNewMention(handlers.handle) + + if status := postEvent(t, bot, "fake", mentionEvent("event-1", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first event status = %d", status) + } + handlers.waitStarted(t, "event-1") + // Slot-waiter: admitted, parked behind MaxConcurrent. + if status := postEvent(t, bot, "fake", mentionEvent("event-2", "fake:v1:thread-2")); status != http.StatusOK { + t.Fatalf("slot-waiting event status = %d", status) + } + + status, body := postEventBody(t, bot, "fake", mentionEvent("event-3", "fake:v1:thread-3")) + if status == http.StatusOK || !strings.Contains(body, "admission rejected") { + t.Fatalf("dispatch with slot-waiter = %d %q, want admission rejection", status, body) + } + + handlers.release() + eventually(t, 5*time.Second, func() bool { + return len(handlers.handledIDs()) == 2 + }, "slot-waiter did not run after a concurrency slot freed") +} + +// hasOutcome reports whether the observer recorded the terminal outcome. +func hasOutcome(observer *recordingObserver, want chat.DispatchOutcome) bool { + for _, outcome := range observer.terminalOutcomes() { + if outcome == want { + return true + } + } + return false +} + +// assertAdmissionRejectedAttrs verifies the admission_rejected observation +// carries the adapter and tenant attributes required by ADR 0015. +func assertAdmissionRejectedAttrs(t *testing.T, observer *recordingObserver, adapter, tenant string) { + t.Helper() + observer.mu.Lock() + defer observer.mu.Unlock() + for _, ev := range observer.events { + if ev.name != chat.ObsAdmissionRejected { + continue + } + var gotAdapter, gotTenant bool + for _, attr := range ev.attrs { + if attr.Key == chat.AttrAdapter && attr.Value == adapter { + gotAdapter = true + } + if attr.Key == chat.AttrTenant && attr.Value == tenant { + gotTenant = true + } + } + if gotAdapter && gotTenant { + return + } + } + t.Fatalf("no admission_rejected observation carried adapter=%q tenant=%q", adapter, tenant) +} diff --git a/command_interaction_hardening_test.go b/command_interaction_hardening_test.go index a539188..8f7fa89 100644 --- a/command_interaction_hardening_test.go +++ b/command_interaction_hardening_test.go @@ -305,6 +305,7 @@ func TestQueuedCommandRunsAfterInFlightHandler(t *testing.T) { ThreadLockTTL: time.Hour, Concurrency: chat.ConcurrencyQueue, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, }), ) diff --git a/concurrent_test.go b/concurrent_test.go index c3998bf..0fa3440 100644 --- a/concurrent_test.go +++ b/concurrent_test.go @@ -41,6 +41,7 @@ func newConcurrentRuntime(t *testing.T, state chat.State, adapter chat.Adapter, Concurrency: chat.ConcurrencyConcurrent, MaxConcurrent: 4, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, } for _, m := range mutate { diff --git a/debounce_test.go b/debounce_test.go index a4b4e71..7ff16b6 100644 --- a/debounce_test.go +++ b/debounce_test.go @@ -20,6 +20,7 @@ func newDebounceRuntime(t *testing.T, state chat.State, adapter chat.Adapter, lo Concurrency: chat.ConcurrencyDebounce, DebounceInterval: 50 * time.Millisecond, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, } for _, m := range mutate { @@ -449,6 +450,7 @@ func TestDebounceConstructionValidation(t *testing.T) { Concurrency: chat.ConcurrencyDebounce, DebounceInterval: 50 * time.Millisecond, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: time.Second, } mutate(&options) diff --git a/dispatch_deferred_test.go b/dispatch_deferred_test.go index 4ccb7a5..0d768bb 100644 --- a/dispatch_deferred_test.go +++ b/dispatch_deferred_test.go @@ -21,6 +21,7 @@ func newDeferredRuntime(t *testing.T, state chat.State, adapter chat.Adapter, op ThreadLockTTL: time.Hour, Concurrency: chat.ConcurrencyDrop, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, } for _, opt := range opts { @@ -505,6 +506,7 @@ func TestRuntimeConstructionValidatesDispatchAndConcurrency(t *testing.T) { DedupeTTL: time.Hour, ThreadLockTTL: time.Hour, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 0, }); err == nil { t.Fatal("expected deferred dispatch without detach timeout to fail") @@ -564,6 +566,7 @@ func TestQueueStrategyDispatchesMostRecentSupersededFollowUp(t *testing.T) { ThreadLockTTL: 40 * time.Millisecond, Concurrency: chat.ConcurrencyQueue, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, }), ) @@ -708,6 +711,7 @@ func TestDeferredDispatchObservationRecords(t *testing.T) { ThreadLockTTL: 40 * time.Millisecond, Concurrency: chat.ConcurrencyDrop, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, }), ) @@ -777,6 +781,7 @@ func TestDeferredHandlerCancelledOnLeaseLoss(t *testing.T) { ThreadLockTTL: 100 * time.Millisecond, Concurrency: chat.ConcurrencyDrop, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, }), ) diff --git a/dispatch_edge_test.go b/dispatch_edge_test.go index ae6d7c6..866eb8b 100644 --- a/dispatch_edge_test.go +++ b/dispatch_edge_test.go @@ -91,6 +91,7 @@ func TestDeferredDispatchAckPrecedesHandlerWorkWithRecordingAdapter(t *testing.T ThreadLockTTL: time.Hour, Concurrency: chat.ConcurrencyDrop, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, }), ) diff --git a/docs/adr/0012-concurrency-strategy.md b/docs/adr/0012-concurrency-strategy.md index 8466400..08de0c0 100644 --- a/docs/adr/0012-concurrency-strategy.md +++ b/docs/adr/0012-concurrency-strategy.md @@ -2,7 +2,13 @@ ## Status -Accepted (implementation staged: `drop`, `queue`, `debounce`, and `concurrent` strategies plus the `LockScope` option ship in the runtime, with `debounce` requiring `DispatchDeferred`; `burst` and force/steerability are staged behind the deferred-dispatch admission-bound and fenced-coordination design work — see the cross-instance coalescing and admission issues — with their names still reserved). +Accepted (implementation staged: `drop`, `queue`, `debounce`, and `concurrent` strategies plus the `LockScope` option ship in the runtime, with `debounce` requiring `DispatchDeferred`; the `burst` and force/steerability names remain reserved). + +Three statements of this ADR are superseded by ADR 0015 (deferred-dispatch admission bound; cross-instance coalescing rejected for now): + +- The proposed force surface — key-only `ForceReleaseLock` plus the `OnLockConflict` steerability hook — is rejected finally: key-only identity cannot distinguish victim from successor, and no force primitive ships in v0.x. Any future force/steerability design is gated on ADR 0015's formal-design bar. +- The staging gate that held `burst` on fenced-coordination design work is lifted: `burst` is gated only on the admission bound and its own revival decisions (the PR #53 outcome in ADR 0015). +- The expectation that `queue`/`debounce`/`burst` "need wait/coalesce coordination" expanding every State implementation is withdrawn for v0.x: per-instance supersession is the decided contract, needs no State expansion, and any future cross-instance coordination goes through ADR 0015's reopening bar. ## Context diff --git a/docs/adr/0015-runtime-coordination.md b/docs/adr/0015-runtime-coordination.md index a9c4c97..e29cd19 100644 --- a/docs/adr/0015-runtime-coordination.md +++ b/docs/adr/0015-runtime-coordination.md @@ -2,7 +2,7 @@ ## Status -Proposed. This decides the deferred-dispatch admission bound (issue #44) and explicitly rejects, for now, the cross-instance coalescing State extension (issue #50) after a full design attempt — see the rejection section for the evidence and the reopening bar. Issue #50 stays open as gated future work. This ADR also gives the staged burst/preemption branch (PR #53) its verdict. +Accepted (the admission bound is implemented; issue #44). This decides the deferred-dispatch admission bound (issue #44) and explicitly rejects, for now, the cross-instance coalescing State extension (issue #50) after a full design attempt — see the rejection section for the evidence and the reopening bar. Issue #50 stays open as gated future work. This ADR also gives the staged burst/preemption branch (PR #53) its verdict. This is a decision-level document: it fixes decisions, invariants, and non-goals. Implementation mechanics — slot bookkeeping, timer and shutdown lifecycles, counter management — are deliberately not specified here; they are decided in the implementing PRs, where code and hardening tests can actually verify them against the invariants below. diff --git a/docs/how-to/deferred-dispatch.md b/docs/how-to/deferred-dispatch.md index df95a51..fe41756 100644 --- a/docs/how-to/deferred-dispatch.md +++ b/docs/how-to/deferred-dispatch.md @@ -63,6 +63,19 @@ bot, err := chat.New(ctx, follow-up is cancelled without running (it was already deduped, so it will not be redelivered). Size `DetachTimeout` to cover your longest handler *plus* the queue wait behind it. +- `MaxDetached` (required under `DispatchDeferred`; `DefaultRuntimeOptions()` + sets 1024) is the admission bound from + [ADR 0015](../adr/0015-runtime-coordination.md): it caps + admitted-but-incomplete deferred deliveries — running handlers, queued and + debounced waiters, concurrent slot-waiters — so an event flood cannot grow + goroutines and retained payloads without limit. A delivery arriving at the + cap is rejected with `chat.ErrAdmissionRejected` **before** the ack and + **before** dedupe marking; the adapter maps that to a retry-inducing 503 for + platform-redelivered shapes (Slack Events API callbacks, Linear webhooks) and + a truthful busy signal for direct invocations (Slack slash commands and + interactivity). The optional `MaxDetachedPerTenant` additionally caps one + installation's share through the same rejection path. Sizing guidance lives + on the `MaxDetached` GoDoc. ## Write Handlers For The Detached Context diff --git a/errors.go b/errors.go index 8004952..472e54d 100644 --- a/errors.go +++ b/errors.go @@ -9,6 +9,16 @@ var ( // runtime instance, expired, or no longer refreshable), so mutual // exclusion could no longer be guaranteed. ErrPreempted = errors.New("chat: handler preempted") + // ErrAdmissionRejected is returned by deferred dispatch when the Admission + // Bound rejects a delivery: the instance is at MaxDetached, the delivery's + // (adapter, tenant) installation is at MaxDetachedPerTenant, or Runtime + // Shutdown has closed admission. The delivery was rejected before + // acknowledgement and before dedupe marking — it is never recorded in + // Event Identity, so a platform retry is not deduped away. Adapters map it + // to a shape-aware overload response: a retry-inducing status for + // platform-redelivered shapes, a truthful busy signal for direct + // invocations the platform does not redeliver (ADR 0015). + ErrAdmissionRejected = errors.New("chat: deferred dispatch admission rejected") ) func assert(ok bool, message string) { diff --git a/examples/linear-agent-hello-world/main.go b/examples/linear-agent-hello-world/main.go index f6c3272..1f1de60 100644 --- a/examples/linear-agent-hello-world/main.go +++ b/examples/linear-agent-hello-world/main.go @@ -41,6 +41,10 @@ func main() { ThreadLockTTL: 2 * time.Minute, Dispatch: chat.DispatchDeferred, DetachTimeout: 5 * time.Minute, + // Deferred dispatch requires an Admission Bound: MaxDetached caps + // admitted-but-incomplete deliveries so a webhook flood cannot grow + // goroutines/memory without limit (ADR 0015). + MaxDetached: 1024, }), ) if err != nil { diff --git a/export_test.go b/export_test.go new file mode 100644 index 0000000..0798539 --- /dev/null +++ b/export_test.go @@ -0,0 +1,12 @@ +package chat + +// AdmissionTenantEntries reports the number of live per-tenant admission +// counter entries, for hardening tests proving a tenant whose deliveries all +// released capacity retains no counter entry (ADR 0015 bounded retention). +func AdmissionTenantEntries(c *Chat) int { + assert(c != nil, "AdmissionTenantEntries called on nil runtime") + if c.admission == nil { + return 0 + } + return c.admission.tenantEntries() +} diff --git a/lockscope_test.go b/lockscope_test.go index abdafb2..85b7300 100644 --- a/lockscope_test.go +++ b/lockscope_test.go @@ -40,6 +40,7 @@ func newLockScopeRuntime(t *testing.T, state chat.State, adapter chat.Adapter, l Concurrency: chat.ConcurrencyDrop, LockScope: chat.LockScopeChannel, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, } for _, m := range mutate { diff --git a/observer.go b/observer.go index ea14d0f..a1a80b3 100644 --- a/observer.go +++ b/observer.go @@ -32,6 +32,10 @@ const ( // observes platform rate limiting (ADR 0005). Adapter-owned, like // ObsAdapterCall. ObsRateLimit ObservationName = "rate_limit" + // ObsAdmissionRejected is emitted when the deferred-dispatch Admission + // Bound rejects a delivery (MaxDetached, MaxDetachedPerTenant, or Runtime + // Shutdown). Carries adapter and tenant attributes. + ObsAdmissionRejected ObservationName = "admission_rejected" ) // DispatchOutcome is the closed set of terminal outcomes for one Runtime @@ -48,6 +52,11 @@ const ( // its Lock Lease was lost mid-run: released by another runtime instance, // expired, or no longer refreshable. OutcomePreempted DispatchOutcome = "preempted" + // OutcomeAdmissionRejected is the terminal outcome of a delivery the + // Admission Bound rejected before acknowledgement and before dedupe + // marking: it never became an Accepted Event and owes only the overload + // response. + OutcomeAdmissionRejected DispatchOutcome = "admission-rejected" ) // Attribute key constants form the documented, stable, low-cardinality set. diff --git a/observer_hardening_test.go b/observer_hardening_test.go index f802b27..111f309 100644 --- a/observer_hardening_test.go +++ b/observer_hardening_test.go @@ -149,6 +149,7 @@ func TestObserverDeferredCommandSpanFollowsDetachedTail(t *testing.T) { ThreadLockTTL: time.Hour, Concurrency: chat.ConcurrencyDrop, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, }), ) diff --git a/queue_edge_test.go b/queue_edge_test.go index 10d74ca..d5f1fac 100644 --- a/queue_edge_test.go +++ b/queue_edge_test.go @@ -19,6 +19,7 @@ func newQueueRuntime(t *testing.T, state chat.State, adapter chat.Adapter, logs ThreadLockTTL: 40 * time.Millisecond, Concurrency: chat.ConcurrencyQueue, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, } for _, m := range mutate { @@ -222,6 +223,7 @@ func TestQueueDropRemainsDefaultRegression(t *testing.T) { DedupeTTL: time.Hour, ThreadLockTTL: time.Hour, Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, DetachTimeout: 5 * time.Second, }), ) diff --git a/runtime.go b/runtime.go index 57516d5..53401fe 100644 --- a/runtime.go +++ b/runtime.go @@ -37,8 +37,9 @@ const ( // 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. + // event, serialized by the Thread Lock). Per-instance supersession is the + // decided v0.x contract (ADR 0015); cross-instance coalescing is rejected + // for now behind that ADR's reopening bar. ConcurrencyDebounce // ConcurrencyConcurrent is the explicit opt-out of per-scope serialization: // every routed event dispatches immediately in its own execution, bounded by @@ -47,9 +48,10 @@ const ( ConcurrencyConcurrent ) -// The burst strategy and the force/steerability (OnLockConflict) hook from ADR -// 0012 are staged on a separate branch pending the deferred-dispatch admission -// and fenced-coordination design work; their names remain reserved. +// The burst strategy and force/steerability names from ADR 0012 remain +// reserved. Per ADR 0015, burst revives as its own PR under the Admission +// Bound's invariants, while force/steerability is rejected pending that ADR's +// formal-design bar. // LockScope selects what key the Thread Lock guards. The opaque Thread ID is // unchanged; the scope only chooses the serialization key. @@ -94,6 +96,31 @@ type RuntimeOptions struct { // ConcurrencyConcurrent. It must be positive under that strategy and is // ignored otherwise. MaxConcurrent int + // MaxDetached is the deferred-dispatch Admission Bound (ADR 0015): a + // per-instance cap on admitted-but-incomplete deferred deliveries. + // Everything a delivery retains under DispatchDeferred counts against it — + // running detached tails, parked queue/debounce waiters, and concurrent + // slot-waiters — and capacity frees only when that retention ends. A + // delivery arriving at the cap is rejected with ErrAdmissionRejected + // before acknowledgement and before dedupe marking, so a platform retry is + // never deduped away. It must be positive under DispatchDeferred and is + // ignored under DispatchSync; DefaultRuntimeOptions sets 1024. + // + // Sizing: the bound is a count, not bytes — the runtime cannot measure + // retained platform payloads or handler closures. Budget roughly + // MaxDetached x (platform payload ceiling + one goroutine stack + whatever + // the handler closure pins) against the instance's memory; sustained + // rejection is the platform-visible backpressure signal, so size the cap + // to be reached only under genuine overload and keep front-door rate + // limiting for fleet-level control. + MaxDetached int + // MaxDetachedPerTenant additionally caps any single installation's share + // of MaxDetached, keyed on the delivery's (adapter, tenant) installation + // identity (ADR 0006). It is a ceiling through the same rejection path, + // not a reservation: a hot tenant is capped, but no capacity is guaranteed + // to the others. Zero disables the ceiling; it must not be negative under + // DispatchDeferred and is ignored under DispatchSync. + MaxDetachedPerTenant int } func DefaultRuntimeOptions() RuntimeOptions { @@ -104,6 +131,7 @@ func DefaultRuntimeOptions() RuntimeOptions { Dispatch: DispatchSync, DetachTimeout: 0, LockScope: LockScopeThread, + MaxDetached: 1024, } } @@ -174,6 +202,10 @@ type Chat struct { // ConcurrencyConcurrent; nil under every other strategy. concurrencySlots chan struct{} + // admission is the deferred-dispatch Admission Bound (ADR 0015); nil under + // DispatchSync. + admission *admissionGate + shutdownMu sync.Mutex shutdown bool shutdownDone chan struct{} @@ -227,6 +259,9 @@ func New(ctx context.Context, opts ...Option) (*Chat, error) { if cfg.options.Concurrency == ConcurrencyConcurrent { chat.concurrencySlots = make(chan struct{}, cfg.options.MaxConcurrent) } + if cfg.options.Dispatch == DispatchDeferred { + chat.admission = newAdmissionGate(cfg.options.MaxDetached, cfg.options.MaxDetachedPerTenant) + } for _, adapter := range cfg.adapters { if adapter == nil { baseCancel() @@ -293,6 +328,12 @@ func validateRuntimeOptions(options RuntimeOptions) error { if options.DetachTimeout <= 0 { return errors.New("chat: detach timeout must be positive under deferred dispatch") } + if options.MaxDetached <= 0 { + return errors.New("chat: max detached must be positive under deferred dispatch") + } + if options.MaxDetachedPerTenant < 0 { + return errors.New("chat: max detached per tenant must not be negative under deferred dispatch") + } default: return errors.New("chat: unsupported dispatch mode") } @@ -398,8 +439,13 @@ func (c *Chat) Shutdown(ctx context.Context) error { c.shutdownMu.Unlock() defer close(done) - // Cancel detached tails, then drain (bounded by ctx) before shutting down - // adapters and state. + // Close admission first so a delivery racing Shutdown is rejected with + // ErrAdmissionRejected (the platform's retry covers it) instead of being + // admitted into a runtime about to cancel its work. Then cancel detached + // tails and drain (bounded by ctx) before shutting down adapters and state. + if c.admission != nil { + c.admission.close() + } c.baseCancel() var drainErr error drained := make(chan struct{}) @@ -482,12 +528,35 @@ func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64) error // dispatchDeferred runs the prelude under the request context and, on a routed // event, launches the detached tail and returns so the adapter can acknowledge -// the platform (ack-then-work). +// the platform (ack-then-work). The Admission Bound gate runs first, before +// any acknowledgement and before dedupe marking: a delivery rejected at the +// cap fails fast with ErrAdmissionRejected, is never marked in Event Identity +// (so a platform retry is not deduped away), and does no State work at all — +// under saturation even a redelivered duplicate receives the overload +// response, converging to the ordinary duplicate acknowledgement once +// capacity frees (ADR 0015). func (c *Chat) dispatchDeferred(ctx context.Context, event *Event, seq uint64) error { + if err := validateEvent(event); err != nil { + return err + } + assert(c.admission != nil, "deferred dispatch requires the admission gate") + release, admitted := c.admission.admit(event.Adapter, event.Tenant) + if !admitted { + spanCtx, span := c.safeDispatch(ctx, AdapterAttr(event.Adapter), TenantAttr(event.Tenant)) + c.logger.Warn("chat deferred dispatch admission rejected", "adapter", event.Adapter, "tenant", event.Tenant, "event_id", event.ID) + c.safeEvent(spanCtx, ObsAdmissionRejected, AdapterAttr(event.Adapter), TenantAttr(event.Tenant)) + c.safeEnd(span, OutcomeAdmissionRejected) + return fmt.Errorf("chat: deferred dispatch admission rejected for adapter %q: %w", event.Adapter, ErrAdmissionRejected) + } work, resolved, err := c.prelude(ctx, event, seq) if err != nil || resolved { + // The delivery retains nothing past the prelude: an errored or + // resolved (duplicate, dropped, ignored, unrouted) prelude frees its + // admission slot immediately. + release() return err } + work.releaseAdmission = release c.startDetachedTail(work) return nil } @@ -520,6 +589,11 @@ type preludeWork struct { // noLock is true under the concurrent strategy: no Thread Lock is taken and // the run is bounded by a MaxConcurrent slot instead. noLock bool + // releaseAdmission frees the delivery's Admission Bound slot. It is set + // only under DispatchDeferred and runs when the detached tail goroutine + // returns — not when the handler returns — so stalled cleanup (lock + // release, lease refresh drain) still counts as retention. + releaseAdmission func() // span is opened in the prelude and closed by the tail, so deferred dispatch // measures Ack-Then-Work latency to handler completion. span DispatchSpan @@ -807,9 +881,14 @@ func (c *Chat) routedWork( // superseded or abandoned waiter exits without running the handler. Concurrent // tails wait for a MaxConcurrent slot rather than a lock. func (c *Chat) startDetachedTail(work preludeWork) { + assert(work.releaseAdmission != nil, "detached tail requires a held admission slot") tailCtx, tailCancel := context.WithTimeout(c.baseCtx, c.options.DetachTimeout) c.inflight.Add(1) go func() { + // The admission slot is held until this goroutine returns: everything + // the tail retains — the parked wait, the handler run, and cleanup + // (lock release, refresh-loop drain) — counts against MaxDetached. + defer work.releaseAdmission() defer c.inflight.Done() defer tailCancel() From daa7113e689c9e390f8c5b0bfb5fcfaf84b5ed00 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 27 Aug 2026 21:51:32 +0000 Subject: [PATCH 2/2] fix(runtime): drain prelude-held admissions on shutdown (codex P1) --- admission.go | 23 +++++++++- admission_test.go | 110 ++++++++++++++++++++++++++++++++++++++++++++++ runtime.go | 13 +++++- 3 files changed, 143 insertions(+), 3 deletions(-) diff --git a/admission.go b/admission.go index e85e461..bc8216f 100644 --- a/admission.go +++ b/admission.go @@ -27,6 +27,11 @@ type admissionGate struct { // tenant's count reaches zero so an idle tenant retains no map entry. tenants map[tenantKey]int closed bool + // drainDone is created by close and closed once every held slot has been + // released, so Runtime Shutdown can wait for admitted deliveries that are + // still in their prelude (not yet counted by the tail WaitGroup) as well + // as running tails. + drainDone chan struct{} } func newAdmissionGate(capacity, perTenant int) *admissionGate { @@ -67,6 +72,9 @@ func (g *admissionGate) release(key tenantKey) { defer g.mu.Unlock() assert(g.inUse > 0, "admission release without a held slot") g.inUse-- + if g.closed && g.inUse == 0 && g.drainDone != nil { + close(g.drainDone) + } if g.perTenant > 0 { count := g.tenants[key] assert(count > 0, "admission release without a held tenant slot") @@ -79,11 +87,22 @@ func (g *admissionGate) release(key tenantKey) { } // close permanently rejects new admissions. Held slots are unaffected: their -// releases still run so accounting stays exact while in-flight tails drain. -func (g *admissionGate) close() { +// releases still run so accounting stays exact while in-flight work drains. +// The returned channel is closed once every held slot has been released — +// including slots held by deliveries still in their synchronous prelude, which +// the tail WaitGroup cannot see — so Runtime Shutdown does not tear down +// adapters and state under an admitted delivery that won the shutdown race. +func (g *admissionGate) close() <-chan struct{} { g.mu.Lock() defer g.mu.Unlock() g.closed = true + if g.drainDone == nil { + g.drainDone = make(chan struct{}) + if g.inUse == 0 { + close(g.drainDone) + } + } + return g.drainDone } // tenantEntries reports the number of live per-tenant counter entries, for diff --git a/admission_test.go b/admission_test.go index 413af8b..74d1846 100644 --- a/admission_test.go +++ b/admission_test.go @@ -316,6 +316,116 @@ func TestAdmissionClosedOnShutdown(t *testing.T) { } } +// shutdownRaceState parks the first MarkEvent so a test can hold a delivery +// mid-prelude (admission slot held, no tail spawned yet) across a Shutdown +// call, and records whether state shutdown began before that prelude finished. +type shutdownRaceState struct { + *fakeState + markStarted chan struct{} + markGate chan struct{} + markOnce sync.Once + + raceMu sync.Mutex + markDone bool + shutdownBeforeMarkDone bool +} + +func newShutdownRaceState() *shutdownRaceState { + return &shutdownRaceState{ + fakeState: newFakeState(), + markStarted: make(chan struct{}), + markGate: make(chan struct{}), + } +} + +func (s *shutdownRaceState) MarkEvent(ctx context.Context, id string, ttl time.Duration) (bool, error) { + blocked := false + s.markOnce.Do(func() { blocked = true }) + if blocked { + close(s.markStarted) + <-s.markGate + } + first, err := s.fakeState.MarkEvent(ctx, id, ttl) + s.raceMu.Lock() + s.markDone = true + s.raceMu.Unlock() + return first, err +} + +func (s *shutdownRaceState) Shutdown(ctx context.Context) error { + s.raceMu.Lock() + if !s.markDone { + s.shutdownBeforeMarkDone = true + } + s.raceMu.Unlock() + return s.fakeState.Shutdown(ctx) +} + +// TestShutdownDrainsPreludeHeldAdmissions pins the admission/shutdown race: a +// delivery that passed the admission gate but is still in its synchronous +// prelude holds no tail WaitGroup count, so the shutdown drain must wait on +// admission slots — never tearing down adapters and state under an admitted +// delivery, and never returning before that delivery's retention ends. +func TestShutdownDrainsPreludeHeldAdmissions(t *testing.T) { + t.Parallel() + + state := newShutdownRaceState() + adapter := newFakeAdapter("fake") + observer := &recordingObserver{} + bot := newAdmissionRuntime(t, state, adapter, observer, func(o *chat.RuntimeOptions) { + o.MaxDetached = 4 + }) + bot.OnNewMention(func(ctx context.Context, _ *chat.MessageEvent) error { + return ctx.Err() + }) + + dispatched := make(chan postEventResult, 1) + go func() { + dispatched <- postEventResultFor(bot, "fake", mentionEvent("event-1", "fake:v1:thread-1")) + }() + select { + case <-state.markStarted: + case <-time.After(5 * time.Second): + t.Fatal("delivery did not reach dedupe marking") + } + + shutdownDone := make(chan error, 1) + go func() { shutdownDone <- bot.Shutdown(context.Background()) }() + + // The admitted delivery is parked in its prelude: Shutdown must wait for + // it, not observe zero tails and proceed. + select { + case err := <-shutdownDone: + t.Fatalf("shutdown completed under a prelude-held admission (err = %v)", err) + case <-time.After(150 * time.Millisecond): + } + + close(state.markGate) + select { + case err := <-shutdownDone: + if err != nil { + t.Fatalf("shutdown: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("shutdown did not complete after the prelude finished") + } + if state.shutdownBeforeMarkDone { + t.Fatal("state was shut down before the admitted delivery's prelude finished") + } + result := <-dispatched + if result.err != nil { + t.Fatalf("dispatch: %v", result.err) + } + if result.status != http.StatusOK { + t.Fatalf("admitted delivery status = %d", result.status) + } + // No silent loss: the admitted delivery reached an observable terminal + // outcome even though shutdown cancelled its detached work. + if len(observer.terminalOutcomes()) == 0 { + t.Fatal("admitted delivery left no terminal dispatch outcome") + } +} + // TestAdmissionOptionValidation pins constructor validation: MaxDetached must // be positive and MaxDetachedPerTenant non-negative under DispatchDeferred, // while DispatchSync ignores both (the DetachTimeout precedent). diff --git a/runtime.go b/runtime.go index 53401fe..9989276 100644 --- a/runtime.go +++ b/runtime.go @@ -443,13 +443,24 @@ func (c *Chat) Shutdown(ctx context.Context) error { // ErrAdmissionRejected (the platform's retry covers it) instead of being // admitted into a runtime about to cancel its work. Then cancel detached // tails and drain (bounded by ctx) before shutting down adapters and state. + // + // The drain waits on admission slots before the tail WaitGroup: a delivery + // that won the admission race is retained from admit until its tail + // goroutine returns, so waiting for every slot covers deliveries still in + // their synchronous prelude — which hold no WaitGroup count yet — and + // guarantees no tail is spawned (and no WaitGroup Add happens) after the + // slot drain completes. + var admissionDrained <-chan struct{} if c.admission != nil { - c.admission.close() + admissionDrained = c.admission.close() } c.baseCancel() var drainErr error drained := make(chan struct{}) go func() { + if admissionDrained != nil { + <-admissionDrained + } c.inflight.Wait() close(drained) }()