feat: durable hitl webhook delivery on restate with a 24h retry window and dead letter

docs/content/docs/cloud/monitoring.mdx

@@ -126,6 +126,18 @@ attempt at a host-to-host artifact copy (create seeding, off-host fork,
 migrate, checkpoint deltas), with restate on or off. Failures concentrated on
 one host pair point at the network between them or one agent.
 
+**HITL webhooks**: `orchestrator_hitl_webhook_attempts_total{endpoint}`,
+`orchestrator_hitl_webhook_failures_total{endpoint}` and
+`orchestrator_hitl_webhook_last_status{endpoint}` (the http status of the last
+try, `0` when no receiver answered) cover every delivery attempt, with restate
+on or off. With restate on, `orchestrator_hitl_webhook_backlog{endpoint}` is
+the deliveries not yet delivered or dead-lettered and
+`orchestrator_hitl_webhook_dead_letters_total{endpoint}` the ones still failing
+after 24 hours. `endpoint` is the webhook url's host and port only: the path
+and query can carry a token. The backlog is per orchestrator process, so after
+a restart a delivery waiting out its backoff is counted again from its next
+try.
+
 **HTTP health**: the standard RED metrics apply directly.
 
 - `rate(orchestrator_http_requests_total{code=~"5.."}[5m])` for server error
@@ -149,6 +161,7 @@ saturation cannot be broken down by route.
 | `/ready` returning `503` for 2+ minutes | Critical | The orchestrator can't serve traffic; likely a Postgres outage. |
 | `rate(orchestrator_http_requests_total{code=~"5.."}[5m]) > 0` sustained | Warning | Server errors on the API surface. |
 | `orchestrator_reconcile_orphans_dead_lettered_total` increasing | Warning | VMs stuck in an unrecoverable state, needs manual cleanup. |
+| `orchestrator_hitl_webhook_dead_letters_total` increasing | Warning | A webhook receiver was unreachable or failing for 24 hours; its notifications were dropped. |
 | `orchestrator_fleet_egress_endpoints{state="unhealthy"} > 0` for 2+ cycles | Warning | Proxy-mode environments with no outbound access; the backend on that host is down. |
 | `time() - orchestrator_host_last_seen_timestamp_seconds > 300` | Warning | The orchestrator can't reach that host's agent. |
 | `orchestrator_host_allocated{resource="vms"} / orchestrator_host_capacity{resource="vms"} > 0.9` | Warning | That host is nearly full; creates will start landing elsewhere or failing placement. |

docs/content/docs/cloud/orchestrator-deploy.mdx

@@ -71,15 +71,15 @@ range covers the fleet.
 | `FUSE_RESTATE_INGRESS_URL` | | | Restate ingress URL, required unless the mode is `off`. `http://127.0.0.1:18080` with `ops/restate/compose.yaml`. |
 | `FUSE_RESTATE_API_KEY` / `FUSE_RESTATE_IDENTITY_KEY` | | | Mode `cloud` only: the Cloud API key and the `publickeyv1_...` signing key. Other cloud settings are on [Restate Cloud](/docs/cloud/restate-cloud#settings). |
 | `ORCH_RESTATE_LISTEN` | | `127.0.0.1:9080` | Restate handler endpoint (h2c). Only restate-server, or the TLS proxy in front of it in cloud public mode, should reach it. |
-| `FUSE_RESTATE_RETRY_<SERVICE>_INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `<SERVICE>` is `ENVIRONMENT`, `HOST`, `ARTIFACTS` or `ARTIFACTPIN`. The `ARTIFACTS` policy also bounds the retries of an artifact move's pull; the `ENVIRONMENT` policy bounds the host steps of a destroy, snapshot, restore or token rotation upload (a rotation's state store write retries without bound), and the retries of a snapshot gc deletion. |
+| `FUSE_RESTATE_RETRY_<SERVICE>_INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `<SERVICE>` is `ENVIRONMENT`, `HOST`, `ARTIFACTS`, `ARTIFACTPIN` or `WEBHOOKS`. The `ARTIFACTS` policy also bounds the retries of an artifact move's pull; the `ENVIRONMENT` policy bounds the host steps of a destroy, snapshot, restore or token rotation upload (a rotation's state store write retries without bound), and the retries of a snapshot gc deletion. The `WEBHOOKS` policy only covers restate failing to run one delivery attempt; a receiver that is down is retried on its own 24-hour schedule (see [webhooks](/docs/guides/human-in-the-loop)). |
 
 With `FIRECRACKER_BASE_URL` unset, the orchestrator runs the in-memory stub provider
 (fine for local dev, not for production). With `DATABASE_URL` unset, state is
 in-memory and does not survive a restart.
 
 ## Restate
 
