From fc24c27dfd660fa05c601b7fb571ac6b480ea0e2 Mon Sep 17 00:00:00 2001 From: Michael Suchacz <203725896+ibetitsmike@users.noreply.github.com> Date: Mon, 3 Aug 2026 19:04:54 +0200 Subject: [PATCH] fix: reserve chat hook dispatch capacity for running turns (#27656) ## Context Follow-up fix from live UAT of the merged chat lifecycle hooks stack (#27430). Its companion UAT fix (#27655) has merged, so this targets `main` directly. ## Why? UAT measured a burst of 1,500 concurrent chat creations against a consumer with 1.2s latency. 255 were admitted and 1,245 got `502 hook_dispatch_failed (over_capacity)`, which is correct fail-closed behavior. The collateral wasn't: the same burst failed 24 `stop` dispatches, parking chats that had already been admitted and had already executed tools. One 256-slot semaphore served every event, so new-work admission could take every slot and kill turns in flight. Callers now classify each dispatch as admission or generation, and admission draws from a 192-slot gate held *before* the shared pool. At least 64 shared slots stay reachable only by dispatches for work a chat already admitted. The dispatcher is per `coderd` replica, so these limits are per replica, not deployment-wide, and the docs say so. **The caller classifies, not the event type.** Event type isn't a reliable proxy in either direction: a subagent spawn dispatches `user_prompt_submit` from inside a running turn, and the edit path dispatches `session_start` at admission time. `CapacityClassUnset` is rejected in `Dispatch`, so a new call site fails closed rather than silently inheriting a share. **Acquisition order is load-bearing.** Admission takes its own gate first. Taking a shared slot first would let admissions queued on the gate occupy the very capacity the reserve protects. `acquireCapacity` is the only path that takes either pool, so the order can't be bypassed. ## What this does not guarantee Nothing bounds how many turns generate concurrently, so the 192/64 split is a judgement call, not a derived ceiling. This stops an *admission* burst from consuming every slot; it does not make the remainder sufficient. A large enough generation load can still exhaust the reserve and error a running chat. The docs say so explicitly rather than promising a guarantee the code doesn't deliver. Generation can now take all 256 slots, so generation traffic starves admission harder than before. That's the intended priority: rejecting a new prompt is recoverable, ending a turn that already ran tools is not. ## Testing Red-green proved both new tests. Removing the release-on-failure path fails `RefusedSharedAcquireReleasesAdmission` deterministically; removing the expired-deadline check fails `ExpiredDeadlineRefusesFreeSlot` in 18/30 runs. That deadline check fixes a real race found in review. `acquire` previously shared one `time.Timer` across both acquires. Because `select` picks a ready case at random, an admission dispatch could take a slot after its capacity deadline had passed. Measured over 300 trials: 135 late acquisitions, worst overshoot 2.1ms. `acquire` now takes an absolute deadline and refuses an expired one before selecting, which measures 0/300. Go: `coderd/x/agenthooks/...` and `coderd/x/chatd/...`, plus `-race -count=3` on the dispatcher. > Mux opened this PR on Mike's behalf. --- coderd/x/agenthooks/dispatch/dispatcher.go | 101 ++++++++++++---- .../dispatch/dispatcher_internal_test.go | 111 +++++++++++++++++- coderd/x/chatd/ARCHITECTURE.md | 2 + coderd/x/chatd/chatd.go | 10 +- .../x/chatd/chathooks/hooks_internal_test.go | 12 +- coderd/x/chatd/chathooks/tooluse.go | 5 +- coderd/x/chatd/chathooks/trigger.go | 8 +- coderd/x/chatd/generation.go | 9 +- coderd/x/chatd/subagent.go | 2 +- docs/admin/setup/chat-lifecycle-hooks.md | 7 ++ 10 files changed, 224 insertions(+), 43 deletions(-) diff --git a/coderd/x/agenthooks/dispatch/dispatcher.go b/coderd/x/agenthooks/dispatch/dispatcher.go index db7e1566e1..b0c58e0ac5 100644 --- a/coderd/x/agenthooks/dispatch/dispatcher.go +++ b/coderd/x/agenthooks/dispatch/dispatcher.go @@ -28,11 +28,22 @@ import ( const ( maxConcurrentDispatches = 256 - maxResponseBodyBytes = 1_048_576 - maxModelContextBytes = 16_384 - capacityWaitLimit = 250 * time.Millisecond - retryBackoff = 250 * time.Millisecond - clockSkewLeeway = 30 * time.Second + // Keep capacity reachable by work that a chat already admitted. + maxAdmissionDispatches = 192 + maxResponseBodyBytes = 1_048_576 + maxModelContextBytes = 16_384 + capacityWaitLimit = 250 * time.Millisecond + retryBackoff = 250 * time.Millisecond + clockSkewLeeway = 30 * time.Second +) + +// CapacityClass selects which share of dispatch capacity an event draws from. +type CapacityClass int + +const ( + CapacityClassUnset CapacityClass = iota + CapacityClassAdmission + CapacityClassGeneration ) // Result classifies the terminal outcome of a dispatch attempt. @@ -53,7 +64,8 @@ const ( type Event struct { Type agenthooks.EventType agenthooks.ChatRef - Data any + Data any + Capacity CapacityClass } // Error preserves the attempt ID and failure class. @@ -90,6 +102,7 @@ type Dispatcher struct { deploymentID string userAgent string semaphore chan struct{} + admission chan struct{} metrics *metrics } @@ -167,6 +180,7 @@ func New( deploymentID: deploymentID, userAgent: "coderd-agenthooks/" + coderVersion, semaphore: make(chan struct{}, maxConcurrentDispatches), + admission: make(chan struct{}, maxAdmissionDispatches), metrics: newMetrics(reg), } } @@ -187,25 +201,21 @@ func (d *Dispatcher) Dispatch(ctx context.Context, event Event) (agenthooks.Resp return agenthooks.Response{}, uuid.Nil, xerrors.Errorf("chat hook URL rejected: %w", d.hookURLErr) } + switch event.Capacity { + case CapacityClassAdmission, CapacityClassGeneration: + default: + return agenthooks.Response{}, uuid.Nil, xerrors.Errorf("dispatch event %q has no capacity class", event.Type) + } + startedAt := time.Now() dispatchID := uuid.New() - wait := max(min(d.timeout, capacityWaitLimit), 0) - capacityTimer := time.NewTimer(wait) - defer capacityTimer.Stop() + capacityDeadline := startedAt.Add(max(min(d.timeout, capacityWaitLimit), 0)) - // The capacity wait runs against its own timer rather than a dispatch - // deadline, so a timeout shorter than capacityWaitLimit cannot make the - // over-capacity and caller-cancellation cases race. - select { - case d.semaphore <- struct{}{}: - defer func() { <-d.semaphore }() - case <-ctx.Done(): - outcome := dispatchOutcome{result: ResultTimeout, err: ctx.Err()} - return agenthooks.Response{}, dispatchID, d.finish(ctx, event, dispatchID, startedAt, outcome) - case <-capacityTimer.C: - outcome := dispatchOutcome{result: ResultOverCapacity, err: context.DeadlineExceeded} - return agenthooks.Response{}, dispatchID, d.finish(ctx, event, dispatchID, startedAt, outcome) + release, refused, ok := d.acquireCapacity(ctx, event.Capacity, capacityDeadline) + if !ok { + return agenthooks.Response{}, dispatchID, d.finish(ctx, event, dispatchID, startedAt, refused) } + defer release() // Both post attempts share whatever remains of the configured timeout so // that waiting for capacity cannot extend the dispatch past it. @@ -219,6 +229,55 @@ func (d *Dispatcher) Dispatch(ctx context.Context, event Event) (agenthooks.Resp return outcome.response, dispatchID, nil } +// Admission acquires its gate before the shared pool so queued admission +// dispatches cannot occupy the reserved capacity. +func (d *Dispatcher) acquireCapacity( + ctx context.Context, + capacity CapacityClass, + deadline time.Time, +) (release func(), outcome dispatchOutcome, ok bool) { + if capacity == CapacityClassAdmission { + releaseAdmission, refused, admitted := acquire(ctx, d.admission, deadline) + if !admitted { + return nil, refused, false + } + releaseShared, refused, acquired := acquire(ctx, d.semaphore, deadline) + if !acquired { + releaseAdmission() + return nil, refused, false + } + return func() { + releaseShared() + releaseAdmission() + }, dispatchOutcome{}, true + } + return acquire(ctx, d.semaphore, deadline) +} + +// Check the deadline before select because select randomly chooses among ready +// cases, including a free slot and an expired timer. +func acquire( + ctx context.Context, + pool chan struct{}, + deadline time.Time, +) (release func(), outcome dispatchOutcome, ok bool) { + overCapacity := dispatchOutcome{result: ResultOverCapacity, err: context.DeadlineExceeded} + remaining := time.Until(deadline) + if remaining <= 0 { + return nil, overCapacity, false + } + timer := time.NewTimer(remaining) + defer timer.Stop() + select { + case pool <- struct{}{}: + return func() { <-pool }, dispatchOutcome{}, true + case <-ctx.Done(): + return nil, dispatchOutcome{result: ResultTimeout, err: ctx.Err()}, false + case <-timer.C: + return nil, overCapacity, false + } +} + func (d *Dispatcher) finish( ctx context.Context, event Event, diff --git a/coderd/x/agenthooks/dispatch/dispatcher_internal_test.go b/coderd/x/agenthooks/dispatch/dispatcher_internal_test.go index 320fd33b14..04915f7e40 100644 --- a/coderd/x/agenthooks/dispatch/dispatcher_internal_test.go +++ b/coderd/x/agenthooks/dispatch/dispatcher_internal_test.go @@ -671,7 +671,8 @@ func newTestEvent(t *testing.T, eventType agenthooks.EventType, data any) Event ChatID: uuid.New(), OwnerID: uuid.New(), }, - Data: data, + Data: data, + Capacity: CapacityClassGeneration, } } @@ -699,3 +700,111 @@ type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) } + +func TestDispatcherCapacityClassRequired(t *testing.T) { + t.Parallel() + + event := newTestEvent(t, agenthooks.EventStop, agenthooks.StopData{}) + event.Capacity = CapacityClassUnset + dispatcher := newTestDispatcher(t, nil, "https://unused.test", time.Second) + + _, dispatchID, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitShort), event) + require.ErrorContains(t, err, "no capacity class") + require.Equal(t, uuid.Nil, dispatchID) +} + +func TestDispatcherAdmissionReserve(t *testing.T) { + t.Parallel() + + fill := func(t *testing.T, pool chan struct{}, count int) { + t.Helper() + for range count { + pool <- struct{}{} + } + t.Cleanup(func() { + for range count { + <-pool + } + }) + } + + t.Run("AdmissionReleasesBothPools", func(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, err := w.Write([]byte(`{}`)) + assert.NoError(t, err) + })) + t.Cleanup(server.Close) + + dispatcher := New( + testutil.Logger(t), server.Client(), server.URL, testSecret, testutil.WaitShort, + testDeploymentID, testVersion, prometheus.NewRegistry(), + ) + event := newTestEvent(t, agenthooks.EventUserPromptSubmit, agenthooks.UserPromptSubmitData{Prompt: "hi"}) + event.Capacity = CapacityClassAdmission + + _, _, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitLong), event) + require.NoError(t, err) + require.Empty(t, dispatcher.admission) + require.Empty(t, dispatcher.semaphore) + }) + + t.Run("SaturatedAdmissionRefusesAdmission", func(t *testing.T) { + t.Parallel() + + dispatcher := newTestDispatcher(t, nil, "https://unused.test", 10*time.Millisecond) + fill(t, dispatcher.admission, maxAdmissionDispatches) + + event := newTestEvent(t, agenthooks.EventUserPromptSubmit, agenthooks.UserPromptSubmitData{Prompt: "hi"}) + event.Capacity = CapacityClassAdmission + _, _, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitLong), event) + assertDispatchErrorClass(t, err, ResultOverCapacity) + require.Empty(t, dispatcher.semaphore, "a refused admission must not hold a shared slot") + }) + + t.Run("ExpiredDeadlineRefusesFreeSlot", func(t *testing.T) { + t.Parallel() + + pool := make(chan struct{}, 1) + release, outcome, ok := acquire(testutil.Context(t, testutil.WaitShort), pool, time.Now().Add(-time.Millisecond)) + require.False(t, ok) + require.Nil(t, release) + require.Equal(t, ResultOverCapacity, outcome.result) + require.Empty(t, pool) + }) + + t.Run("RefusedSharedAcquireReleasesAdmission", func(t *testing.T) { + t.Parallel() + + dispatcher := newTestDispatcher(t, nil, "https://unused.test", 10*time.Millisecond) + fill(t, dispatcher.semaphore, maxConcurrentDispatches) + + event := newTestEvent(t, agenthooks.EventUserPromptSubmit, agenthooks.UserPromptSubmitData{Prompt: "hi"}) + event.Capacity = CapacityClassAdmission + _, _, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitLong), event) + assertDispatchErrorClass(t, err, ResultOverCapacity) + require.Empty(t, dispatcher.admission, "an admission refused by the shared pool must release its gate token") + }) + + t.Run("SaturatedAdmissionStillServesGeneration", func(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, err := w.Write([]byte(`{}`)) + assert.NoError(t, err) + })) + t.Cleanup(server.Close) + + dispatcher := New( + testutil.Logger(t), server.Client(), server.URL, testSecret, testutil.WaitShort, + testDeploymentID, testVersion, prometheus.NewRegistry(), + ) + fill(t, dispatcher.admission, maxAdmissionDispatches) + fill(t, dispatcher.semaphore, maxAdmissionDispatches) + + event := newTestEvent(t, agenthooks.EventStop, agenthooks.StopData{}) + _, _, err := dispatcher.Dispatch(testutil.Context(t, testutil.WaitLong), event) + require.NoError(t, err) + }) +} diff --git a/coderd/x/chatd/ARCHITECTURE.md b/coderd/x/chatd/ARCHITECTURE.md index f2bfab782a..26c553af8c 100644 --- a/coderd/x/chatd/ARCHITECTURE.md +++ b/coderd/x/chatd/ARCHITECTURE.md @@ -915,6 +915,8 @@ The consumer can observe activity, add model-only or user-visible context, repla Lifecycle hooks fail closed. If the consumer cannot be reached or returns an invalid response, Coder stops the triggering operation rather than continuing without the consumer's decision. Affected chats can enter an error state until the consumer recovers or hooks are disabled. +Concurrent dispatches are capped per replica, and each dispatch declares whether it admits new work into a chat or belongs to work a chat already admitted. Admission can hold only part of the cap, so a burst of new submissions cannot consume the capacity that already-admitted work depends on. The caller declares this, because the event type does not determine it: a subagent spawn submits a prompt from inside a running turn, and editing a message starts a session at admission time. + Coder stores no hook-specific dispatch or decision state. Delivery is best-effort and can duplicate, and a failed dispatch is never redelivered, so the consumer owns durable policy state, audit records, and deduplication based on stable event identifiers. # Stream loop diff --git a/coderd/x/chatd/chatd.go b/coderd/x/chatd/chatd.go index ccc12ad80c..3cfb17a4bf 100644 --- a/coderd/x/chatd/chatd.go +++ b/coderd/x/chatd/chatd.go @@ -1334,7 +1334,7 @@ func (p *Server) CreateChat(ctx context.Context, opts CreateOptions) (database.C OwnerID: opts.OwnerID, WorkspaceID: opts.WorkspaceID, TurnID: &turnID, - }, promptMessage, agenthooks.EventUserPromptSubmit) + }, promptMessage, agenthooks.EventUserPromptSubmit, dispatch.CapacityClassAdmission) if err != nil { return database.Chat{}, chathooks.UserPromptDenial(err) } @@ -1487,7 +1487,7 @@ func (p *Server) SendMessage( if err != nil { return SendMessageResult{}, err } - promptResult, err := p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), promptMessage, agenthooks.EventUserPromptSubmit) + promptResult, err := p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), promptMessage, agenthooks.EventUserPromptSubmit, dispatch.CapacityClassAdmission) if err != nil { return SendMessageResult{}, p.handleUserPromptDispatchError(ctx, opts.ChatID, chathooks.UserPromptDenial(err)) } @@ -1791,7 +1791,7 @@ func (p *Server) EditMessage( if _, err := validateModelConfigOverride(ctx, p.db, opts.ModelConfigID); err != nil { return EditMessageResult{}, err } - sessionStartHookResult, err = p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), chathooks.Message{Source: chathooks.SessionStartSourceClear}, agenthooks.EventSessionStart) + sessionStartHookResult, err = p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), chathooks.Message{Source: chathooks.SessionStartSourceClear}, agenthooks.EventSessionStart, dispatch.CapacityClassAdmission) if err != nil { return EditMessageResult{}, p.handleAPIDispatchError(ctx, opts.ChatID, agenthooks.EventSessionStart, err) } @@ -1799,7 +1799,7 @@ func (p *Server) EditMessage( if err != nil { return EditMessageResult{}, err } - promptResult, err := p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), promptMessage, agenthooks.EventUserPromptSubmit) + promptResult, err := p.hooks.Trigger(ctx, chathooks.ChatFor(chat, &turnID), promptMessage, agenthooks.EventUserPromptSubmit, dispatch.CapacityClassAdmission) if err != nil { return EditMessageResult{}, p.handleUserPromptDispatchError(ctx, opts.ChatID, chathooks.UserPromptDenial(err)) } @@ -2170,7 +2170,7 @@ func (p *Server) SubmitToolResults( return err } for _, result := range opts.Results { - response, err := p.hooks.Trigger(ctx, chathooks.ChatFor(state.chat, nil), chathooks.DynamicPostToolUseMessage(result, state.toolNames[result.ToolCallID]), agenthooks.EventPostToolUse) + response, err := p.hooks.Trigger(ctx, chathooks.ChatFor(state.chat, nil), chathooks.DynamicPostToolUseMessage(result, state.toolNames[result.ToolCallID]), agenthooks.EventPostToolUse, dispatch.CapacityClassGeneration) if err != nil { // Leave pending calls intact so the client can resubmit after recovery. return chathooks.GenerationDispatchError(agenthooks.EventPostToolUse, err) diff --git a/coderd/x/chatd/chathooks/hooks_internal_test.go b/coderd/x/chatd/chathooks/hooks_internal_test.go index 58a8dab9ad..f30b475129 100644 --- a/coderd/x/chatd/chathooks/hooks_internal_test.go +++ b/coderd/x/chatd/chathooks/hooks_internal_test.go @@ -65,9 +65,9 @@ func TestSessionStartDispatchSources(t *testing.T) { turnID := uuid.New() ctx := testutil.Context(t, testutil.WaitLong) - _, err := trigger.Trigger(ctx, ChatFor(chat, &turnID), Message{Source: SessionStartSource(nil)}, agenthooks.EventSessionStart) + _, err := trigger.Trigger(ctx, ChatFor(chat, &turnID), Message{Source: SessionStartSource(nil)}, agenthooks.EventSessionStart, dispatch.CapacityClassGeneration) require.NoError(t, err) - _, err = trigger.Trigger(ctx, ChatFor(chat, &turnID), Message{Source: SessionStartSource([]database.ChatMessage{{Role: database.ChatMessageRoleAssistant}})}, agenthooks.EventSessionStart) + _, err = trigger.Trigger(ctx, ChatFor(chat, &turnID), Message{Source: SessionStartSource([]database.ChatMessage{{Role: database.ChatMessageRoleAssistant}})}, agenthooks.EventSessionStart, dispatch.CapacityClassGeneration) require.NoError(t, err) startup := <-receivedCh @@ -120,7 +120,7 @@ func TestHookTriggerDisabled(t *testing.T) { t.Run(name, func(t *testing.T) { t.Parallel() require.False(t, trigger.Enabled()) - result, err := trigger.Trigger(t.Context(), Chat{ID: uuid.New()}, Message{}, agenthooks.EventStop) + result, err := trigger.Trigger(t.Context(), Chat{ID: uuid.New()}, Message{}, agenthooks.EventStop, dispatch.CapacityClassGeneration) require.NoError(t, err) require.Empty(t, result.GetModelContext()) require.Empty(t, result.GetUserMessage()) @@ -145,7 +145,7 @@ func TestHookTriggerDeny(t *testing.T) { ToolUseID: "call_1", ToolName: "execute", ToolInput: json.RawMessage(`{}`), - }, agenthooks.EventPreToolUse) + }, agenthooks.EventPreToolUse, dispatch.CapacityClassGeneration) require.Nil(t, result) var denied *deniedError require.ErrorAs(t, err, &denied) @@ -174,7 +174,7 @@ func TestHookTriggerEventPayloads(t *testing.T) { ctx := testutil.Context(t, testutil.WaitShort) dispatchEvent := func(t *testing.T, msg Message, event agenthooks.EventType) agenthooks.Request { t.Helper() - _, err := trigger.Trigger(ctx, chat, msg, event) + _, err := trigger.Trigger(ctx, chat, msg, event, dispatch.CapacityClassGeneration) require.NoError(t, err) request := <-requests require.Equal(t, event, request.Type) @@ -215,7 +215,7 @@ func TestHookTriggerEventPayloads(t *testing.T) { dispatchEvent(t, Message{}, event) } - _, err := trigger.Trigger(ctx, chat, Message{}, agenthooks.EventType("bogus")) + _, err := trigger.Trigger(ctx, chat, Message{}, agenthooks.EventType("bogus"), dispatch.CapacityClassGeneration) require.ErrorContains(t, err, "unsupported hook event") } diff --git a/coderd/x/chatd/chathooks/tooluse.go b/coderd/x/chatd/chathooks/tooluse.go index 09dd15e2d4..daa19202a8 100644 --- a/coderd/x/chatd/chathooks/tooluse.go +++ b/coderd/x/chatd/chathooks/tooluse.go @@ -8,6 +8,7 @@ import ( "charm.land/fantasy" "golang.org/x/xerrors" + "github.com/coder/coder/v2/coderd/x/agenthooks/dispatch" "github.com/coder/coder/v2/codersdk" "github.com/coder/coder/v2/codersdk/x/agenthooks" ) @@ -59,7 +60,7 @@ func (t *Trigger) PreflightPendingToolCalls( ToolUseID: toolCall.ToolCallID, ToolName: toolCall.ToolName, ToolInput: json.RawMessage(toolCall.Input), - }, agenthooks.EventPreToolUse) + }, agenthooks.EventPreToolUse, dispatch.CapacityClassGeneration) if err != nil { denied, ok := errors.AsType[*deniedError](err) if !ok { @@ -139,7 +140,7 @@ func (t *Trigger) PostToolUseResults( } continue } - result, err := t.Trigger(ctx, chat, msg, agenthooks.EventPostToolUse) + result, err := t.Trigger(ctx, chat, msg, agenthooks.EventPostToolUse, dispatch.CapacityClassGeneration) if err != nil { if firstErr == nil { firstErr = err diff --git a/coderd/x/chatd/chathooks/trigger.go b/coderd/x/chatd/chathooks/trigger.go index 3af38fa65b..6b705fd4dc 100644 --- a/coderd/x/chatd/chathooks/trigger.go +++ b/coderd/x/chatd/chathooks/trigger.go @@ -142,6 +142,7 @@ func (t *Trigger) Trigger( chat Chat, msg Message, event agenthooks.EventType, + capacity dispatch.CapacityClass, ) (*Result, error) { if !t.Enabled() { return emptyResult, nil @@ -166,9 +167,10 @@ func (t *Trigger) Trigger( return nil, xerrors.Errorf("unsupported hook event %q", event) } response, _, err := t.dispatcher.Dispatch(ctx, dispatch.Event{ - Type: event, - ChatRef: chat.ref(), - Data: data, + Type: event, + ChatRef: chat.ref(), + Data: data, + Capacity: capacity, }) if err != nil { return nil, err diff --git a/coderd/x/chatd/generation.go b/coderd/x/chatd/generation.go index 8a440187a0..23179195b9 100644 --- a/coderd/x/chatd/generation.go +++ b/coderd/x/chatd/generation.go @@ -14,6 +14,7 @@ import ( "cdr.dev/slog/v3" "github.com/coder/coder/v2/coderd/database" + "github.com/coder/coder/v2/coderd/x/agenthooks/dispatch" "github.com/coder/coder/v2/coderd/x/chatd/chatdebug" "github.com/coder/coder/v2/coderd/x/chatd/chaterror" "github.com/coder/coder/v2/coderd/x/chatd/chathooks" @@ -399,7 +400,7 @@ func (s *taskStarter) startGenerationSession( // Re-arm the claim until its response is applied so a replacement task // can replay session_start effects. defer func() { complete(completed) }() - response, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(chat, input.hookTurnID()), chathooks.Message{Source: chathooks.SessionStartSource(messages)}, agenthooks.EventSessionStart) + response, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(chat, input.hookTurnID()), chathooks.Message{Source: chathooks.SessionStartSource(messages)}, agenthooks.EventSessionStart, dispatch.CapacityClassGeneration) if err != nil { return sessionStartResult{}, true, chathooks.GenerationDispatchError(agenthooks.EventSessionStart, err) } @@ -940,7 +941,7 @@ func (s *taskStarter) generateCompaction( overrideModel.modelConfig, ) } - preResult, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(prepared.Chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventPreCompact) + preResult, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(prepared.Chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventPreCompact, dispatch.CapacityClassGeneration) if err != nil { return chathooks.GenerationDispatchError(agenthooks.EventPreCompact, err) } @@ -987,7 +988,7 @@ func (s *taskStarter) generateCompaction( // Hook effects and fail-closed errors must commit atomically with // compaction; a separate commit races the runner and can be dropped // on crash. - postResult, postDispatchErr := s.server.hooks.Trigger(ctx, chathooks.ChatFor(prepared.Chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventPostCompact) + postResult, postDispatchErr := s.server.hooks.Trigger(ctx, chathooks.ChatFor(prepared.Chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventPostCompact, dispatch.CapacityClassGeneration) var postCommitErr error if postDispatchErr != nil { postCommitErr = chathooks.GenerationDispatchError(agenthooks.EventPostCompact, postDispatchErr) @@ -1349,7 +1350,7 @@ func (s *taskStarter) finishGenerationTurn( if err != nil { return normalizeTaskTransitionError(err, "load stop hook state") } - response, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventStop) + response, err := s.server.hooks.Trigger(ctx, chathooks.ChatFor(chat, input.hookTurnID()), chathooks.Message{}, agenthooks.EventStop, dispatch.CapacityClassGeneration) if err != nil { return s.finishGenerationError(ctx, machine, input, chathooks.GenerationDispatchError(agenthooks.EventStop, err), fence) } diff --git a/coderd/x/chatd/subagent.go b/coderd/x/chatd/subagent.go index 8a99b07acd..a9fccfcc17 100644 --- a/coderd/x/chatd/subagent.go +++ b/coderd/x/chatd/subagent.go @@ -1313,7 +1313,7 @@ func (p *Server) createChildSubagentChatWithOptions( ParentChatID: uuid.NullUUID{UUID: parent.ID, Valid: true}, RootChatID: uuid.NullUUID{UUID: rootChatID, Valid: true}, TurnID: &mintedTurnID, - }, promptMessage, agenthooks.EventUserPromptSubmit) + }, promptMessage, agenthooks.EventUserPromptSubmit, dispatch.CapacityClassGeneration) if err != nil { return database.Chat{}, chathooks.UserPromptDenial(err) } diff --git a/docs/admin/setup/chat-lifecycle-hooks.md b/docs/admin/setup/chat-lifecycle-hooks.md index e33cea7991..5be326cb4e 100644 --- a/docs/admin/setup/chat-lifecycle-hooks.md +++ b/docs/admin/setup/chat-lifecycle-hooks.md @@ -167,6 +167,13 @@ Coder checks admission before dispatching, but concurrent requests can still fai The consumer then observes an event for a request that Coder rejects, and the rejected request doesn't persist a prompt or tool result. Treat events as attempt notifications rather than proof of a committed operation, and key idempotent tool-event processing on `tool_use_id`. +Each `coderd` replica runs at most 256 dispatches at once and waits up to 250 ms for a free slot; a dispatch that waits out that limit fails as over capacity. +The limit is per replica rather than deployment-wide, so size the consumer for 256 concurrent requests per replica. +Slow consumer responses hold slots for longer, so a slow consumer turns a burst of chat activity into over-capacity failures. +Prompt admission (creating, sending, or editing a message) can hold at most 192 of a replica's slots, so at least 64 stay reachable only by dispatches for work a chat already admitted. +That bound stops a burst of new submissions from consuming every slot, but it doesn't make the remaining slots sufficient: a saturated dispatcher can still fail a dispatch for a running chat and leave that chat in the error state. +Watch `coderd_chatd_hook_dispatches_total{result="over_capacity"}` to see whether the consumer's latency is turning normal traffic into rejections. + Delivery is best-effort and can duplicate. Coder never queues a failed dispatch for redelivery, so plan for duplicates without assuming every event arrives. Coder retries one connection failure per dispatch with the same JWT, so use `dispatch_id` to recognize a repeated HTTP attempt and return the same response.