[shawnburke]

Summary

Adds a failure hook to the Go and Python handler SDKs (WithFailureHook / on_failure) that fires for every failed invocation. The hook receives the original message and a reason that separates handler failures from Axon failures. A customer can then route failed webhooks to a store of their own choosing and replay them later. Before this change, those messages failed silently. Tracks COR2-1098.

The three failure cases asked for, and what the hook receives for each:

Scenario	Detected by	reason	source
Handler raises / returns an error / panics	SDK	HANDLER_ERROR	APPLICATION
Handler runs past its timeout	SDK	TIMEOUT	APPLICATION, or RUNTIME if it was waiting on a Cortex API call or for a free handler
Invocation never answered (client hung or gone)	Agent	TIMEOUT	RUNTIME
Agent queue full (webhook got 503)	Agent	NO_RESOURCES	RUNTIME, caller_rejected=true
Expired in the agent queue before any client took it	Agent	NO_RESOURCES	RUNTIME

Every failure, wherever it is detected, goes through the agent. The agent sends it to the SDK's hook, the hook answers drop or retry, and the agent acts on the answer. Each failure is delivered once per invocation; one the hook hadn't answered when its connection dropped is sent again.

Python usage

from cortex_axon import AxonClient, FailureAction, FailureSource
from cortex_axon.handler import cortex_webhook


@cortex_webhook(id="employee-sync")
def employee_sync(context):
    update_records(context.args["body"])   # raising here is a HANDLER_ERROR


def on_failure(info, failure):
    dead_letter_topic.publish({
        "invocation_id": info.invocation_id,
        "handler": info.handler_name,
        "attempt": info.attempt,
        "reason": failure.reason.name,      # HANDLER_ERROR / TIMEOUT / NO_RESOURCES
        "source": failure.source.name,      # APPLICATION / RUNTIME
        "caller_rejected": failure.caller_rejected,
        "message": info.args,               # body, content-type, url
    })
    if failure.source == FailureSource.RUNTIME and info.attempt < 3:
        return FailureAction.RETRY             # agent re-queues it
    return FailureAction.DROP


AxonClient(scope=globals(), on_failure=on_failure).run()

Design

Flow

handler error / timeout in SDK ─┐ ReportInvocation{error, failure}
agent timeout (no report)      ─┤ completeInvocation(report with failure)
queue full / expired in queue  ─┘ invocation.Complete(FailureError)
                                     │
                       Manager: one completion watcher
                                     │ FailureError
                                     ▼
                              FailureManager  (queues, metric, retries)
                                     │ DispatchEvents (bidi)
                                     ▼
                         SDK runs on_failure → answers DROP | RETRY
                                     │
                     FailureEvent.Resolve → Manager.Trigger(attempt+1)

Agent

• One path into FailureManager. Every agent-visible failure completes its invocation with a FailureError, and a single watcher in Manager (watch) hands it to the FailureManager. The four sources are:

  • the SDK's report: ReportInvocationRequest.failure;
  • the agent's own timeout: a synthesized report through completeInvocation;
  • a full queue;
  • expiry while queued.

  An error reported without failure, as older SDKs send them, is only recorded.

• FailureManager (handler/failure_manager.go) owns everything failure-related:

  • the per-dispatch queues, which are unbounded so no failure is dropped;
  • events awaiting an answer;
  • the axon_handler_failure_events_pending metric;
  • retries, which go back through Manager.Trigger, and the 5-attempt cap.

  It queues only for clients that set DispatchRequest.failure_events.

• Generic sources and a simple send loop. InvocationSource[T] has one method, Dequeue(ctx, dispatchId), and both Manager and FailureManager implement it. A failure is handed out as a *FailureEvent handle, and the stream calls its Resolve(action) when the answer arrives, or Requeue() if the stream ends first. drain() sends whatever a source hands out: Dispatch drains Manager, and DispatchEvents drains FailureManager. They stay separate streams because the Python SDK stops reading Dispatch while all its handler slots are busy, which is exactly when failures happen.

• Invocation state. A single SetState with an explicit transition table. A queued invocation moves once, to dispatched, expired or rejected. Once dispatched, it ends only by a report or by the agent's timeout, not by its queue timeout.

• Deadline. Each dispatched invocation carries an absolute deadline: when the agent stops waiting and reports the failure itself (timeout + 5s grace).

  • The SDKs won't start an invocation past its deadline and cap a handler's timeout at it.
  • This matters because a saturated client otherwise reads invocations from the stream buffer long after the agent has reported them, and runs them anyway. A replaying hook would then process them twice. I reproduced that against a real agent: 109 of 110 stale invocations still ran.
  • The agent normally runs as a localhost sidecar, so clock skew isn't a concern.

SDKs

• They only report and answer. The SDKs don't run the hook for their own failures and don't call any retry RPC. They classify the failure into ReportInvocationRequest.failure, which covers:

  • APPLICATION vs RUNTIME;
  • a timeout during a Cortex API call, or while waiting for a free handler, counted as RUNTIME.

  The hook runs only for events from the agent, and its return value is sent back as the answer.