-Restate is optional and off by default. With it on, destroys, artifact moves, snapshots, restores, the checkpoint chain, snapshot gc and token rotation run on it. To run it next to the orchestrator:
+Restate is optional and off by default. With it on, destroys, artifact moves, snapshots, restores, the checkpoint chain, snapshot gc, token rotation and hitl webhook deliveries run on it. To run it next to the orchestrator:
 
 ```bash
 docker compose -f ops/restate/compose.yaml up -d

docs/content/docs/guides/human-in-the-loop.mdx

@@ -1253,10 +1253,21 @@ def verify_fuse(secret: bytes, header: str, raw_body: bytes, tolerance: int = 30
     return hmac.compare_digest(want, v1)
 ```
 
-A delivery is tried up to 4 times: at once, then after 1s, 4s and 16s, on
-any transport error or non-2xx answer, with a 10s timeout per try. Redirects
-are not followed. Deliveries are best-effort and not stored; the audit log is
-the record, the webhook is a nudge.
+A delivery is retried on any transport error or non-2xx answer, with a 10s
+timeout per try. Redirects are not followed. With restate off it is tried up
+to 4 times: at once, then after 1s, 4s and 16s, in the orchestrator process,
+so a restart drops it; the audit log is the record, the webhook is a nudge.
+
+With restate on (`FUSE_RESTATE_MODE=self` or `cloud`) and `TOKEN_ENCRYPTION_KEY`
+set, each delivery is a restate `Webhooks.Deliver` invocation and survives an
+orchestrator restart. It is retried after 1s, then four times longer each try
+up to once an hour, for 24 hours from the event; a delivery still failing
+after that is recorded as a `hitl_webhook` dead letter. The same notification
+can arrive more than once if the orchestrator dies mid-post, so dedupe on
+`intervention_id` and `event`. Restate journals the delivery, but the
+endpoint url and signing secret go in sealed with `TOKEN_ENCRYPTION_KEY`;
+the journal holds the notification's ids, gate, deadline and link. Without
+the key, deliveries stay in process as with restate off.
 
 The signing secrets are sealed with `TOKEN_ENCRYPTION_KEY`. Without it they
 are kept in memory only, so deliveries stop after an orchestrator restart

docs/runbooks/restate.mdx

