diff --git a/docs/content/docs/cloud/monitoring.mdx b/docs/content/docs/cloud/monitoring.mdx index 68b4513..2efe753 100644 --- a/docs/content/docs/cloud/monitoring.mdx +++ b/docs/content/docs/cloud/monitoring.mdx @@ -126,6 +126,14 @@ 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 gates**: `orchestrator_hitl_decision_seconds{kind,outcome}` is the time +from a gate holding an action to a human (or the deadline) deciding it, +`orchestrator_hitl_decider_duration_seconds{verdict}` each gate decider call by +its answer (`allow`, `deny`, `ask`, or `error` when the call failed and the +action fell through to a human), and +`orchestrator_hitl_run_duration_seconds{action,result}` the time from an +approval to the held action's result, with restate on or off. + **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 diff --git a/docs/content/docs/cloud/orchestrator-deploy.mdx b/docs/content/docs/cloud/orchestrator-deploy.mdx index f20f824..3de6e26 100644 --- a/docs/content/docs/cloud/orchestrator-deploy.mdx +++ b/docs/content/docs/cloud/orchestrator-deploy.mdx @@ -71,7 +71,7 @@ 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__INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate 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)). | +| `FUSE_RESTATE_RETRY__INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `` is `ENVIRONMENT`, `HOST`, `ARTIFACTS`, `ARTIFACTPIN`, `WEBHOOKS` or `INTERVENTION`. 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)). The `INTERVENTION` policy covers restate failing to deliver an approved hitl action's run or a deadline timer; the run's own state store writes retry without bound. | 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 diff --git a/docs/content/docs/guides/human-in-the-loop.mdx b/docs/content/docs/guides/human-in-the-loop.mdx index 9836346..205d80a 100644 --- a/docs/content/docs/guides/human-in-the-loop.mdx +++ b/docs/content/docs/guides/human-in-the-loop.mdx @@ -298,6 +298,21 @@ approved action failed. `GET answers `403` with `details.reason: hitl_denied`, the gate in `details.gate`, and the gate's `reason` as the message. +**Across an orchestrator restart.** With restate off, an approved action the +orchestrator was running when it stopped is settled with `result.error` saying +it may or may not have run. With restate on (`FUSE_RESTATE_MODE=self` or +`cloud`), the approval is a restate `Intervention/{id}.Run` invocation and +the restarted orchestrator carries it on: an exec runs once (the host agent +replays a command it already finished instead of running it again; an exec +whose first attempt is more than an hour old gets the may-or-may-not error +instead, since the agent may have forgotten it), and a +destroy, snapshot or token rotation runs as that operation's own durable +handler. A computer action, drain, fork or migrate that was already under way +when the orchestrator died still gets the may-or-may-not-have-run error, so +it is never done twice. A waiting `?wait=` or SDK call gets the result once +it is stored. Each pending intervention's deadline is a restate timer too, so +`on_timeout` applies on time after a restart, even with nobody reading it. + The SDKs wait on a `202` for you: `exec` returns the approved command's result, `computer` the action's output and screenshot, `fork` the new environment, and so on. A rejection raises diff --git a/internal/durable/hitl.go b/internal/durable/hitl.go new file mode 100644 index 0000000..de75067 --- /dev/null +++ b/internal/durable/hitl.go @@ -0,0 +1,147 @@ +package durable + +import ( + "context" + "errors" + "time" + + restate "github.com/restatedev/sdk-go" + "github.com/restatedev/sdk-go/ingress" + + "github.com/folsomintel/fuse/internal/hitl" + "github.com/folsomintel/fuse/internal/orchestrator" +) + +// opHitlRun and opHitlExpire are the op labels on the Intervention object's +// step spans and metrics. +const ( + opHitlRun = "hitl_run" + opHitlExpire = "hitl_expire" +) + +// hitlRunAbortTimeout is how long restate lets one attempt at an approved +// action run without journal progress before it aborts and retries it: the +// longest action, a live migrate, plus a margin. +const hitlRunAbortTimeout = orchestrator.HitlRunTimeout + 5*time.Minute + +// Hitl is the part of the fleet the Intervention object drives, one method +// per step. *FleetManager satisfies it. +type Hitl interface { + HitlRunBegin(ctx context.Context, id string) (orchestrator.HitlRunPlan, error) + HitlRunInProcess(ctx context.Context, id string) error + HitlRunFinish(ctx context.Context, id string, plan orchestrator.HitlRunPlan, res hitl.Result) error + HitlExpire(ctx context.Context, id string) (time.Duration, error) +} + +// hitlPermanent makes an intervention that no longer exists terminal: no +// retry brings it back. +func hitlPermanent(err error) error { + if errors.Is(err, orchestrator.ErrInterventionNotFound) { + return restate.ToTerminalError(err) + } + return err +} + +// unbounded retries a step until it succeeds, under policy's backoff. the +// run's steps write the intervention row, and a state store that is down +// holds the run there until it is back rather than leaving it unsettled. +func unbounded(policy RetryPolicy) []restate.RunOption { + return []restate.RunOption{ + restate.WithInitialRetryInterval(policy.InitialInterval), + restate.WithMaxRetryInterval(policy.MaxInterval), + } +} + +// hitlRun is Intervention/{id}.Run: the approved action, once. the +// decision itself lives on the intervention row, where DecideIntervention +// or the deadline writes it; the orchestrator sends Run once it is an +// approval, keyed on the id, so there is no invocation waiting on a human. +// +// begin picks the route. an exec, computer, drain, fork or migrate runs in +// one in-process step that writes its own result. destroy, snapshot and +// token rotation are durable handlers on Environment/{env}, so the run calls +// them there and writes what they returned; their steps are not run again +// here. +func hitlRun(ctx restate.ObjectContext, fleet Hitl, policy RetryPolicy) error { + id := restate.Key(ctx) + retry := unbounded(policy) + plan, err := Run(ctx, opHitlRun, "begin", func(c context.Context) (orchestrator.HitlRunPlan, error) { + plan, err := fleet.HitlRunBegin(c, id) + return plan, hitlPermanent(err) + }, retry...) + if err != nil { + return err + } + + var res hitl.Result + switch plan.Route { + case orchestrator.HitlRouteNone: + return nil + case orchestrator.HitlRouteInProcess: + return runStep(ctx, opHitlRun, "run", func(c context.Context) error { + return hitlPermanent(fleet.HitlRunInProcess(c, id)) + }, retry...) + case orchestrator.HitlRouteDestroy: + _, err = restate.Object[restate.Void](ctx, EnvironmentObject, plan.EnvID, "Destroy"). + Request(orchestrator.DestroyRequest{Reason: orchestrator.DestroyReasonManual}) + case orchestrator.HitlRouteSnapshot: + var rec orchestrator.SnapshotRecord + rec, err = restate.Object[orchestrator.SnapshotRecord](ctx, EnvironmentObject, plan.EnvID, "Snapshot"). + Request(*plan.Snapshot) + res.SnapshotID = rec.SnapshotID + case orchestrator.HitlRouteRotateToken: + _, err = restate.Object[restate.Void](ctx, EnvironmentObject, plan.EnvID, "RotateToken"). + Request(restate.Void{}) + default: + err = restate.TerminalErrorf("unknown hitl run route %q", plan.Route) + } + if err != nil { + if !restate.IsTerminalError(err) { + // suspension or cancellation: let restate handle it. + return err + } + res = hitl.Result{Error: err.Error()} + } + return runStep(ctx, opHitlRun, "finish", func(c context.Context) error { + return hitlPermanent(fleet.HitlRunFinish(c, id, plan, res)) + }, retry...) +} + +// hitlExpire is Intervention/{id}.Expire, the durable timer of a pending +// intervention's deadline: the orchestrator sends it with the deadline as +// its delay when the intervention is raised. it applies on_timeout, and an +// approval that makes is run like a human's. a timer that fires early by +// the orchestrator's clock sends itself again for the rest; one that finds +// the intervention already decided does nothing. +func hitlExpire(ctx restate.ObjectContext, fleet Hitl, policy RetryPolicy) error { + id := restate.Key(ctx) + left, err := Run(ctx, opHitlExpire, "expire", func(c context.Context) (time.Duration, error) { + return fleet.HitlExpire(c, id) + }, unbounded(policy)...) + if err != nil { + return err + } + if left > 0 { + restate.ObjectSend(ctx, InterventionObject, id, "Expire").Send(restate.Void{}, restate.WithDelay(left)) + } + return nil +} + +// SendHitlRun enqueues Intervention/{id}.Run. the idempotency key is the id, +// so a second send while one is in flight, or within the retention window +// after it finished, attaches to it instead of starting another; one after +// that finds the result on the row and does nothing. +func (c *Client) SendHitlRun(ctx context.Context, id string) error { + _, err := ingress.ObjectSend[restate.Void](c.ingress, InterventionObject, id, "Run"). + Send(ctx, restate.Void{}, restate.WithIdempotencyKey("hitl-run:"+id)) + return err +} + +// SendHitlExpire enqueues Intervention/{id}.Expire to run after delay. +func (c *Client) SendHitlExpire(ctx context.Context, id string, delay time.Duration) error { + _, err := ingress.ObjectSend[restate.Void](c.ingress, InterventionObject, id, "Expire"). + Send(ctx, restate.Void{}, restate.WithDelay(delay)) + return err +} + +var _ orchestrator.DurableHitl = (*Client)(nil) diff --git a/internal/durable/hitl_test.go b/internal/durable/hitl_test.go new file mode 100644 index 0000000..cca6a2d --- /dev/null +++ b/internal/durable/hitl_test.go @@ -0,0 +1,358 @@ +package durable + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/folsomintel/fuse/internal/hitl" + "github.com/folsomintel/fuse/internal/hostwire" + "github.com/folsomintel/fuse/internal/orchestrator" +) + +// hitlAgent stands in for a host agent's exec endpoint: it runs a command +// once per idempotency key and replays the first answer to a repeat, as +// fc-agent does. +type hitlAgent struct { + mu sync.Mutex + runs int + keys []string + seen map[string][]byte + srv *httptest.Server +} + +func newHitlAgent(t *testing.T) *hitlAgent { + a := &hitlAgent{seen: map[string][]byte{}} + a.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key := r.Header.Get(hostwire.IdempotencyHeader) + a.mu.Lock() + defer a.mu.Unlock() + a.keys = append(a.keys, key) + out, ok := a.seen[key] + if !ok || key == "" { + a.runs++ + out = []byte(fmt.Sprintf(`{"stdout":"secret-output-%d"}`, a.runs)) + a.seen[key] = out + } + _, _ = w.Write(out) + })) + t.Cleanup(a.srv.Close) + return a +} + +func (a *hitlAgent) state() (int, []string) { + a.mu.Lock() + defer a.mu.Unlock() + return a.runs, append([]string(nil), a.keys...) +} + +// hitlRows is the intervention rows the fleets of an endpoint and its +// restart share. +type hitlRows struct { + mu sync.Mutex + plans map[string]orchestrator.HitlRunPlan + results map[string]hitl.Result + expired map[string]int + settled chan string +} + +func newHitlRows() *hitlRows { + return &hitlRows{ + plans: map[string]orchestrator.HitlRunPlan{}, + results: map[string]hitl.Result{}, + expired: map[string]int{}, + settled: make(chan string, 16), + } +} + +func (r *hitlRows) result(id string) (hitl.Result, bool) { + r.mu.Lock() + defer r.mu.Unlock() + res, ok := r.results[id] + return res, ok +} + +func (r *hitlRows) store(id string, res hitl.Result) { + r.mu.Lock() + r.results[id] = res + r.mu.Unlock() + r.settled <- id +} + +// hitlFleet scripts the Intervention object's steps over hitlRows. an +// in-process run is an exec on the hitlAgent; block, when set, holds the run +// after the agent answered and before the result is written. +type hitlFleet struct { + *snapFleet + rows *hitlRows + agent *hitlAgent + client *http.Client + block chan struct{} + ran chan struct{} + mu sync.Mutex + calls []string + earlyBy time.Duration +} + +func newHitlFleet(rows *hitlRows, agent *hitlAgent) *hitlFleet { + return &hitlFleet{snapFleet: newSnapFleet(), rows: rows, agent: agent, client: hostwire.NewClient(), ran: make(chan struct{}, 4)} +} + +func (f *hitlFleet) record(call string) { + f.mu.Lock() + f.calls = append(f.calls, call) + f.mu.Unlock() +} + +func (f *hitlFleet) seen() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.calls...) +} + +func (f *hitlFleet) HitlRunBegin(_ context.Context, id string) (orchestrator.HitlRunPlan, error) { + f.record("begin") + if _, ok := f.rows.result(id); ok { + return orchestrator.HitlRunPlan{}, nil + } + f.rows.mu.Lock() + defer f.rows.mu.Unlock() + return f.rows.plans[id], nil +} + +func (f *hitlFleet) HitlRunInProcess(ctx context.Context, id string) error { + f.record("run") + if _, ok := f.rows.result(id); ok { + return nil + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, f.agent.srv.URL+"/v1/vm/x/exec", bytes.NewReader([]byte(`{}`))) + if err != nil { + return err + } + res, err := f.client.Do(req) + if err != nil { + return err + } + raw, _ := io.ReadAll(res.Body) + _ = res.Body.Close() + var out struct { + Stdout string `json:"stdout"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return err + } + f.ran <- struct{}{} + if f.block != nil { + select { + case <-f.block: + case <-ctx.Done(): + return ctx.Err() + } + } + f.rows.store(id, hitl.Result{Exec: &hitl.ExecResult{Stdout: out.Stdout}}) + return nil +} + +func (f *hitlFleet) HitlRunFinish(_ context.Context, id string, _ orchestrator.HitlRunPlan, res hitl.Result) error { + f.record("finish") + if _, ok := f.rows.result(id); ok { + return nil + } + f.rows.store(id, res) + return nil +} + +func (f *hitlFleet) HitlExpire(_ context.Context, id string) (time.Duration, error) { + f.record("expire") + f.mu.Lock() + early := f.earlyBy + f.earlyBy = 0 + f.mu.Unlock() + if early > 0 { + return early, nil + } + f.rows.mu.Lock() + f.rows.expired[id]++ + f.rows.mu.Unlock() + f.rows.settled <- id + return 0, nil +} + +func uniqueIntervention(name string) string { + return fmt.Sprintf("int-%s-%d", name, time.Now().UnixNano()) +} + +func waitSettled(t *testing.T, ctx context.Context, rows *hitlRows, id string) { + t.Helper() + for { + select { + case got := <-rows.settled: + if got == id { + return + } + case <-ctx.Done(): + t.Fatalf("intervention %s never settled", id) + } + } +} + +// TestHitlRunSurvivesRestartBeforeRun stands in for kill -9 of the +// orchestrator right after an approval was sent and before the run started: +// restate holds the run while the endpoint is gone and delivers it to the +// restarted one, which runs it once. a second send for the same approval, +// as the restart's resume pass makes, starts nothing. +func TestHitlRunSurvivesRestartBeforeRun(t *testing.T) { + rows := newHitlRows() + agent := newHitlAgent(t) + id := uniqueIntervention("before") + rows.plans[id] = orchestrator.HitlRunPlan{Route: orchestrator.HitlRouteInProcess, EnvID: "fuse-a"} + + first := newHitlFleet(rows, agent) + client, addr, stop := serveFleetAt(t, first, "") + stop() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := client.SendHitlRun(ctx, id); err != nil { + t.Fatalf("send: %v", err) + } + second := newHitlFleet(rows, agent) + _, _, _ = serveFleetAt(t, second, addr) + waitSettled(t, ctx, rows, id) + + if err := client.SendHitlRun(ctx, id); err != nil { + t.Fatalf("second send: %v", err) + } + time.Sleep(500 * time.Millisecond) + runs, _ := agent.state() + res, _ := rows.result(id) + if runs != 1 || res.Exec == nil || res.Exec.Stdout != "secret-output-1" { + t.Fatalf("agent ran %d times, result %+v; want one run and its output", runs, res) + } + if got := first.seen(); len(got) != 0 { + t.Fatalf("the stopped endpoint ran %v", got) + } + if got := fmt.Sprint(second.seen()); got != "[begin run]" { + t.Fatalf("restarted endpoint ran %s, want one begin and one run", got) + } +} + +// TestHitlRunExecNotRerunAfterRestart stands in for kill -9 while a held exec +// runs: the agent has run it, the orchestrator dies before writing the +// result, and the restarted endpoint retries the step. the retry carries the +// same idempotency key, so the agent replays its first answer and the command +// runs once. while the run is held, the journal has no trace of the output. +func TestHitlRunExecNotRerunAfterRestart(t *testing.T) { + rows := newHitlRows() + agent := newHitlAgent(t) + id := uniqueIntervention("during") + rows.plans[id] = orchestrator.HitlRunPlan{Route: orchestrator.HitlRouteInProcess, EnvID: "fuse-a"} + + first := newHitlFleet(rows, agent) + first.block = make(chan struct{}) + defer close(first.block) + client, addr, stop := serveFleetAt(t, first, "") + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := client.SendHitlRun(ctx, id); err != nil { + t.Fatalf("send: %v", err) + } + select { + case <-first.ran: + case <-ctx.Done(): + t.Fatal("the exec never ran") + } + journal := readJournalOf(t, ctx, InterventionObject, id, "begin") + if bytes.Contains(journal, []byte("secret-output")) { + t.Fatalf("journal of %s holds the exec output", id) + } + + stop() + second := newHitlFleet(rows, agent) + _, _, _ = serveFleetAt(t, second, addr) + waitSettled(t, ctx, rows, id) + + runs, keys := agent.state() + res, _ := rows.result(id) + if runs != 1 || res.Exec == nil || res.Exec.Stdout != "secret-output-1" { + t.Fatalf("agent ran %d times, result %+v; want one run and its output", runs, res) + } + if len(keys) != 2 || keys[0] == "" || keys[0] != keys[1] { + t.Fatalf("idempotency keys %q, want the same key on the run and its retry", keys) + } + if got := fmt.Sprint(second.seen()); got != "[run]" { + t.Fatalf("restarted endpoint ran %s, want only the run step", got) + } +} + +// TestHitlExpireSurvivesRestart: a deadline timer armed before a restart +// fires on the restarted endpoint, and one that fires early by the +// orchestrator's clock arms itself again for the rest. +func TestHitlExpireSurvivesRestart(t *testing.T) { + rows := newHitlRows() + id := uniqueIntervention("expire") + first := newHitlFleet(rows, nil) + client, addr, stop := serveFleetAt(t, first, "") + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := client.SendHitlExpire(ctx, id, time.Second); err != nil { + t.Fatalf("send: %v", err) + } + stop() + time.Sleep(1500 * time.Millisecond) + second := newHitlFleet(rows, nil) + second.earlyBy = 300 * time.Millisecond + _, _, _ = serveFleetAt(t, second, addr) + waitSettled(t, ctx, rows, id) + + rows.mu.Lock() + n := rows.expired[id] + rows.mu.Unlock() + if n != 1 { + t.Fatalf("expired %d times, want once", n) + } + if got := first.seen(); len(got) != 0 { + t.Fatalf("the stopped endpoint ran %v", got) + } + if got := fmt.Sprint(second.seen()); got != "[expire expire]" { + t.Fatalf("restarted endpoint ran %s, want the early timer and its re-arm", got) + } +} + +// TestHitlRunSnapshotCallsEnvironment: an approved snapshot is taken by the +// Environment object's own handler, and the run writes the snapshot id it +// returned. +func TestHitlRunSnapshotCallsEnvironment(t *testing.T) { + rows := newHitlRows() + id := uniqueIntervention("snap") + rows.plans[id] = orchestrator.HitlRunPlan{Route: orchestrator.HitlRouteSnapshot, EnvID: uniqueVM("hitl-snap"), + Snapshot: &orchestrator.SnapshotOptions{Comment: "approved"}} + fleet := newHitlFleet(rows, nil) + client := serveFleet(t, fleet, fastPolicy) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := client.SendHitlRun(ctx, id); err != nil { + t.Fatalf("send: %v", err) + } + waitSettled(t, ctx, rows, id) + res, _ := rows.result(id) + if res.Error != "" || res.SnapshotID != "snap-1" { + t.Fatalf("result %+v, want snap-1", res) + } + if got, want := fleet.steps(), "begin,checkpoint,record,list,ready"; got != want { + t.Fatalf("snapshot steps %s, want %s", got, want) + } + if got := fmt.Sprint(fleet.seen()); got != "[begin finish]" { + t.Fatalf("run steps %s, want begin and finish", got) + } +} diff --git a/internal/durable/server.go b/internal/durable/server.go index 6340709..5d1a5eb 100644 --- a/internal/durable/server.go +++ b/internal/durable/server.go @@ -31,10 +31,13 @@ const ( // WebhooksService is a plain service: each Deliver invocation is one // attempt at one hitl webhook delivery. WebhooksService = "Webhooks" + // InterventionObject is keyed by hitl intervention id: the run of an + // approved action and the timer of its deadline. + InterventionObject = "Intervention" ) // services is every service this endpoint registers, in a fixed order. -var services = []string{EnvironmentObject, HostObject, ArtifactsObject, ArtifactPinObject, WebhooksService} +var services = []string{EnvironmentObject, HostObject, ArtifactsObject, ArtifactPinObject, WebhooksService, InterventionObject} // 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, @@ -49,6 +52,7 @@ type Fleet interface { Checkpoints Tokens Webhooks + Hitl } // snapshotAbortTimeout is how long restate lets a snapshot, restore or @@ -155,13 +159,24 @@ func NewEndpoint(cfg Config, fleet Fleet, logger *slog.Logger) *server.Restate { return restate.Void{}, deliverWebhook(ctx, fleet, d) })) + interventions := restate.NewObject(InterventionObject, retryOption(cfg, InterventionObject)). + Handler("Run", restate.NewObjectHandler( + func(ctx restate.ObjectContext, _ restate.Void) (restate.Void, error) { + return restate.Void{}, hitlRun(ctx, fleet, servicePolicy(cfg, InterventionObject)) + }, restate.WithAbortTimeout(hitlRunAbortTimeout))). + Handler("Expire", restate.NewObjectHandler( + func(ctx restate.ObjectContext, _ restate.Void) (restate.Void, error) { + return restate.Void{}, hitlExpire(ctx, fleet, servicePolicy(cfg, InterventionObject)) + })) + srv := server.NewRestate(). WithLogger(logger.Handler(), true). Bind(env). Bind(host). Bind(artifacts). Bind(pins). - Bind(hooks) + Bind(hooks). + Bind(interventions) // 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 != "" { diff --git a/internal/durable/server_test.go b/internal/durable/server_test.go index 24f54ca..a17385f 100644 --- a/internal/durable/server_test.go +++ b/internal/durable/server_test.go @@ -20,6 +20,7 @@ import ( "github.com/mr-tron/base58" "github.com/restatedev/sdk-go/server" + "github.com/folsomintel/fuse/internal/hitl" "github.com/folsomintel/fuse/internal/orchestrator" ) @@ -116,6 +117,15 @@ func (fakeFleet) WebhookDeadLetter(context.Context, orchestrator.WebhookDelivery return nil } +func (fakeFleet) HitlRunBegin(context.Context, string) (orchestrator.HitlRunPlan, error) { + return orchestrator.HitlRunPlan{}, nil +} +func (fakeFleet) HitlRunInProcess(context.Context, string) error { return nil } +func (fakeFleet) HitlRunFinish(context.Context, string, orchestrator.HitlRunPlan, hitl.Result) error { + return nil +} +func (fakeFleet) HitlExpire(context.Context, string) (time.Duration, error) { return 0, nil } + var testLogger = slog.New(slog.NewTextHandler(io.Discard, nil)) // discovery is the manifest restate-server reads on registration. checking it @@ -185,6 +195,8 @@ 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 InterventionObject: + want = map[string]string{"Run": "EXCLUSIVE", "Expire": "EXCLUSIVE"} case WebhooksService: // a plain service's handlers carry no type. want = map[string]string{"Deliver": ""} diff --git a/internal/durable/token_test.go b/internal/durable/token_test.go index 0f323f6..52ef99e 100644 --- a/internal/durable/token_test.go +++ b/internal/durable/token_test.go @@ -233,10 +233,17 @@ func TestRotateTokenSurvivesEndpointRestart(t *testing.T) { // restate's sql introspection, with every payload (rendered as an array of // bytes) decoded back into its bytes. func readJournal(t *testing.T, ctx context.Context, vm string) []byte { + t.Helper() + return readJournalOf(t, ctx, EnvironmentObject, vm, "generate") +} + +// readJournalOf is readJournal for any object key, checking that the entry +// named want is there. +func readJournalOf(t *testing.T, ctx context.Context, service, key, want string) []byte { t.Helper() admin, _ := restateTestURLs(t) query := fmt.Sprintf("SELECT j.name, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id "+ - "WHERE i.target_service_name = '%s' AND i.target_service_key = '%s'", EnvironmentObject, vm) + "WHERE i.target_service_name = '%s' AND i.target_service_key = '%s'", service, key) body, _ := json.Marshal(map[string]string{"query": query}) req, _ := http.NewRequestWithContext(ctx, http.MethodPost, admin+"/query", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") @@ -269,8 +276,8 @@ func readJournal(t *testing.T, ctx context.Context, vm string) []byte { } all = appendPayloads(all, entry) } - if !strings.Contains(strings.Join(names, ","), "generate") { - t.Fatalf("journal of %s has entries %v, want the generate step", vm, names) + if !strings.Contains(strings.Join(names, ","), want) { + t.Fatalf("journal of %s has entries %v, want the %s step", key, names, want) } return all } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index f28800a..2729a9b 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -65,6 +65,11 @@ type PrometheusMetrics struct { stepDuration *prometheus.HistogramVec stepRetries *prometheus.CounterVec + // hitl gated actions, through orchestrator.HitlMetrics. + hitlDecision *prometheus.HistogramVec + hitlDeciderLatency *prometheus.HistogramVec + hitlRunDuration *prometheus.HistogramVec + // HTTP handler metrics (used by the middleware). HTTPRequestsTotal *prometheus.CounterVec HTTPRequestDuration *prometheus.HistogramVec @@ -273,6 +278,28 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics { Help: "Durable (restate) step executions that failed with a retryable error, by op and step.", }, []string{"op", "step"}), + hitlDecision: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "orchestrator", + Subsystem: "hitl", + Name: "decision_seconds", + Help: "Time from a gate holding an action to its decision by a human or the deadline, by kind and outcome.", + Buckets: []float64{1, 5, 15, 30, 60, 120, 300, 600, 1800, 3600, 4 * 3600, 24 * 3600}, + }, []string{"kind", "outcome"}), + hitlDeciderLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "orchestrator", + Subsystem: "hitl", + Name: "decider_duration_seconds", + Help: "Gate decider call latency by verdict (allow, deny, ask, or error when the call failed).", + Buckets: []float64{.05, .1, .25, .5, 1, 2, 5, 10}, + }, []string{"verdict"}), + hitlRunDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "orchestrator", + Subsystem: "hitl", + Name: "run_duration_seconds", + Help: "Time from approving a gated action to its result, by action and result (ok or error).", + Buckets: []float64{.1, .5, 1, 5, 10, 30, 60, 300, 600, 1800}, + }, []string{"action", "result"}), + HTTPRequestsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "orchestrator", Subsystem: "http", @@ -326,6 +353,9 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics { m.operationDuration, m.stepDuration, m.stepRetries, + m.hitlDecision, + m.hitlDeciderLatency, + m.hitlRunDuration, m.HTTPRequestsTotal, m.HTTPRequestDuration, m.HTTPRequestsInFlight, @@ -429,7 +459,27 @@ func (m *PrometheusMetrics) StepRetried(op, step string) { m.stepRetries.WithLabelValues(op, step).Inc() } +// HitlDecided implements orchestrator.HitlMetrics. +func (m *PrometheusMetrics) HitlDecided(kind, outcome string, d time.Duration) { + m.hitlDecision.WithLabelValues(kind, outcome).Observe(d.Seconds()) +} + +// HitlDeciderAsked implements orchestrator.HitlMetrics. +func (m *PrometheusMetrics) HitlDeciderAsked(verdict string, d time.Duration) { + m.hitlDeciderLatency.WithLabelValues(verdict).Observe(d.Seconds()) +} + +// HitlRunFinished implements orchestrator.HitlMetrics. +func (m *PrometheusMetrics) HitlRunFinished(action string, d time.Duration, failed bool) { + result := "ok" + if failed { + result = "error" + } + m.hitlRunDuration.WithLabelValues(action, result).Observe(d.Seconds()) +} + var ( + _ orchestrator.HitlMetrics = (*PrometheusMetrics)(nil) _ orchestrator.EgressMetrics = (*PrometheusMetrics)(nil) _ orchestrator.OperationMetrics = (*PrometheusMetrics)(nil) _ orchestrator.ArtifactMoveMetrics = (*PrometheusMetrics)(nil) diff --git a/internal/orchestrator/fleet.go b/internal/orchestrator/fleet.go index b0e3db7..2422894 100644 --- a/internal/orchestrator/fleet.go +++ b/internal/orchestrator/fleet.go @@ -10,9 +10,11 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "github.com/folsomintel/fuse/internal/egress" + "github.com/folsomintel/fuse/internal/hitl" "github.com/folsomintel/fuse/internal/secrets" "github.com/folsomintel/fuse/internal/telemetry" "go.opentelemetry.io/otel/attribute" @@ -645,6 +647,21 @@ type FleetManager struct { webhookMu sync.Mutex webhookBacklog map[string]map[string]struct{} + // durableHitlRuns runs approved hitl actions and intervention deadlines + // on restate; nil in mode off. see hitl_durable.go. hitlSendSlots + // bounds the one-way sends in flight and hitlSends tracks them. + // hitlResend asks the next reconcile to send every approved action + // still owed a result again, after a send failed. hitlUnstored holds a + // result whose write failed, for the step's retry; guarded by hitlMu. + durableHitlRuns DurableHitl + hitlSendSlots chan struct{} + hitlSends sync.WaitGroup + hitlResend atomic.Bool + hitlUnstored map[string]hitl.Result + // hitlQueued is the interventions with a run send queued or in flight, + // guarded by hitlMu. + hitlQueued map[string]bool + cancel context.CancelFunc done chan struct{} } diff --git a/internal/orchestrator/gate_run.go b/internal/orchestrator/gate_run.go index 4c09e09..fc035d7 100644 --- a/internal/orchestrator/gate_run.go +++ b/internal/orchestrator/gate_run.go @@ -169,6 +169,9 @@ func (fm *FleetManager) storeRunResult(ctx context.Context, id string, res hitl. } fm.hitlMu.Unlock() + if m := fm.hitlMetrics(); m != nil && in.Decision != nil { + m.HitlRunFinished(runAction(in), res.FinishedAt.Sub(in.Decision.DecidedAt), res.Error != "") + } fm.appendEvent(ctx, "intervention", id, "intervention.ran", map[string]any{ "env_id": in.EnvID, "error": res.Error != "", }) diff --git a/internal/orchestrator/gates.go b/internal/orchestrator/gates.go index 56d48f4..083d9e8 100644 --- a/internal/orchestrator/gates.go +++ b/internal/orchestrator/gates.go @@ -164,7 +164,15 @@ func (fm *FleetManager) applyVerdict(ctx context.Context, envID string, spec hit } in := open(g) if g.Decide != nil { + start := time.Now() res, err := fm.askDecider(ctx, g, in) + if m := fm.hitlMetrics(); m != nil { + verdict := string(res.Decision) + if err != nil { + verdict = "error" + } + m.HitlDeciderAsked(verdict, time.Since(start)) + } switch { case err != nil: fm.logger.Warn("hitl decider failed; holding for a human", "vm", envID, "gate", g.Name, "err", err) @@ -225,6 +233,7 @@ func (fm *FleetManager) recordGateIntervention(ctx context.Context, in hitl.Inte if in.State.Terminal() { fm.auditIntervention(ctx, in, auditMeta{}) } + fm.scheduleExpiry(in) return nil } diff --git a/internal/orchestrator/hitl_durable.go b/internal/orchestrator/hitl_durable.go new file mode 100644 index 0000000..b76ac4b --- /dev/null +++ b/internal/orchestrator/hitl_durable.go @@ -0,0 +1,417 @@ +package orchestrator + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/folsomintel/fuse/internal/hitl" +) + +// maxHitlSends bounds the one-way hitl sends in flight, so a restate that +// accepts connections and never answers holds at most this many. +const maxHitlSends = 8 + +// errRunInterrupted is the result an approved action gets when a previous +// attempt at it began and never reported back: it may or may not have run, +// and running it again could do it twice. +const errRunInterrupted = "the orchestrator restarted while the approved action was running; it may or may not have run" + +// DurableHitl runs approved hitl actions and intervention deadlines on the +// restate Intervention/{id} object. *durable.Client satisfies it. both only +// enqueue. +type DurableHitl interface { + // SendHitlRun enqueues Intervention/{id}.Run, keyed on id, so sending it + // again while one is in flight or done starts nothing new. + SendHitlRun(ctx context.Context, id string) error + // SendHitlExpire enqueues Intervention/{id}.Expire to run after delay. + SendHitlExpire(ctx context.Context, id string, delay time.Duration) error +} + +// HitlMetrics is an optional extension of ReconcileMetrics for gated +// actions. +type HitlMetrics interface { + // HitlDecided is a gated action decided by a human or its deadline, + // d after the gate held it. + HitlDecided(kind, outcome string, d time.Duration) + // HitlDeciderAsked is one call to a gate's decider. verdict is the + // decider's answer, or "error". + HitlDeciderAsked(verdict string, d time.Duration) + // HitlRunFinished is an approved action's result, d after the approval. + // action is exec, computer or the lifecycle action. + HitlRunFinished(action string, d time.Duration, failed bool) +} + +// SetDurableHitl routes approved hitl actions and intervention deadlines +// through restate. nil (mode off, the default) keeps the in-process run and +// the orphan pass. set it before Start. +func (fm *FleetManager) SetDurableHitl(d DurableHitl) { + fm.mu.Lock() + fm.durableHitlRuns = d + if fm.hitlSendSlots == nil { + fm.hitlSendSlots = make(chan struct{}, maxHitlSends) + } + fm.mu.Unlock() +} + +func (fm *FleetManager) durableHitl() DurableHitl { + fm.mu.RLock() + defer fm.mu.RUnlock() + return fm.durableHitlRuns +} + +// needsRun reports whether in is an approved gated action with no result yet. +func needsRun(in hitl.Intervention) bool { + return in.Runs() && in.Decision != nil && in.Decision.Outcome == hitl.OutcomeApprove && in.Result == nil +} + +// hitlSend runs send off the caller's goroutine, at most maxHitlSends at a +// time. +func (fm *FleetManager) hitlSend(what, id string, send func(context.Context) error) { + fm.hitlSends.Add(1) + go func() { + defer fm.hitlSends.Done() + fm.hitlSendSlots <- struct{}{} + defer func() { <-fm.hitlSendSlots }() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := send(ctx); err != nil { + fm.logger.Error("send durable hitl "+what+" failed", "intervention", id, "err", err) + if what == "run" { + // the next reconcile sends every approved action still + // owed a result again. + fm.hitlResend.Store(true) + } + } + }() +} + +// sendHitlRun hands an approved action to restate. an intervention whose +// send is already queued or in flight is not queued again, so a restate +// outage holds at most one waiting send per approval owed a result. +func (fm *FleetManager) sendHitlRun(d DurableHitl, id string) { + fm.hitlMu.Lock() + if fm.hitlQueued[id] { + fm.hitlMu.Unlock() + return + } + if fm.hitlQueued == nil { + fm.hitlQueued = make(map[string]bool) + } + fm.hitlQueued[id] = true + fm.hitlMu.Unlock() + fm.hitlSend("run", id, func(ctx context.Context) error { + defer func() { + fm.hitlMu.Lock() + delete(fm.hitlQueued, id) + fm.hitlMu.Unlock() + }() + return d.SendHitlRun(ctx, id) + }) +} + +// scheduleExpiry arms the durable timer that applies a pending +// intervention's on_timeout at its deadline. a delayed send, not a sleeping +// invocation, so a day-long deadline pins nothing to a deployment. a send +// that fails is covered by the reconcile expiry sweep. +func (fm *FleetManager) scheduleExpiry(in hitl.Intervention) { + d := fm.durableHitl() + if d == nil || in.State != hitl.StatePending { + return + } + // the delay is taken when the send goes out, so time spent waiting for + // a slot does not push the timer past the deadline. + fm.hitlSend("expire", in.ID, func(ctx context.Context) error { + return d.SendHitlExpire(ctx, in.ID, max(time.Until(in.Deadline), 0)) + }) +} + +// resumeApprovedRuns sends every approved action still owed a result to +// restate again: one whose send was lost, or whose approval landed just +// before a crash. the send is keyed on the intervention, so one already in +// flight is not started twice. it replaces recoverOrphanRuns in modes self +// and cloud: restate resumes the run instead of the result saying it may or +// may not have happened. +func (fm *FleetManager) resumeApprovedRuns(ctx context.Context, d DurableHitl) { + fm.hitlResend.Store(false) + for _, state := range []hitl.State{hitl.StateApproved, hitl.StateExpired} { + recs, err := fm.store.ListInterventions(ctx, InterventionFilter{State: string(state)}) + if err != nil { + fm.logger.Warn("list approved interventions failed", "err", err) + fm.hitlResend.Store(true) + return + } + for _, rec := range recs { + if rec.Outcome != string(hitl.OutcomeApprove) { + continue + } + in, err := fm.interventionFromRecord(rec) + if err != nil || !needsRun(in) { + continue + } + fm.sendHitlRun(d, in.ID) + } + } +} + +// HitlRunRoute is how the durable run carries out an approved action. +type HitlRunRoute string + +const ( + // HitlRouteNone: nothing is left to run. + HitlRouteNone HitlRunRoute = "" + // HitlRouteInProcess runs the action in this process, in one step: + // exec, computer, drain, fork and migrate. + HitlRouteInProcess HitlRunRoute = "in_process" + // the rest are durable already, so the run calls their handler on the + // environment's object instead of running their steps itself. + HitlRouteDestroy HitlRunRoute = "destroy" + HitlRouteSnapshot HitlRunRoute = "snapshot" + HitlRouteRotateToken HitlRunRoute = "rotate_token" +) + +// HitlRunPlan is what HitlRunBegin settled. restate journals it, so it holds +// no argv, text or output: only the route, the environment and, for a +// snapshot, its options. +// +// a route other than in process is also the lifecycle operation it runs, +// and StartedAt when the run began, so HitlRunFinish can record it with the +// other operations. +type HitlRunPlan struct { + Route HitlRunRoute `json:"route"` + EnvID string `json:"env_id,omitempty"` + Snapshot *SnapshotOptions `json:"snapshot,omitempty"` + StartedAt time.Time `json:"started_at,omitempty"` +} + +// the methods below are the steps of the durable Intervention/{id}.Run and +// Expire handlers. each runs inside a journaled restate step and must be +// safe to run again. + +// HitlRunBegin reads the approved action and picks its route. an action +// with a result already, or one not approved, has nothing left to run. a +// destroy of an environment the fleet no longer tracks, or snapshot params +// that do not decode, get their error result here. +func (fm *FleetManager) HitlRunBegin(ctx context.Context, id string) (HitlRunPlan, error) { + in, err := fm.loadIntervention(ctx, id) + if err != nil { + return HitlRunPlan{}, err + } + if !needsRun(in) { + return HitlRunPlan{}, nil + } + if in.Lifecycle == nil { + return HitlRunPlan{Route: HitlRouteInProcess, EnvID: in.EnvID}, nil + } + plan := HitlRunPlan{EnvID: in.EnvID, StartedAt: time.Now()} + switch in.Lifecycle.Action { + case hitl.ActionDestroy: + if _, ok := fm.GetVM(in.EnvID); !ok { + return HitlRunPlan{}, fm.storeRunResult(ctx, id, hitl.Result{Error: fmt.Sprintf("%v: %s", ErrVMNotFound, in.EnvID), FinishedAt: time.Now()}) + } + plan.Route = HitlRouteDestroy + case hitl.ActionSnapshot: + var opts SnapshotOptions + if len(in.Lifecycle.Params) > 0 { + if err := json.Unmarshal(in.Lifecycle.Params, &opts); err != nil { + return HitlRunPlan{}, fm.storeRunResult(ctx, id, hitl.Result{Error: err.Error(), FinishedAt: time.Now()}) + } + } + plan.Route, plan.Snapshot = HitlRouteSnapshot, &opts + case hitl.ActionRotateToken: + plan.Route = HitlRouteRotateToken + default: + plan.Route = HitlRouteInProcess + } + return plan, nil +} + +// HitlRunInProcess runs an approved exec, computer, drain, fork or migrate +// and writes the result to the intervention. the result never leaves this +// step, so exec output and screenshots stay out of the journal. +// +// the step marks the row before it starts. a retried exec within +// execReplayWindow of the mark carries the step's idempotency key to the +// host agent, which hands back the first run's output instead of running it +// again. the others have no such key, so they run at most once: an attempt +// that finds the mark settles the action as possibly run rather than doing +// it a second time, and so does an exec retried after the window. +func (fm *FleetManager) HitlRunInProcess(ctx context.Context, id string) error { + in, err := fm.loadIntervention(ctx, id) + if err != nil { + return err + } + if !needsRun(in) { + return nil + } + if res, ok := fm.takeUnstored(id); ok { + return fm.storeDurableResult(ctx, id, res) + } + started, err := fm.markRunStarted(ctx, id) + if err != nil { + return err + } + // an exec retried well inside the agent's key ttl is replayed by the + // agent. past that the agent may have forgotten the key and would run + // the command again, so it is settled like the others. + if !started.IsZero() && (in.Exec == nil || time.Since(started) > execReplayWindow) { + return fm.storeDurableResult(ctx, id, hitl.Result{Error: errRunInterrupted, FinishedAt: time.Now()}) + } + actx, cancel := context.WithTimeout(ctx, runTimeout(in)) + res := fm.runApproved(actx, in) + cancel() + res.FinishedAt = time.Now() + return fm.storeDurableResult(ctx, id, res) +} + +// HitlRunFinish writes the result of an action the run handed to its +// environment's handler, and records it as the lifecycle operation it was, +// as DestroyVM, CreateSnapshot and RotateToken do. a result already there is +// kept and not recorded again. +func (fm *FleetManager) HitlRunFinish(ctx context.Context, id string, plan HitlRunPlan, res hitl.Result) error { + in, err := fm.loadIntervention(ctx, id) + if err != nil { + return err + } + if !needsRun(in) { + return nil + } + res.FinishedAt = time.Now() + if err := fm.storeRunResult(ctx, id, res); err != nil { + return err + } + if om, ok := fm.metrics.(OperationMetrics); ok && !plan.StartedAt.IsZero() { + var opErr error + if res.Error != "" { + opErr = errors.New(res.Error) + } + om.OperationCompleted(string(plan.Route), res.FinishedAt.Sub(plan.StartedAt), opErr) + } + return nil +} + +// HitlExpire applies an intervention's on_timeout once its deadline has +// passed, through the same path a reader takes. it returns how long is +// left when the timer fired early by this process's clock; zero otherwise. +// an approval the expiry makes is handed to restate like a human's. +func (fm *FleetManager) HitlExpire(ctx context.Context, id string) (time.Duration, error) { + in, err := fm.GetIntervention(ctx, id) + if errors.Is(err, ErrInterventionNotFound) { + return 0, nil + } + if err != nil { + return 0, err + } + if in.State == hitl.StatePending { + return max(time.Until(in.Deadline), time.Millisecond), nil + } + return 0, nil +} + +// loadIntervention reads one intervention as stored, with no expiry applied. +func (fm *FleetManager) loadIntervention(ctx context.Context, id string) (hitl.Intervention, error) { + rec, err := fm.store.GetIntervention(ctx, id) + if err != nil { + return hitl.Intervention{}, err + } + return fm.interventionFromRecord(rec) +} + +// storeDurableResult stores res, and keeps it in memory when the store +// fails, so the step's retry in this process stores what ran instead of +// finding its own start mark. +func (fm *FleetManager) storeDurableResult(ctx context.Context, id string, res hitl.Result) error { + if err := fm.storeRunResult(ctx, id, res); err != nil { + fm.hitlMu.Lock() + if fm.hitlUnstored == nil { + fm.hitlUnstored = make(map[string]hitl.Result) + } + fm.hitlUnstored[id] = res + fm.hitlMu.Unlock() + return err + } + return nil +} + +func (fm *FleetManager) takeUnstored(id string) (hitl.Result, bool) { + fm.hitlMu.Lock() + defer fm.hitlMu.Unlock() + res, ok := fm.hitlUnstored[id] + delete(fm.hitlUnstored, id) + return res, ok +} + +// execReplayWindow is how long after its first attempt an approved exec is +// retried on the host agent's idempotency key. well inside the agents' 24h +// default key ttl, which an operator can shorten. +const execReplayWindow = time.Hour + +// markRunStarted records on the row that an attempt at the approved action +// is about to begin, and returns when an earlier one began, zero when none +// had. the mark lives in the sealed result column next to the answer, so it +// is not part of the intervention anyone reads; the result replaces it. +func (fm *FleetManager) markRunStarted(ctx context.Context, id string) (time.Time, error) { + fm.hitlMu.Lock() + defer fm.hitlMu.Unlock() + rec, err := fm.store.GetIntervention(ctx, id) + if err != nil { + return time.Time{}, err + } + raw, err := fm.unseal(rec.Result, rec.Sealed) + if err != nil { + return time.Time{}, fmt.Errorf("intervention %s: result: %w", id, err) + } + var r interventionResult + if len(raw) > 0 { + if err := json.Unmarshal(raw, &r); err != nil { + return time.Time{}, fmt.Errorf("intervention %s: result: %w", id, err) + } + } + if r.Started != nil { + return *r.Started, nil + } + now := time.Now().UTC() + r.Started = &now + if raw, err = json.Marshal(r); err != nil { + return time.Time{}, err + } + if rec.Result, _, err = fm.seal(raw); err != nil { + return time.Time{}, fmt.Errorf("seal intervention result: %w", err) + } + return time.Time{}, fm.store.UpsertIntervention(ctx, rec) +} + +// runTimeout bounds one approved action. +func runTimeout(in hitl.Intervention) time.Duration { + switch { + case in.Exec != nil: + return execTimeout(in.Exec.TimeoutMS) + time.Minute + case in.Computer != nil: + return guestComputerTimeout + time.Minute + } + return lifecycleRunTimeout +} + +// HitlRunTimeout is the longest an approved action runs, the durable run's +// abort timeout. +const HitlRunTimeout = lifecycleRunTimeout + +// runAction names an approved action for metrics. +func runAction(in hitl.Intervention) string { + switch { + case in.Exec != nil: + return "exec" + case in.Computer != nil: + return "computer" + case in.Lifecycle != nil: + return string(in.Lifecycle.Action) + } + return "unknown" +} + +func (fm *FleetManager) hitlMetrics() HitlMetrics { + m, _ := fm.metrics.(HitlMetrics) + return m +} diff --git a/internal/orchestrator/hitl_durable_test.go b/internal/orchestrator/hitl_durable_test.go new file mode 100644 index 0000000..6549906 --- /dev/null +++ b/internal/orchestrator/hitl_durable_test.go @@ -0,0 +1,487 @@ +package orchestrator + +import ( + "context" + "encoding/json" + "errors" + "slices" + "sync" + "testing" + "time" + + "github.com/folsomintel/fuse/internal/hitl" +) + +// fakeDurableHitl records what the fleet hands restate. +type fakeDurableHitl struct { + mu sync.Mutex + runs []string + expires map[string]time.Duration + err error +} + +func (d *fakeDurableHitl) SendHitlRun(_ context.Context, id string) error { + d.mu.Lock() + defer d.mu.Unlock() + if d.err != nil { + return d.err + } + d.runs = append(d.runs, id) + return nil +} + +func (d *fakeDurableHitl) SendHitlExpire(_ context.Context, id string, delay time.Duration) error { + d.mu.Lock() + defer d.mu.Unlock() + if d.expires == nil { + d.expires = map[string]time.Duration{} + } + d.expires[id] = delay + return nil +} + +func (d *fakeDurableHitl) sent() ([]string, map[string]time.Duration) { + d.mu.Lock() + defer d.mu.Unlock() + out := map[string]time.Duration{} + for k, v := range d.expires { + out[k] = v + } + return append([]string(nil), d.runs...), out +} + +// fakeHitlMetrics records the hitl metrics. +type fakeHitlMetrics struct { + mu sync.Mutex + decided []string + deciders []string + runs []string +} + +func (m *fakeHitlMetrics) ReconcileCompleted(ReconcileSummary) {} +func (m *fakeHitlMetrics) HitlDecided(kind, outcome string, _ time.Duration) { + m.mu.Lock() + m.decided = append(m.decided, kind+":"+outcome) + m.mu.Unlock() +} +func (m *fakeHitlMetrics) HitlDeciderAsked(verdict string, _ time.Duration) { + m.mu.Lock() + m.deciders = append(m.deciders, verdict) + m.mu.Unlock() +} +func (m *fakeHitlMetrics) HitlRunFinished(action string, _ time.Duration, failed bool) { + m.mu.Lock() + r := action + ":ok" + if failed { + r = action + ":error" + } + m.runs = append(m.runs, r) + m.mu.Unlock() +} + +// fakeOpMetrics records lifecycle operations. +type fakeOpMetrics struct { + mu sync.Mutex + ops []string +} + +func (m *fakeOpMetrics) ReconcileCompleted(ReconcileSummary) {} +func (m *fakeOpMetrics) OperationCompleted(op string, _ time.Duration, err error) { + m.mu.Lock() + defer m.mu.Unlock() + r := op + ":ok" + if err != nil { + r = op + ":error" + } + m.ops = append(m.ops, r) +} + +func (m *fakeOpMetrics) seen() []string { + m.mu.Lock() + defer m.mu.Unlock() + return append([]string(nil), m.ops...) +} + +// setRunMark puts a start mark from at on the intervention, as an attempt +// that began then and died would have left it. +func setRunMark(t *testing.T, fm *FleetManager, id string, at time.Time) { + t.Helper() + ctx := context.Background() + if _, err := fm.markRunStarted(ctx, id); err != nil { + t.Fatal(err) + } + rec, err := fm.store.GetIntervention(ctx, id) + if err != nil { + t.Fatal(err) + } + raw, err := json.Marshal(interventionResult{Started: &at}) + if err != nil { + t.Fatal(err) + } + if rec.Result, _, err = fm.seal(raw); err != nil { + t.Fatal(err) + } + if err := fm.store.UpsertIntervention(ctx, rec); err != nil { + t.Fatal(err) + } +} + +// an exec retried soon after its first attempt goes back to the agent, which +// replays it on the step's key; one retried after the window could run again +// on an agent that forgot the key, so it is settled as possibly run. +func TestDurableHitlExecReplayWindow(t *testing.T) { + fm, vmID, _, _ := durableHitlFleet(t) + env := mockEnvOf(fm, vmID) + ctx := context.Background() + + recent, _ := gateExec(fm, vmID, "git", "push", "--force") + stale, _ := gateExec(fm, vmID, "git", "push", "--force", "origin") + for _, id := range []string{recent.ID, stale.ID} { + if _, err := fm.DecideIntervention(ctx, id, DecideRequest{Outcome: hitl.OutcomeApprove, By: approver}); err != nil { + t.Fatal(err) + } + } + setRunMark(t, fm, recent.ID, time.Now().Add(-time.Minute)) + setRunMark(t, fm, stale.ID, time.Now().Add(-execReplayWindow-time.Minute)) + + for _, id := range []string{recent.ID, stale.ID} { + if err := fm.HitlRunInProcess(ctx, id); err != nil { + t.Fatal(err) + } + } + if got, _ := fm.GetIntervention(ctx, recent.ID); got.Result == nil || got.Result.Exec == nil { + t.Fatalf("recent retry result %+v, want the exec's", got.Result) + } + if got, _ := fm.GetIntervention(ctx, stale.ID); got.Result == nil || got.Result.Error != errRunInterrupted { + t.Fatalf("stale retry result %+v, want the interrupted error", got.Result) + } + if n := len(env.execArgv); n != 1 { + t.Fatalf("exec reached the agent %d times, want only the recent retry", n) + } +} + +// a send waiting on restate is not queued a second time for the same +// approval. +func TestDurableHitlRunSendQueuedOnce(t *testing.T) { + fm, _, _, _ := durableHitlFleet(t) + block := make(chan struct{}) + d := &blockingDurableHitl{block: block} + fm.SetDurableHitl(d) + for range 5 { + fm.sendHitlRun(d, "int_1") + } + close(block) + fm.hitlSends.Wait() + if n := d.count(); n != 1 { + t.Fatalf("%d sends for one approval, want 1", n) + } + fm.sendHitlRun(d, "int_1") + fm.hitlSends.Wait() + if n := d.count(); n != 2 { + t.Fatalf("%d sends after the first finished, want a new one", n) + } +} + +type blockingDurableHitl struct { + fakeDurableHitl + block chan struct{} + n int +} + +func (d *blockingDurableHitl) SendHitlRun(context.Context, string) error { + <-d.block + d.mu.Lock() + d.n++ + d.mu.Unlock() + return nil +} + +func (d *blockingDurableHitl) count() int { + d.mu.Lock() + defer d.mu.Unlock() + return d.n +} + +func durableHitlFleet(t *testing.T) (*FleetManager, string, *fakeDurableHitl, *fakeHitlMetrics) { + t.Helper() + m := &fakeHitlMetrics{} + fm, vmID := hitlFleet(t, FleetConfig{Metrics: m}, gateSpec()) + d := &fakeDurableHitl{} + fm.SetDurableHitl(d) + return fm, vmID, d, m +} + +// approved exec in mode self: the approval is handed to restate and nothing +// runs in process until the durable run's step does; the step runs it once, +// writes the result and wakes a waiter. +func TestDurableHitlApprovalGoesToRestate(t *testing.T) { + fm, vmID, d, m := durableHitlFleet(t) + env := mockEnvOf(fm, vmID) + env.execResult = ExecResult{Stdout: []byte("pushed")} + ctx := context.Background() + + held, err := gateExec(fm, vmID, "git", "push", "--force") + if err != nil || held == nil { + t.Fatalf("held=%v err=%v", held, err) + } + fm.hitlSends.Wait() + _, expires := d.sent() + if delay, ok := expires[held.ID]; !ok || delay <= 0 || delay > 600*time.Second { + t.Fatalf("expire timer %v (armed %v), want the deadline", delay, ok) + } + + waited := make(chan hitl.Intervention, 1) + go func() { + in, _ := fm.WaitIntervention(ctx, held.ID, 5*time.Second) + waited <- in + }() + if _, err := fm.DecideIntervention(ctx, held.ID, DecideRequest{Outcome: hitl.OutcomeApprove, By: approver}); err != nil { + t.Fatal(err) + } + fm.hitlSends.Wait() + if runs, _ := d.sent(); !slices.Equal(runs, []string{held.ID}) { + t.Fatalf("runs sent %v, want %s", runs, held.ID) + } + time.Sleep(50 * time.Millisecond) + if n := len(env.execArgv); n != 0 { + t.Fatalf("the approval ran in process %d times", n) + } + + plan, err := fm.HitlRunBegin(ctx, held.ID) + if err != nil || plan.Route != HitlRouteInProcess || plan.EnvID != vmID { + t.Fatalf("plan %+v err %v, want in process on %s", plan, err, vmID) + } + if err := fm.HitlRunInProcess(ctx, held.ID); err != nil { + t.Fatal(err) + } + // a replayed step finds the result and does not run again. + if err := fm.HitlRunInProcess(ctx, held.ID); err != nil { + t.Fatal(err) + } + if plan, _ := fm.HitlRunBegin(ctx, held.ID); plan.Route != HitlRouteNone { + t.Fatalf("plan after the run %+v, want nothing left", plan) + } + if n := len(env.execArgv); n != 1 { + t.Fatalf("exec ran %d times, want once", n) + } + select { + case in := <-waited: + if in.Result == nil || in.Result.Exec == nil || in.Result.Exec.Stdout != "pushed" { + t.Fatalf("waiter got %+v", in.Result) + } + case <-time.After(5 * time.Second): + t.Fatal("waiter never woke") + } + m.mu.Lock() + defer m.mu.Unlock() + if !slices.Equal(m.decided, []string{"exec:approve"}) || !slices.Equal(m.runs, []string{"exec:ok"}) { + t.Fatalf("metrics decided %v runs %v", m.decided, m.runs) + } +} + +// an action with no idempotency key runs at most once: an attempt that finds +// an earlier one's start mark settles it as possibly run. +func TestDurableHitlRunInProcessAtMostOnce(t *testing.T) { + fm, vmID, _, _ := durableHitlFleet(t) + ctx := context.Background() + held, err := fm.GateLifecycle(ctx, vmID, hitl.ActionDrain, nil, GateCaller{By: testActor}) + if err != nil || held == nil { + t.Fatalf("drain should be held: %v", err) + } + if _, err := fm.DecideIntervention(ctx, held.ID, DecideRequest{Outcome: hitl.OutcomeApprove, By: approver}); err != nil { + t.Fatal(err) + } + // an earlier attempt began and died. + if started, err := fm.markRunStarted(ctx, held.ID); err != nil || !started.IsZero() { + t.Fatalf("first mark: started=%v err=%v", started, err) + } + if in, _ := fm.GetIntervention(ctx, held.ID); in.Settled() { + t.Fatalf("the start mark must not settle the intervention: %+v", in.Result) + } + if err := fm.HitlRunInProcess(ctx, held.ID); err != nil { + t.Fatal(err) + } + got, _ := fm.GetIntervention(ctx, held.ID) + if got.Result == nil || got.Result.Error != errRunInterrupted { + t.Fatalf("result %+v, want the interrupted error", got.Result) + } + if info, _ := fm.GetVM(vmID); info.State != VMStateRunning { + t.Fatalf("vm state %s: the drain ran a second time", info.State) + } +} + +func TestDurableHitlRunInProcessRunsDrain(t *testing.T) { + fm, vmID, _, _ := durableHitlFleet(t) + ctx := context.Background() + held, _ := fm.GateLifecycle(ctx, vmID, hitl.ActionDrain, nil, GateCaller{By: testActor}) + if _, err := fm.DecideIntervention(ctx, held.ID, DecideRequest{Outcome: hitl.OutcomeApprove, By: approver}); err != nil { + t.Fatal(err) + } + if plan, _ := fm.HitlRunBegin(ctx, held.ID); plan.Route != HitlRouteInProcess { + t.Fatalf("drain plan %+v, want in process", plan) + } + if err := fm.HitlRunInProcess(ctx, held.ID); err != nil { + t.Fatal(err) + } + got, _ := fm.GetIntervention(ctx, held.ID) + if got.Result == nil || got.Result.Error != "" { + t.Fatalf("drain result %+v", got.Result) + } + if info, _ := fm.GetVM(vmID); info.State != VMStateDraining { + t.Fatalf("vm state %s, want draining", info.State) + } +} + +// destroy, snapshot and token rotation go to their environment handler. +func TestDurableHitlRunBeginRoutes(t *testing.T) { + spec := &hitl.Spec{TimeoutSeconds: 600, Gates: []hitl.Gate{ + {Name: "life", On: hitl.KindLifecycle, Match: hitl.GateMatch{Action: []string{"destroy", "snapshot", "rotate-token"}}}, + }} + ops := &fakeOpMetrics{} + fm, vmID := hitlFleet(t, FleetConfig{Metrics: ops}, spec) + fm.SetDurableHitl(&fakeDurableHitl{}) + ctx := context.Background() + approve := func(action hitl.LifecycleAction, params any) string { + t.Helper() + held, err := fm.GateLifecycle(ctx, vmID, action, params, GateCaller{By: testActor}) + if err != nil || held == nil { + t.Fatalf("%s should be held: %v", action, err) + } + if _, err := fm.DecideIntervention(ctx, held.ID, DecideRequest{Outcome: hitl.OutcomeApprove, By: approver}); err != nil { + t.Fatal(err) + } + return held.ID + } + + destroy := approve(hitl.ActionDestroy, nil) + snap := approve(hitl.ActionSnapshot, SnapshotOptions{Comment: "before deploy"}) + rotate := approve(hitl.ActionRotateToken, nil) + for id, want := range map[string]HitlRunRoute{destroy: HitlRouteDestroy, snap: HitlRouteSnapshot, rotate: HitlRouteRotateToken} { + plan, err := fm.HitlRunBegin(ctx, id) + if err != nil || plan.Route != want || plan.EnvID != vmID { + t.Fatalf("plan %+v err %v, want %s on %s", plan, err, want, vmID) + } + if want == HitlRouteSnapshot && (plan.Snapshot == nil || plan.Snapshot.Comment != "before deploy") { + t.Fatalf("snapshot plan %+v, want the approved options", plan.Snapshot) + } + } + + plan, _ := fm.HitlRunBegin(ctx, snap) + if err := fm.HitlRunFinish(ctx, snap, plan, hitl.Result{SnapshotID: "snap-9"}); err != nil { + t.Fatal(err) + } + // a replayed finish keeps the first result and is not counted again. + if err := fm.HitlRunFinish(ctx, snap, plan, hitl.Result{Error: "late"}); err != nil { + t.Fatal(err) + } + if got := ops.seen(); !slices.Equal(got, []string{"snapshot:ok"}) { + t.Fatalf("operations %v, want one snapshot", got) + } + if got, _ := fm.GetIntervention(ctx, snap); got.Result == nil || got.Result.SnapshotID != "snap-9" { + t.Fatalf("snapshot result %+v", got.Result) + } + + // a destroy of an environment that is gone settles with not found. + fm.mu.Lock() + delete(fm.vms, vmID) + fm.mu.Unlock() + if plan, err := fm.HitlRunBegin(ctx, destroy); err != nil || plan.Route != HitlRouteNone { + t.Fatalf("plan %+v err %v, want nothing to run", plan, err) + } + if got, _ := fm.loadIntervention(ctx, destroy); got.Result == nil || got.Result.Error == "" { + t.Fatalf("gone destroy result %+v, want an error", got.Result) + } +} + +// an approval whose send never reached restate is sent again by the +// restart's first reconcile, instead of being settled as maybe-run. +func TestDurableHitlResumesApprovedRuns(t *testing.T) { + fm, vmID, d, _ := durableHitlFleet(t) + now := time.Now() + in := hitl.Intervention{ + ID: hitl.NewID(now), EnvID: vmID, Kind: hitl.KindExec, Gate: "force-push", Mode: hitl.ModeAsk, + State: hitl.StateApproved, Exec: &hitl.ExecPayload{Argv: []string{"git"}}, TriggeredBy: testActor, + Deadline: now.Add(time.Minute), Decision: &hitl.Decision{Outcome: hitl.OutcomeApprove, DecidedBy: approver, DecidedAt: now}, + CreatedAt: now, UpdatedAt: now, + } + rec, err := fm.interventionToRecord(in) + if err != nil { + t.Fatal(err) + } + if err := fm.store.UpsertIntervention(context.Background(), rec); err != nil { + t.Fatal(err) + } + fm.reconcileInterventions(context.Background()) + fm.hitlSends.Wait() + if runs, _ := d.sent(); !slices.Equal(runs, []string{in.ID}) { + t.Fatalf("runs sent %v, want %s", runs, in.ID) + } + if got, _ := fm.GetIntervention(context.Background(), in.ID); got.Result != nil { + t.Fatalf("result %+v, want the run left to restate", got.Result) + } +} + +// a send restate refused is sent again on the next reconcile. +func TestDurableHitlResendsAfterFailedSend(t *testing.T) { + fm, vmID, d, _ := durableHitlFleet(t) + ctx := context.Background() + fm.reconcileInterventions(ctx) // the startup pass, with nothing owed. + held, _ := gateExec(fm, vmID, "git", "push", "--force") + d.mu.Lock() + d.err = errors.New("restate down") + d.mu.Unlock() + if _, err := fm.DecideIntervention(ctx, held.ID, DecideRequest{Outcome: hitl.OutcomeApprove, By: approver}); err != nil { + t.Fatal(err) + } + fm.hitlSends.Wait() + if !fm.hitlResend.Load() { + t.Fatal("a failed send did not ask for a resend") + } + d.mu.Lock() + d.err = nil + d.mu.Unlock() + fm.reconcileInterventions(ctx) + fm.hitlSends.Wait() + if runs, _ := d.sent(); !slices.Equal(runs, []string{held.ID}) { + t.Fatalf("runs sent %v, want %s", runs, held.ID) + } + if fm.hitlResend.Load() { + t.Fatal("the resend flag stayed set") + } +} + +// the deadline timer applies on_timeout; an approve on_timeout is run like +// a human's approval. a timer that fires early reports what is left. +func TestDurableHitlExpireAppliesOnTimeout(t *testing.T) { + spec := gateSpec() + spec.OnTimeout = hitl.OnTimeoutAllow + fm, vmID := hitlFleet(t, FleetConfig{}, spec) + d := &fakeDurableHitl{} + fm.SetDurableHitl(d) + ctx := context.Background() + held, _ := gateExec(fm, vmID, "git", "push", "--force") + + left, err := fm.HitlExpire(ctx, held.ID) + if err != nil || left <= 0 { + t.Fatalf("early timer: left=%v err=%v, want the rest of the deadline", left, err) + } + rec, _ := fm.store.GetIntervention(ctx, held.ID) + rec.Deadline = time.Now().Add(-time.Second) + if err := fm.store.UpsertIntervention(ctx, rec); err != nil { + t.Fatal(err) + } + if left, err := fm.HitlExpire(ctx, held.ID); err != nil || left != 0 { + t.Fatalf("due timer: left=%v err=%v", left, err) + } + got, _ := fm.loadIntervention(ctx, held.ID) + if got.State != hitl.StateExpired || got.Decision == nil || got.Decision.Outcome != hitl.OutcomeApprove { + t.Fatalf("expired %+v, want on_timeout approve applied", got) + } + fm.hitlSends.Wait() + if runs, _ := d.sent(); !slices.Equal(runs, []string{held.ID}) { + t.Fatalf("runs sent %v, want the timed-out approval", runs) + } + if left, err := fm.HitlExpire(ctx, "hitl_missing"); err != nil || left != 0 { + t.Fatalf("missing intervention: left=%v err=%v", left, err) + } +} diff --git a/internal/orchestrator/interventions.go b/internal/orchestrator/interventions.go index 1ae0fa0..503fed2 100644 --- a/internal/orchestrator/interventions.go +++ b/internal/orchestrator/interventions.go @@ -195,6 +195,7 @@ func (fm *FleetManager) raiseIntervention(ctx context.Context, in hitl.Intervent }) fm.publishInterventionEvent(in) fm.auditIntervention(ctx, in, auditMeta{remoteAddr: remoteAddr}) + fm.scheduleExpiry(in) return nil } @@ -422,6 +423,9 @@ func (fm *FleetManager) expireIntervention(ctx context.Context, id string) (hitl // returns the stored intervention and ErrInterventionResolved when it was // already terminal. func (fm *FleetManager) resolveIntervention(ctx context.Context, id string, meta auditMeta, apply func(hitl.Intervention, time.Time) (hitl.Intervention, error)) (hitl.Intervention, error) { + // in modes self and cloud restate runs an approval; otherwise this + // process does. + durable := fm.durableHitl() fm.hitlMu.Lock() rec, err := fm.store.GetIntervention(ctx, id) if err != nil { @@ -454,11 +458,17 @@ func (fm *FleetManager) resolveIntervention(ctx context.Context, id string, meta close(ch) delete(fm.hitlWaiters, id) } - run := fm.claimRun(next) + run := durable == nil && fm.claimRun(next) fm.dropLockLocked(next.EnvID, id) fm.hitlMu.Unlock() - if run { + switch { + case run: fm.launchRun(next) + case durable != nil && needsRun(next): + fm.sendHitlRun(durable, id) + } + if m := fm.hitlMetrics(); m != nil && current.State == hitl.StatePending && next.Gate != "" && next.Decision != nil { + m.HitlDecided(string(next.Kind), string(next.Decision.Outcome), next.Decision.DecidedAt.Sub(next.CreatedAt)) } fm.appendEvent(ctx, "intervention", id, "intervention."+string(next.State), map[string]any{ @@ -518,9 +528,13 @@ func (fm *FleetManager) publishInterventionEvent(in hitl.Intervention) { // interventionResult is the sealed result column: an ask's answer, or what // running an approved gated action produced. +// Started is when a durable run of an action with no idempotency key began, +// so a retry knows not to run it again (see markRunStarted); the result +// replaces it. type interventionResult struct { - Answer string `json:"answer,omitempty"` - Run *hitl.Result `json:"run,omitempty"` + Answer string `json:"answer,omitempty"` + Run *hitl.Result `json:"run,omitempty"` + Started *time.Time `json:"started,omitempty"` } // interventionPayload is the sealed payload column. the ask fields keep the diff --git a/internal/orchestrator/interventions_reconcile.go b/internal/orchestrator/interventions_reconcile.go index b4c0116..2f63d87 100644 --- a/internal/orchestrator/interventions_reconcile.go +++ b/internal/orchestrator/interventions_reconcile.go @@ -19,7 +19,17 @@ func (fm *FleetManager) reconcileInterventions(ctx context.Context) { if fm.store == nil { return } - fm.hitlOrphans.Do(func() { fm.recoverOrphanRuns(ctx) }) + if d := fm.durableHitl(); d != nil { + // restate resumes a run that was in flight; this hands it the ones + // that never reached it. + ran := false + fm.hitlOrphans.Do(func() { ran = true; fm.resumeApprovedRuns(ctx, d) }) + if !ran && fm.hitlResend.Load() { + fm.resumeApprovedRuns(ctx, d) + } + } else { + fm.hitlOrphans.Do(func() { fm.recoverOrphanRuns(ctx) }) + } recs, err := fm.store.ListInterventions(ctx, InterventionFilter{State: string(hitl.StatePending)}) if err != nil { fm.logger.Warn("list pending interventions failed", "err", err) diff --git a/orchestrator/main.go b/orchestrator/main.go index 4cd74c1..205ab56 100644 --- a/orchestrator/main.go +++ b/orchestrator/main.go @@ -574,6 +574,7 @@ func run() error { fm.SetDurableSnapshots(restateClient) fm.SetDurableTokenRotator(restateClient) fm.SetDurableWebhooks(restateClient) + fm.SetDurableHitl(restateClient) } fm.Start(rootCtx) defer fm.Stop()