From b179fb3ac040b33c803a9e8cee984a88ba04f1c8 Mon Sep 17 00:00:00 2001 From: Keith Zeto Date: Fri, 11 Sep 2026 13:07:19 -0700 Subject: [PATCH 1/2] Refuse webhooks the agent cannot deliver, and name the timeout in the log Three faults in the webhook path, all of which turn a delayed handler into silently lost messages. The dispatch queue holds 100 invocations and Trigger sent to it with a blocking send. Trigger runs on the inbound HTTP goroutine, so webhook 101 hung that request with no timeout while the goroutines piled up. Make the send non-blocking and return ErrDispatchQueueFull instead. A refused invocation is cancelled, so it completes rather than sitting until its own timeout expires. The webhook handler answered 2xx as soon as it queued the work. When nothing was draining the queue the sender was told the webhook had been received, the invocation timed out unseen, and the message was gone. Return 503 with Retry-After on a full queue so the sender can retry. This is what Paychex reported: "axon is responding with a 2xx response telling us the webhook is being successfully received, but then it never is making it to our code." Both timeout paths report code "timeout" with no message, and the log line rendered the message alone, so the operator saw "invocation error: " and nothing else. Render the code when there is no message, and give the agent's own timer a message saying what it waited for. The handlerManager data races that -race reports here are pre-existing and unrelated, as is TestGRPCServer_ClientAutoClose, which fails on main. Co-Authored-By: Claude Opus 5 (1M context) --- agent/server/handler/handler_manager_test.go | 39 +++++++++++++++ agent/server/handler/manager.go | 26 +++++++++- agent/server/http/http_server_webhook_test.go | 50 +++++++++++++++++++ agent/server/http/webhook_handler.go | 11 ++++ agent/server/report_error_test.go | 29 +++++++++++ agent/server/server.go | 28 +++++++++-- 6 files changed, 178 insertions(+), 5 deletions(-) create mode 100644 agent/server/report_error_test.go diff --git a/agent/server/handler/handler_manager_test.go b/agent/server/handler/handler_manager_test.go index 9496360..ad2588b 100644 --- a/agent/server/handler/handler_manager_test.go +++ b/agent/server/handler/handler_manager_test.go @@ -199,3 +199,42 @@ func TestTriggerAndDequeueTimeoutContext(t *testing.T) { require.NoError(t, err) require.Nil(t, h) } + +func TestTriggerRejectsWhenQueueIsFull(t *testing.T) { + logger, _ := zap.NewDevelopment() + mgr := NewHandlerManager(logger, cron.New(), nil) + + id, err := mgr.RegisterHandler("1", "handler1", defaultTimeout, FixtureHandlerOption()) + require.NoError(t, err) + require.NoError(t, mgr.Start("1")) + + entry := mgr.ListHandlers()[0] + require.Equal(t, id, entry.Id()) + + // Nothing dequeues, so the queue fills and every later trigger must be + // refused rather than block the caller. + for i := 0; i < dispatchQueueDepth; i++ { + require.NoError(t, mgr.Trigger(NewHandlerInvoke(entry, pb.HandlerInvokeType_WEBHOOK, nil))) + } + + rejected := NewHandlerInvoke(entry, pb.HandlerInvokeType_WEBHOOK, nil) + + done := make(chan error, 1) + go func() { done <- mgr.Trigger(rejected) }() + + select { + case err := <-done: + require.ErrorIs(t, err, ErrDispatchQueueFull) + case <-time.After(5 * time.Second): + require.Fail(t, "Trigger blocked on a full queue instead of rejecting") + } + + // A refused invocation is finished, not left dangling for its full timeout. + select { + case <-rejected.Done(): + case <-time.After(5 * time.Second): + require.Fail(t, "refused invocation was never completed") + } + _, err = rejected.GetResult() + require.Error(t, err) +} diff --git a/agent/server/handler/manager.go b/agent/server/handler/manager.go index 55ad489..10358e1 100644 --- a/agent/server/handler/manager.go +++ b/agent/server/handler/manager.go @@ -2,6 +2,7 @@ package handler import ( "context" + "errors" "fmt" "os" "time" @@ -12,6 +13,15 @@ import ( "go.uber.org/zap" ) +// dispatchQueueDepth is how many invocations may wait for an SDK client +// before Trigger starts refusing them. +const dispatchQueueDepth = 100 + +// ErrDispatchQueueFull means no SDK client drained the queue fast enough. +// Callers that front an inbound request must turn it into a retryable +// response rather than accepting work that will only time out. +var ErrDispatchQueueFull = errors.New("dispatch queue is full") + type Manager interface { RegisterHandler(dispatchId string, name string, timeout time.Duration, options ...*pb.HandlerOption) (string, error) UnregisterHandler(id string) @@ -239,7 +249,19 @@ func (s *handlerManager) Trigger(handler Invocable) error { startTime := time.Now() queue := s.getDispatchQueue(entry.DispatchId()) - queue <- message + select { + case queue <- message: + default: + // Never block here: Trigger runs on the inbound HTTP goroutine for a + // webhook, and a blocked send holds that request open with no timeout. + // Cancelling completes the invocation rather than leaking it. + cancel() + s.queueDepthGauge.WithLabelValues(handlerName).Set(float64(len(queue))) + s.invokeCounter.WithLabelValues(handlerName, "rejected").Inc() + logger.Warn("Dispatch queue is full, rejecting invocation", + zap.Int("depth", len(queue))) + return ErrDispatchQueueFull + } s.queueDepthGauge.WithLabelValues(handlerName).Set(float64(len(queue))) go func() { @@ -264,7 +286,7 @@ func (s *handlerManager) getDispatchQueue(DispatchId string) chan Invocable { queue, ok := s.dispatchQueues[DispatchId] if !ok { - queue = make(chan Invocable, 100) + queue = make(chan Invocable, dispatchQueueDepth) s.dispatchQueues[DispatchId] = queue } return queue diff --git a/agent/server/http/http_server_webhook_test.go b/agent/server/http/http_server_webhook_test.go index 829c92f..2d34de0 100644 --- a/agent/server/http/http_server_webhook_test.go +++ b/agent/server/http/http_server_webhook_test.go @@ -72,3 +72,53 @@ func TestHandleWebhook(t *testing.T) { } require.Equal(t, expected, handlerInvocation.ToDispatchInvoke().Args) } + +func TestHandleWebhookRejectsWhenQueueIsFull(t *testing.T) { + logger, _ := zap.NewDevelopment() + manager := handler.NewHandlerManager(logger, cron.New(), nil) + + option := &pb.HandlerOption{ + Option: &pb.HandlerOption_Invoke{ + Invoke: &pb.HandlerInvokeOption{ + Type: pb.HandlerInvokeType_WEBHOOK, + Value: "my-webhook-id", + }, + }, + } + + _, err := manager.RegisterHandler("1", "test", time.Minute, option) + require.NoError(t, err) + require.NoError(t, manager.Start("1")) + + webhookHandler := NewWebhookHandler(config.AgentConfig{}, logger, manager, nil) + router := mux.NewRouter() + webhookHandler.RegisterRoutes(router) + ts := httptest.NewServer(router) + defer ts.Close() + + post := func() int { + req, err := http.NewRequest(http.MethodPost, ts.URL+"/webhook/my-webhook-id", strings.NewReader("payload")) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + return resp.StatusCode + } + + // Nothing is dequeuing, so the queue fills up and every webhook after + // that must be refused instead of accepted and silently dropped. + for i := 0; i < 100; i++ { + require.Equal(t, http.StatusOK, post()) + } + + done := make(chan int, 1) + go func() { done <- post() }() + + select { + case status := <-done: + require.Equal(t, http.StatusServiceUnavailable, status) + case <-time.After(5 * time.Second): + require.Fail(t, "webhook request hung on a full queue instead of returning 503") + } +} diff --git a/agent/server/http/webhook_handler.go b/agent/server/http/webhook_handler.go index c5e5482..05a4401 100644 --- a/agent/server/http/webhook_handler.go +++ b/agent/server/http/webhook_handler.go @@ -2,6 +2,7 @@ package http import ( "encoding/json" + "errors" "fmt" "io" "net/http" @@ -101,6 +102,16 @@ func (h *webhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { err = h.handlerManager.Trigger(handler.NewWebhookHandlerInvoke(entry, r.URL, string(bodyBytes), contentType)) if err != nil { + // A queued-but-undeliverable webhook is worse than a refused one: the + // sender is told 2xx, the invocation times out unseen, and the message + // is gone. 503 lets the sender retry. + if errors.Is(err, handler.ErrDispatchQueueFull) { + h.logger.Warn("Rejecting webhook, no handler is draining the queue", + zap.String("webhookId", webhookId)) + w.Header().Set("Retry-After", "5") + writeStatus(http.StatusServiceUnavailable) + return + } h.logger.Error("Failed to trigger webhook", zap.Error(err)) writeStatus(http.StatusInternalServerError) return diff --git a/agent/server/report_error_test.go b/agent/server/report_error_test.go new file mode 100644 index 0000000..910647d --- /dev/null +++ b/agent/server/report_error_test.go @@ -0,0 +1,29 @@ +package server + +import ( + "testing" + + pb "github.com/cortexapps/axon/.generated/proto/github.com/cortexapps/axon" + "github.com/stretchr/testify/require" +) + +// A timeout is reported with a code and no message, so rendering the message +// alone produced "invocation error: " and hid what had happened. +func TestDescribeError(t *testing.T) { + cases := []struct { + name string + err *pb.Error + expected string + }{ + {"code and message", &pb.Error{Code: "unexpected", Message: "boom"}, "unexpected: boom"}, + {"code only", &pb.Error{Code: "timeout"}, "timeout"}, + {"message only", &pb.Error{Message: "boom"}, "boom"}, + {"neither", &pb.Error{}, "unknown"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.expected, describeError(tc.err)) + }) + } +} diff --git a/agent/server/server.go b/agent/server/server.go index 5de1805..3c0d74e 100644 --- a/agent/server/server.go +++ b/agent/server/server.go @@ -208,7 +208,12 @@ func (s *AxonAgent) sendInvocations(dispatchId string, stream pb.AxonAgent_Dispa <-time.After(time.Duration(msg.TimeoutMs) * time.Millisecond) s.ReportInvocation(context.Background(), &pb.ReportInvocationRequest{ HandlerInvoke: msg, - Message: &pb.ReportInvocationRequest_Error{Error: &pb.Error{Code: "timeout"}}, + Message: &pb.ReportInvocationRequest_Error{Error: &pb.Error{ + Code: "timeout", + Message: fmt.Sprintf( + "handler did not report a result within %dms", + msg.TimeoutMs), + }}, }) }(msg.InvocationId) } @@ -227,6 +232,23 @@ func (s *AxonAgent) setOutstandingRequest(id string, req *inflightRequest) { s.outstandingRequests[id] = *req } +// describeError renders a reported error for a log line. The code carries the +// only detail some senders set — an SDK timeout reports code "timeout" with no +// message — so printing the message alone leaves "invocation error: " and no +// way to tell what went wrong. +func describeError(e *pb.Error) string { + code, message := e.GetCode(), e.GetMessage() + switch { + case code != "" && message != "": + return fmt.Sprintf("%s: %s", code, message) + case code != "": + return code + case message != "": + return message + } + return "unknown" +} + // ReportInvocation is called by the client to report the result of an invocation, which will // log the result of an invocation into the history path. func (s *AxonAgent) ReportInvocation(ctx context.Context, req *pb.ReportInvocationRequest) (*pb.ReportInvocationResponse, error) { @@ -241,8 +263,8 @@ func (s *AxonAgent) ReportInvocation(ctx context.Context, req *pb.ReportInvocati var requestErr error - if req.GetError() != nil { - requestErr = fmt.Errorf("invocation error: %s", req.GetError().GetMessage()) + if e := req.GetError(); e != nil { + requestErr = fmt.Errorf("invocation error: %s", describeError(e)) } requestResult := "" From fa75f2f2572d116fc6c435c899881e24bb971297 Mon Sep 17 00:00:00 2001 From: Keith Zeto Date: Mon, 14 Sep 2026 10:43:12 -0700 Subject: [PATCH 2/2] Stop the queue-full test from racing its own scheduler The test registered the handler with the shared fixture option, which triggers on a 1ms interval, then started the handler. Those scheduled invocations went into the same dispatch queue that the test was filling, so the queue could reach its depth before the fill loop finished and the loop's own Trigger was refused. It failed 58 times in 100 local runs. Register a WEBHOOK option instead: the handler is active, but only the test puts work in the queue. Co-Authored-By: Claude Opus 5 (1M context) --- agent/server/handler/handler_manager_test.go | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/agent/server/handler/handler_manager_test.go b/agent/server/handler/handler_manager_test.go index ad2588b..7bf7c2c 100644 --- a/agent/server/handler/handler_manager_test.go +++ b/agent/server/handler/handler_manager_test.go @@ -204,7 +204,17 @@ func TestTriggerRejectsWhenQueueIsFull(t *testing.T) { logger, _ := zap.NewDevelopment() mgr := NewHandlerManager(logger, cron.New(), nil) - id, err := mgr.RegisterHandler("1", "handler1", defaultTimeout, FixtureHandlerOption()) + // The fixture option triggers on an interval, which would put unrelated + // invocations into the same queue while it fills below. + webhookOnly := &pb.HandlerOption{ + Option: &pb.HandlerOption_Invoke{ + Invoke: &pb.HandlerInvokeOption{ + Type: pb.HandlerInvokeType_WEBHOOK, + }, + }, + } + + id, err := mgr.RegisterHandler("1", "handler1", defaultTimeout, webhookOnly) require.NoError(t, err) require.NoError(t, mgr.Start("1"))