• Python: on_failure=, returning FailureAction.DROP / RETRY. Returning None or raising means DROP. Events are read on their own thread and run on their own pool, never waiting for a handler slot. At the deadline the SDK reports the timeout itself and drops anything the handler does afterwards, since a thread can't be stopped.

• Go: WithFailureHook, returning FailureActionDrop / FailureActionRetry.

• Fixes found along the way:

  • The Go SDK's invokeHandler let a timed-out handler write to the report while it was being sent.
  • Run(ctx) ignored cancellation: it opened its stream with context.Background(), and its exit flag was raced.

• Test support: the agent's webhook port can be set with WEBHOOK_PORT.

Compatibility

Agent	SDK	Behavior
new	old	Unchanged. The old SDK doesn't set failure_events, so nothing is queued, and it ignores the new fields.
old	new	Handlers run as before. DispatchEvents returns Unimplemented, so the SDK logs once and stops asking, without redialing in a loop. The hook doesn't fire, because failures are delivered by the agent.

DispatchEvents and the failure messages are new in this PR, so changing them breaks no released contract.

Known limits

• If the handler process is wedged, the hook can't run. Failures wait in the agent and are delivered when the process reconnects.
• Failures live in agent memory, so an agent restart loses any not yet answered. A durable agent-side sink would close this and is worth a follow-up.
• Delivery is at-least-once: a failure whose answer was lost with its connection is sent again, so a hook may see the same invocation_id twice.
• Under SDK-side saturation, invocations the agent already sent come back as TIMEOUT/RUNTIME rather than NO_RESOURCES. The source is still right.

Test plan

• Agent (go test ./..., make test-race):
  • unit tests:
    • state transitions;
    • each source through the single path (queue full, expired in the queue, an SDK-reported failure), and a plain error is not a failure;
    • a dispatched invocation doesn't expire on its queue timeout;
    • 10k failures with none dropped, kept in order;
    • unanswered failures requeued first;
    • resolve: retry gives attempt 2, drop, a second answer is rejected, retry cap, lookup after re-registration;
  • e2e over real gRPC:
    • unreported → TIMEOUT → answered RETRY → attempt 2;
    • a client-reported failure is delivered once and DROP doesn't rerun it;
    • an unanswered failure is resent after a disconnect;
  • webhook 503 produces a caller_rejected failure.
• Go SDK (go test -race ./...):
  • reported reason/source for error, panic, timeout, a timeout during an API call, and a deadline-capped timeout;
  • the hook isn't run locally;
  • event → hook → answer, for both RETRY and DROP;
  • a panicking hook answers DROP and the stream continues;
  • an Unimplemented agent is not redialed;
  • the failure_events flag.
• Python SDK (31 tests including e2e, ruff clean):
  • the same reported mapping; the hook isn't run locally;
  • a slow handler is APPLICATION and its late result is dropped;
  • an expired invocation is not started;
  • a deadline-capped timeout is RUNTIME;
  • event → hook → answer; returning nothing or raising answers DROP;
  • 20 events are answered while every handler slot is held;
  • an older agent is handled.
• Automated end-to-end, SDK against a real agent built from agent/ and triggered over its webhook endpoint:
  • Go (sdks/go/e2e_failure_test.go, skipped with -short): a handler error and a timeout each go SDK report → agent → hook → RETRY, the retry succeeds with the original body, and the failure is not delivered twice.
  • Python (tests/test_e2e_failures.py): the same two cases, plus saturation. One slot is stuck, the agent reports the queued invocations, the hook retries them, and each message succeeds exactly once, with no expired original run. The Python workflow now sets up Go to build the agent; without Go the tests skip.
