Refuse webhooks the agent cannot deliver, and name the timeout in the log

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)
+}

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

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")
+	}
+}

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

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))
+		})
+	}
+}

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 := ""