feat: durable restate destroy with retries and a destroy_failed state

cli/render.go

@@ -222,7 +222,7 @@ func stateStyle(state string) string {
 	switch state {
 	case fuse.StateRunning:
 		return styleGood.Render(state)
-	case fuse.StateFailed:
+	case fuse.StateFailed, fuse.StateDestroyFailed:
 		return styleBad.Render(state)
 	case fuse.StateProvisioning, fuse.StateDraining, fuse.StateDestroying:
 		return styleWarn.Render(state)

docs/content/docs/concepts/environments.mdx

@@ -16,6 +16,7 @@ lifecycle. A background reconcile loop catches orphans and stuck VMs.
 ```
 provisioning -> running -> draining -> destroying -> destroyed
                                                     -> failed
+                                       destroying -> destroy_failed -> destroying (on DELETE)
 ```
 
 - **provisioning**: a host has been picked and the VM is booting.
@@ -25,6 +26,10 @@ provisioning -> running -> draining -> destroying -> destroyed
   **not yet destroyed**. Only running environments can be drained; any other state
   returns a conflict.
 - **destroying / destroyed**: the VM is being (or has been) torn down.
+- **destroy_failed**: with restate on (`FUSE_RESTATE_MODE=self` or `cloud`), a
+  destroy whose host kept failing after every retry. The VM may still exist on
+  its host and still counts against the host's capacity. The last error is on
+  the environment and in a `destroy` dead letter. `DELETE` it again to retry.
 - **failed**: provisioning or an operation failed.
 
 `destroyed` and `failed` are terminal states you only ever observe on the event

docs/content/docs/concepts/reconcile.mdx

@@ -157,6 +157,15 @@ such a VM once it has been `destroying` for more than two minutes, so it goes aw
 once the host recovers. This applies to every background teardown, not only
 draining ones.
 
+With restate on, every destroy runs as the `Environment/{vm_id}/Destroy`
+handler instead, and retries the host call itself under the `Environment`
+retry policy (`FUSE_RESTATE_RETRY_ENVIRONMENT_*`). It survives an orchestrator
+restart: the handler resumes at the first unfinished step. Reconcile then only
+re-sends `Destroy` for a VM that has sat in `destroying` for two minutes with
+no attempt in flight, such as one left over from before the upgrade. A destroy
+that runs out of retries moves the VM to `destroy_failed` and writes a
+`destroy` dead letter, counted by `orchestrator_destroy_dead_letters_total`.
+
 ## Health verdicts
 
 This is the one pass that is **not** a detector. It reads and reports; it never

docs/content/reference/cli/environment/list.mdx

@@ -19,7 +19,7 @@ whole fleet.
 | Flag | Purpose |
 | --- | --- |
 | `--task-id` | Filter by task ID. |
-| `--state` | Filter by state: `provisioning`\|`running`\|`draining`\|`destroying`. |
+| `--state` | Filter by state: `provisioning`\|`running`\|`draining`\|`destroying`\|`destroy_failed`. |
 | `--host-id` | Filter by host ID. Defaults to the active host if `--all` is not set. |
 | `--all` | List across all hosts, ignoring the active host. |
 

docs/content/reference/go/environments.mdx

@@ -174,9 +174,10 @@ const (
 	StateProvisioning = "provisioning"
 	StateRunning      = "running"
 	StateDraining     = "draining"
-	StateDestroying   = "destroying"
-	StateDestroyed    = "destroyed"
-	StateFailed       = "failed"
+	StateDestroying    = "destroying"
+	StateDestroyFailed = "destroy_failed"
+	StateDestroyed     = "destroyed"
+	StateFailed        = "failed"
 )
 
 func IsTerminalState(state string) bool

docs/content/reference/python/environments.mdx

@@ -118,6 +118,7 @@ STATE_PROVISIONING = "provisioning"
 STATE_RUNNING = "running"
 STATE_DRAINING = "draining"
 STATE_DESTROYING = "destroying"
+STATE_DESTROY_FAILED = "destroy_failed"
 STATE_DESTROYED = "destroyed"
 STATE_FAILED = "failed"
 

docs/content/reference/typescript/environments.mdx

@@ -122,6 +122,7 @@ export const State = {
   Running: "running",
   Draining: "draining",
   Destroying: "destroying",
+  DestroyFailed: "destroy_failed",
   Destroyed: "destroyed",
   Failed: "failed",
 } as const;

docs/runbooks/restate.mdx

@@ -126,11 +126,20 @@ docker exec fuse-restate restate invocations cancel <id>   # or kill
 
 or the ui at `http://127.0.0.1:9070/ui` (through the ssh tunnel).
 
+destroys run as `Environment/<vm_id>/Destroy`. one stuck retrying a host shows
+its last error in the ui; one that ran out of retries left the environment in
+`destroy_failed` with a `destroy` dead letter
+(`orchestrator_destroy_dead_letters_total`). fix the host, then `DELETE` the
+environment again.
+
 ## restate-server is down
 
 - the orchestrator keeps serving its api; `/ready` reports restate degraded.
   operations that run on restate wait and resume once it is back; nothing
   in flight is lost, it is in `/var/lib/restate`.
+- a `DELETE /v1/environments/{id}` fails with a 500 until restate is back.
+  background teardowns (task completion, reconcile) stay `destroying` and
+  are re-sent by reconcile once it answers.
 - `systemctl status restate-server`, `journalctl -u restate-server -n 200`.
 - `docker` down takes restate with it (`Requires=docker.service`):
   `systemctl status docker`.

internal/api/docs/swagger.json

@@ -201,7 +201,8 @@
                         "enum": [
                             "provisioning",
                             "running",
-                            "destroying"
+                            "destroying",
+                            "destroy_failed"
                         ],
                         "type": "string",
                         "description": "Filter by state",

internal/api/docs/swagger.yaml

@@ -452,6 +452,7 @@ paths:
         - provisioning
         - running
         - destroying
+        - destroy_failed
         in: query
         name: state
         type: string

internal/api/handlers.go

@@ -692,7 +692,7 @@ func validateTimeoutSpec(s ResourceSpec) error {
 //	@Tags		environments
 //	@Produce	json
 //	@Param		task_id	query		string	false	"Filter by task ID"
-//	@Param		state	query		string	false	"Filter by state"	Enums(provisioning, running, destroying)
+//	@Param		state	query		string	false	"Filter by state"	Enums(provisioning, running, destroying, destroy_failed)
 //	@Param		host_id	query		string	false	"Filter by host ID"
 //	@Param		limit	query		int		false	"Max results per page (default 50, max 200)"
 //	@Param		cursor	query		string	false	"Opaque pagination cursor from a previous response's next_cursor"

internal/durable/destroy.go

@@ -0,0 +1,107 @@
+package durable
+
+import (
+	"context"
+
+	restate "github.com/restatedev/sdk-go"
+	"github.com/restatedev/sdk-go/ingress"
+
+	"github.com/folsomintel/fuse/internal/orchestrator"
+)
+
+// opDestroy is the op label on destroy's step spans and metrics.
+const opDestroy = "destroy"
+
+// Destroyer is the part of the fleet the destroy handler drives, one method
+// per step. *FleetManager satisfies it.
+type Destroyer interface {
+	DestroyBegin(ctx context.Context, vmID, reason string) (orchestrator.DestroyTarget, error)
+	DestroyReleaseEgress(ctx context.Context, vmID string) error
+	DestroyOnHost(ctx context.Context, vmID, hostID string) error
+	DestroyUnpublishIngress(ctx context.Context, vmID string) error
+	DestroyFinish(ctx context.Context, vmID, reason string) error
+	DestroyFailed(ctx context.Context, vmID, reason, lastErr string) error
+}
+
+// destroy is Environment/{vm_id}.Destroy. every step is journaled, so after a
+// crash the handler resumes at the first unfinished one. the host destroy
+// retries under policy; when it runs out the vm is left destroy_failed with a
+// dead letter, and the invocation fails with the last error.
+func destroy(ctx restate.ObjectContext, fleet Destroyer, policy RetryPolicy, req orchestrator.DestroyRequest) error {
+	vmID := restate.Key(ctx)
+
+	target, err := Run(ctx, opDestroy, "begin", func(c context.Context) (orchestrator.DestroyTarget, error) {
+		return fleet.DestroyBegin(c, vmID, req.Reason)
+	})
+	if err != nil {
+		return err
+	}
+	if target.Replaced {
+		// a create cleanup whose vm id a retried create now holds.
+		return nil
+	}
+	hostID := target.HostID
+	if !target.Tracked {
+		if req.HostID == "" {
+			// neither tracked nor placed: an earlier destroy finished it.
+			return nil
+		}
+		hostID = req.HostID
+	}
+
+	if err := step(ctx, "release_egress", func(c context.Context) error {
+		return fleet.DestroyReleaseEgress(c, vmID)
+	}); err != nil {
+		return err
+	}
+
+	hostErr := step(ctx, "host_destroy", func(c context.Context) error {
+		return fleet.DestroyOnHost(c, vmID, hostID)
+	}, restate.WithInitialRetryInterval(policy.InitialInterval),
+		restate.WithMaxRetryInterval(policy.MaxInterval),
+		restate.WithMaxRetryAttempts(uint(policy.MaxAttempts)))
+	if hostErr != nil {
+		if !restate.IsTerminalError(hostErr) {
+			// suspension or cancellation: let restate handle it.
+			return hostErr
+		}
+		if err := step(ctx, "dead_letter", func(c context.Context) error {
+			return fleet.DestroyFailed(c, vmID, req.Reason, hostErr.Error())
+		}); err != nil {
+			return err
+		}
+		return hostErr
+	}
+
+	if err := step(ctx, "unpublish_ingress", func(c context.Context) error {
+		return fleet.DestroyUnpublishIngress(c, vmID)
+	}); err != nil {
+		return err
+	}
+	return step(ctx, "finish", func(c context.Context) error {
+		return fleet.DestroyFinish(c, vmID, req.Reason)
+	})
+}
+
+// step is Run for a step with no result.
+func step(ctx restate.Context, name string, fn func(context.Context) error, opts ...restate.RunOption) error {
+	_, err := Run(ctx, opDestroy, name, func(c context.Context) (restate.Void, error) {
+		return restate.Void{}, fn(c)
+	}, opts...)
+	return err
+}
+
+// Destroy runs Environment/{vmID}.Destroy and waits for it. the invocation
+// keeps running if ctx ends first.
+func (c *Client) Destroy(ctx context.Context, vmID string, req orchestrator.DestroyRequest) error {
+	_, err := ingress.Object[orchestrator.DestroyRequest, restate.Void](c.ingress, EnvironmentObject, vmID, "Destroy").
+		Request(ctx, req)
+	return err
+}
+
+// SendDestroy enqueues Environment/{vmID}.Destroy without waiting.
+func (c *Client) SendDestroy(ctx context.Context, vmID string, req orchestrator.DestroyRequest) error {
+	_, err := ingress.ObjectSend[orchestrator.DestroyRequest](c.ingress, EnvironmentObject, vmID, "Destroy").
+		Send(ctx, req)
+	return err
+}