diff --git a/agent/server/handler/handler_manager_test.go b/agent/server/handler/handler_manager_test.go index 9496360..7bf7c2c 100644 --- a/agent/server/handler/handler_manager_test.go +++ b/agent/server/handler/handler_manager_test.go @@ -199,3 +199,52 @@ 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) + + // 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")) + + 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 := ""