@@ -202,6 +202,27 @@ decrypts; run it again. a rotation's journal can be read with the query below. r
 q "SELECT j.index, j.name, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id WHERE i.target_handler_name = 'RotateToken'"
 ```
 
+hitl webhooks run as `Webhooks/Deliver`, one invocation per attempt at one
+delivery to one endpoint. a failed attempt does not sleep: it sends the next
+attempt to itself delayed (1s, then four times longer, capped at an hour) and
+ends, so a receiver down for hours is a chain of short invocations, each on
+the deployment current when it starts, and an orchestrator upgrade in between
+needs nothing. once the next attempt would land more than 24h after the event,
+the delivery is written as a `hitl_webhook` dead letter keyed by its delivery
+id (`orchestrator_hitl_webhook_dead_letters_total`) and the chain ends. an
+endpoint that no longer decrypts (a `TOKEN_ENCRYPTION_KEY` change) is
+dead-lettered at once. the input journals the notification (event, intervention
+and env ids, gate, deadline, the intervention link), the endpoint host, and
+the url and signing secret as ciphertext; the attempt result journals the
+status and the error with the url stripped. read them with:
+
+```sh
+q "SELECT i.id, j.index, j.name, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id WHERE i.target_service_name = 'Webhooks'"
+q "SELECT id, status, scheduled_at FROM sys_invocation WHERE target_service_name = 'Webhooks' AND status <> 'completed'"
+```
+
+the second query is the deliveries waiting for their next attempt.
+
 copies are metered by `orchestrator_artifact_move_duration_seconds`,
 `orchestrator_artifact_move_bytes_total` and
 `orchestrator_artifact_move_failures_total`, labeled `source_host` and

internal/durable/config.go

@@ -130,7 +130,8 @@ type Config struct {
 //	FUSE_RESTATE_RETRY_<SERVICE>_MAX_INTERVAL       e.g. 30s
 //	FUSE_RESTATE_RETRY_<SERVICE>_MAX_ATTEMPTS       e.g. 10
 //
-// <SERVICE> is the upper-cased service name: ENVIRONMENT or HOST.
+// <SERVICE> is the upper-cased service name: ENVIRONMENT, HOST, ARTIFACTS,
+// ARTIFACTPIN or WEBHOOKS.
 //
 // mode cloud also reads:
 //

internal/durable/server.go

@@ -28,10 +28,13 @@ const (
 	// ArtifactPinObject is keyed by snapshot id and holds the pins that keep
 	// artifact gc off it.
 	ArtifactPinObject = "ArtifactPin"
+	// WebhooksService is a plain service: each Deliver invocation is one
+	// attempt at one hitl webhook delivery.
+	WebhooksService = "Webhooks"
 )
 
 // services is every service this endpoint registers, in a fixed order.
-var services = []string{EnvironmentObject, HostObject, ArtifactsObject, ArtifactPinObject}
+var services = []string{EnvironmentObject, HostObject, ArtifactsObject, ArtifactPinObject, WebhooksService}
 
 // Fleet is all the handlers may see of the FleetManager. kept narrow on
 // purpose: handlers go through these calls, never into the manager's vm map,
@@ -45,6 +48,7 @@ type Fleet interface {
 	Snapshots
 	Checkpoints
 	Tokens
+	Webhooks
 }
 
 // snapshotAbortTimeout is how long restate lets a snapshot, restore or
@@ -145,12 +149,19 @@ func NewEndpoint(cfg Config, fleet Fleet, logger *slog.Logger) *server.Restate {
 				return pinned(ctx)
 			}))
 
+	hooks := restate.NewService(WebhooksService, retryOption(cfg, WebhooksService)).
+		Handler("Deliver", restate.NewServiceHandler(
+			func(ctx restate.Context, d orchestrator.WebhookDelivery) (restate.Void, error) {
+				return restate.Void{}, deliverWebhook(ctx, fleet, d)
+			}))
+
 	srv := server.NewRestate().
 		WithLogger(logger.Handler(), true).
 		Bind(env).
 		Bind(host).
 		Bind(artifacts).
-		Bind(pins)
+		Bind(pins).
+		Bind(hooks)
 	// a request without a valid signature from this key gets a 401 before
 	// any handler runs. self mode leaves it unset: the endpoint is loopback.
 	if cfg.IdentityKey != "" {

internal/durable/server_test.go

@@ -109,6 +109,13 @@ func (fakeFleet) RotateTokenPersist(context.Context, string, orchestrator.Rotate
 	return nil
 }
 
+func (fakeFleet) WebhookPost(context.Context, orchestrator.WebhookDelivery) (orchestrator.WebhookAttempt, error) {
+	return orchestrator.WebhookAttempt{Delivered: true}, nil
+}
+func (fakeFleet) WebhookDeadLetter(context.Context, orchestrator.WebhookDelivery, string) error {
+	return nil
+}
+
 var testLogger = slog.New(slog.NewTextHandler(io.Discard, nil))
 
 // discovery is the manifest restate-server reads on registration. checking it
@@ -157,8 +164,12 @@ func TestDiscoveryManifest(t *testing.T) {
 
 	got := map[string]RetryPolicy{}
 	for _, s := range manifest.Services {
-		if s.Ty != "VIRTUAL_OBJECT" {
-			t.Errorf("%s: type %q, want VIRTUAL_OBJECT", s.Name, s.Ty)
+		wantTy := "VIRTUAL_OBJECT"
+		if s.Name == WebhooksService {
+			wantTy = "SERVICE"
+		}
+		if s.Ty != wantTy {
+			t.Errorf("%s: type %q, want %s", s.Name, s.Ty, wantTy)
 		}
 		handlers := map[string]string{}
 		for _, h := range s.Handlers {
@@ -174,6 +185,9 @@ func TestDiscoveryManifest(t *testing.T) {
 			want = map[string]string{"Move": "EXCLUSIVE"}
 		case ArtifactPinObject:
 			want = map[string]string{"Pin": "EXCLUSIVE", "Release": "EXCLUSIVE", "Expire": "EXCLUSIVE", "Pinned": "SHARED"}
+		case WebhooksService:
+			// a plain service's handlers carry no type.
+			want = map[string]string{"Deliver": ""}
 		}
 		if fmt.Sprint(handlers) != fmt.Sprint(want) {
 			t.Errorf("%s: handlers %v, want %v", s.Name, handlers, want)

internal/durable/webhooks.go

@@ -0,0 +1,88 @@
+package durable
+
+import (
+	"context"
+	"time"
+
+	restate "github.com/restatedev/sdk-go"
+	"github.com/restatedev/sdk-go/ingress"
+
+	"github.com/folsomintel/fuse/internal/orchestrator"
+)
+
+// opWebhook is the op label on webhook delivery's step spans and metrics.
+const opWebhook = "hitl_webhook"
+
+// Webhooks is the part of the fleet the webhook delivery handler drives, one
+// method per step. *FleetManager satisfies it.
+type Webhooks interface {
+	WebhookPost(ctx context.Context, d orchestrator.WebhookDelivery) (orchestrator.WebhookAttempt, error)
+	WebhookDeadLetter(ctx context.Context, d orchestrator.WebhookDelivery, lastErr string) error
+}
+
+// the retry schedule of a delivery: a failed attempt waits
+// webhookFirstBackoff, four times longer after each further failure, capped
+// at webhookMaxBackoff, until the next attempt would land past webhookWindow
+// from the first. vars so tests can shorten them.
+var (
+	webhookFirstBackoff = time.Second
+	webhookMaxBackoff   = time.Hour
+	webhookWindow       = 24 * time.Hour
+)
+
+// webhookBackoff is the wait after failed attempt n (counting from 1).
+func webhookBackoff(attempt int) time.Duration {
+	d := webhookFirstBackoff
+	for i := 1; i < attempt && d < webhookMaxBackoff; i++ {
+		d *= 4
+	}
+	return min(d, webhookMaxBackoff)
+}
+
+// deliverWebhook is Webhooks.Deliver: one attempt at one delivery. a failed
+// attempt does not sleep in the invocation; it sends the next attempt to
+// itself with a delay and ends, so a delivery that keeps failing for a day is
+// a chain of short invocations, each on the deployment current when it
+// starts. once the next attempt would fall outside the window, or the
+// endpoint no longer decrypts, the delivery is dead-lettered and ends.
+func deliverWebhook(ctx restate.Context, fleet Webhooks, d orchestrator.WebhookDelivery) error {
+	res, err := Run(ctx, opWebhook, "post", func(c context.Context) (orchestrator.WebhookAttempt, error) {
+		res, err := fleet.WebhookPost(c, d)
+		if orchestrator.WebhookEndpointSealed(err) {
+			return res, restate.ToTerminalError(err)
+		}
+		return res, err
+	})
+	lastErr := res.Error
+	switch {
+	case err != nil && !restate.IsTerminalError(err):
+		// suspension or cancellation: let restate handle it.
+		return err
+	case err != nil:
+		lastErr = err.Error()
+	case res.Delivered:
+		return nil
+	default:
+		wait := webhookBackoff(d.Attempt)
+		if res.At.Add(wait).Sub(d.FirstAt) <= webhookWindow {
+			next := d
+			next.Attempt++
+			restate.ServiceSend(ctx, WebhooksService, "Deliver").Send(next, restate.WithDelay(wait))
+			return nil
+		}
+	}
+	// the failure is recorded in the dead letter; the invocation itself is
+	// done.
+	return runStep(ctx, opWebhook, "dead_letter", func(c context.Context) error {
+		return fleet.WebhookDeadLetter(c, d, lastErr)
+	})
+}
+
+// SendWebhook enqueues the first attempt of d without waiting.
+func (c *Client) SendWebhook(ctx context.Context, d orchestrator.WebhookDelivery) error {
+	_, err := ingress.ServiceSend[orchestrator.WebhookDelivery](c.ingress, WebhooksService, "Deliver").
+		Send(ctx, d)
+	return err
+}
+
+var _ orchestrator.DurableWebhooks = (*Client)(nil)