• Manual end-to-end (before this iteration's refactor; the automated e2e tests above cover the same cases now): real agent binary (DRYRUN) with the Python SDK and the HTTP webhook endpoint:
  • Raising handler: HANDLER_ERROR/APPLICATION, then a successful retry as attempt 2.
  • Slow handler: one TIMEOUT/APPLICATION, and the late result was dropped.
  • 110 webhooks against 1 handler slot: each message reached the hook exactly once, and 109 stale invocations were skipped instead of run.

🤖 Generated with Claude Code

[shawnburke]

What this means for handler authors

The problem

When a webhook message failed to process, it disappeared without a trace. A message could fail because:

• the handler raised;
• it ran out of time;
• Axon was too busy to hand it to a handler at all.

Your code got no signal. The only way to find out was to notice later that data hadn't updated. Even when you did notice, you couldn't tell whether your handler or Axon was to blame.

The solution

You can now register a failure hook with the Axon SDK. Axon calls it for every message that fails, including the ones that never reached your handler. It passes:

• the original message: the webhook body, content type and URL, plus the handler name and an invocation id;
• why it failed: one of
  • HANDLER_ERROR: your handler raised.
  • TIMEOUT: it didn't finish in time.
  • NO_RESOURCES: Axon had no free capacity to run it.
• whose fault it was:
  • APPLICATION: your code.
  • RUNTIME: Axon itself, or a Cortex API call it was waiting on.

What the hook does is up to you: save the message to a topic, a table or a file, or raise an alert. Then tell Axon to drop the message or retry it. Axon doesn't store or replay anything itself; your hook decides what happens.

Example

from cortex_axon import AxonClient, FailureAction, FailureSource
from cortex_axon.handler import cortex_webhook


# A normal webhook handler. If it raises, the failure hook is called.
@cortex_webhook(id="employee-sync")
def employee_sync(context):
    update_records(context.args["body"])


# Called for every message that fails.
def on_failure(info, failure):
    # Keep the failed message so it can be replayed later.
    dead_letter_topic.publish({
        "invocation_id": info.invocation_id,
        "handler": info.handler_name,
        "attempt": info.attempt,
        "reason": failure.reason.name,   # HANDLER_ERROR, TIMEOUT or NO_RESOURCES
        "source": failure.source.name,   # APPLICATION or RUNTIME
        "message": info.args,            # the original webhook: body, content-type, url
    })

    # Have Axon try Axon-side failures again (up to 5 attempts); drop the rest.
    if failure.source == FailureSource.RUNTIME and info.attempt < 3:
        return FailureAction.RETRY
    return FailureAction.DROP


AxonClient(scope=globals(), on_failure=on_failure).run()

Good to know

• Busy periods: when Axon is too busy to accept a webhook, the sender still gets a 503 so it can resend. Your hook is told too, with failure.caller_rejected = True, so you can avoid replaying a message that will also be resent.
• Every failure gets to your hook, even under heavy load. If the connection drops before your hook answers, the failure is sent again, so use invocation_id to avoid saving it twice.
• Retries: returning FailureAction.RETRY has Axon queue the message again as a new attempt. Returning FailureAction.DROP, or nothing, drops it.
• Timeouts in Python: a handler that runs past its timeout is reported as a TIMEOUT at the deadline. Anything it does after that is ignored.
• Upgrade both sides: the failure hook needs both the Axon agent and the SDK upgraded, because Axon is what sends failures to your hook. With an older agent, handlers run as before but the hook doesn't fire.
• Go: the same hook is available as axon.WithFailureHook.

[shawnburke]

please collapse into SetState(state) so that these are not discrete operations and we have room to grow.

[shawnburke]

Done in 23b4a53. ClaimDispatch/ClaimExpired are now one SetState(state) bool plus State(), checked against an explicit transition table (invokeTransitions in handler_invoke.go). Today a queued invocation can move to InvokeDispatched, InvokeExpired or InvokeRejected (new: the queue-full case), and nothing else. A new state is a new constant plus a table entry.

[shawnburke]

wait why is the SDK deciding when to retry, shouldn't the server do this on its own? this is moving flow control outside of the SDK.

[shawnburke]

prob should be debug

[shawnburke]

Done in 71f8c6f: "Dropping failed invocation" now logs at debug.

[shawnburke]

why do we need this versus just a q.poll(ctx) -> dispatchEvent? So:

• Wait duration is controlled by Context timeout
• All interested parties can call poll which, if queue is empty, sits on an internal chan struct{}, then the first one to atomically dequeue an event wins eg callers use a select to do this in a loop
• This atomically sets events to pending (or adds them to the map, whatever).

This means we don't need to separate wake and poll, but rather whenever a new item is added to the queue we add an item to the channel and one of the waiters gets it. I think it means we don't need the timer poll either.

The whole next/take thing seems weird. there should just be poll and that return should have a methods like requeue() or close() so the caller doesn't need to stash the invocation ID.

[shawnburke]

Agreed, reworked in 71f8c6f along these lines:

• One call. Dequeue(ctx, dispatchId) waits until there is a failure or ctx ends. The timer and the separate wake are gone.
• Waiting. While the queue is non-empty, an internal chan struct{} is closed. Every waiter wakes, the first to take the lock pops the event, and the rest wait again. I used close-to-broadcast rather than one token per item: a buffered channel can't hold a token for every item in an unbounded queue, and the old one-slot version needed the re-wake you flagged.
• A handle, not an id. Dequeue returns a *FailureEvent with Resolve(action) and Requeue(), and only the first call takes effect. The queue no longer keeps a pending map, so next/take are gone. DispatchEvents still keeps the handles by invocation id, because the client's answer arrives on the stream with only that id.

[shawnburke]

From above It seems like this is just Dequeue(ctx, dispatchId) (event, error) which is more idomatic Go

[shawnburke]

Done: InvocationSource[T] is now Dequeue(ctx, dispatchId) (T, error). When the context ends it returns ctx.Err(), and the zero-value check is gone. Manager and FailureManager both implement it. drain adds a per-call timeout only for Dispatch, which needs a periodic idle check to notice when every handler has finished.