diff --git a/cli/render.go b/cli/render.go index 490b5886..7692425b 100644 --- a/cli/render.go +++ b/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) diff --git a/docs/content/docs/concepts/environments.mdx b/docs/content/docs/concepts/environments.mdx index 16b210b5..a5a58dc2 100644 --- a/docs/content/docs/concepts/environments.mdx +++ b/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 diff --git a/docs/content/docs/concepts/reconcile.mdx b/docs/content/docs/concepts/reconcile.mdx index 4168eba5..cd03f069 100644 --- a/docs/content/docs/concepts/reconcile.mdx +++ b/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 diff --git a/docs/content/reference/cli/environment/list.mdx b/docs/content/reference/cli/environment/list.mdx index e4f1b2d9..46772269 100644 --- a/docs/content/reference/cli/environment/list.mdx +++ b/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. | diff --git a/docs/content/reference/go/environments.mdx b/docs/content/reference/go/environments.mdx index 45b02b02..596dc831 100644 --- a/docs/content/reference/go/environments.mdx +++ b/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 diff --git a/docs/content/reference/python/environments.mdx b/docs/content/reference/python/environments.mdx index 223b98c4..63a44fc6 100644 --- a/docs/content/reference/python/environments.mdx +++ b/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" diff --git a/docs/content/reference/typescript/environments.mdx b/docs/content/reference/typescript/environments.mdx index c6feab5b..86fd1783 100644 --- a/docs/content/reference/typescript/environments.mdx +++ b/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; diff --git a/docs/runbooks/restate.mdx b/docs/runbooks/restate.mdx index 62171796..bdadccd8 100644 --- a/docs/runbooks/restate.mdx +++ b/docs/runbooks/restate.mdx @@ -126,11 +126,20 @@ docker exec fuse-restate restate invocations cancel # or kill or the ui at `http://127.0.0.1:9070/ui` (through the ssh tunnel). +destroys run as `Environment//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`. diff --git a/internal/api/docs/swagger.json b/internal/api/docs/swagger.json index a9eeb3e8..a7e36bd8 100644 --- a/internal/api/docs/swagger.json +++ b/internal/api/docs/swagger.json @@ -201,7 +201,8 @@ "enum": [ "provisioning", "running", - "destroying" + "destroying", + "destroy_failed" ], "type": "string", "description": "Filter by state", diff --git a/internal/api/docs/swagger.yaml b/internal/api/docs/swagger.yaml index b522c607..0f45c37a 100644 --- a/internal/api/docs/swagger.yaml +++ b/internal/api/docs/swagger.yaml @@ -452,6 +452,7 @@ paths: - provisioning - running - destroying + - destroy_failed in: query name: state type: string diff --git a/internal/api/handlers.go b/internal/api/handlers.go index 7dd6f0db..c7606451 100644 --- a/internal/api/handlers.go +++ b/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" diff --git a/internal/durable/destroy.go b/internal/durable/destroy.go new file mode 100644 index 00000000..f9e190e0 --- /dev/null +++ b/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 +} diff --git a/internal/durable/destroy_test.go b/internal/durable/destroy_test.go new file mode 100644 index 00000000..343fce21 --- /dev/null +++ b/internal/durable/destroy_test.go @@ -0,0 +1,204 @@ +package durable + +import ( + "context" + "errors" + "fmt" + "net" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/folsomintel/fuse/internal/orchestrator" +) + +// scriptedFleet records the destroy steps it is driven through. the host +// destroy fails hostFails times per vm before it succeeds; -1 never succeeds. +type scriptedFleet struct { + fakeFleet + hostFails int + + mu sync.Mutex + tracked map[string]bool + calls map[string][]string + attempts map[string]int + failed map[string]string +} + +func newScriptedFleet(hostFails int, vms ...string) *scriptedFleet { + f := &scriptedFleet{ + hostFails: hostFails, + tracked: map[string]bool{}, + calls: map[string][]string{}, + attempts: map[string]int{}, + failed: map[string]string{}, + } + for _, v := range vms { + f.tracked[v] = true + } + return f +} + +func (f *scriptedFleet) record(vm, step string) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls[vm] = append(f.calls[vm], step) +} + +func (f *scriptedFleet) steps(vm string) string { + f.mu.Lock() + defer f.mu.Unlock() + return strings.Join(f.calls[vm], ",") +} + +func (f *scriptedFleet) DestroyBegin(_ context.Context, vm, _ string) (orchestrator.DestroyTarget, error) { + f.record(vm, "begin") + f.mu.Lock() + defer f.mu.Unlock() + return orchestrator.DestroyTarget{Tracked: f.tracked[vm], HostID: "h1"}, nil +} + +func (f *scriptedFleet) DestroyReleaseEgress(_ context.Context, vm string) error { + f.record(vm, "egress") + return nil +} + +func (f *scriptedFleet) DestroyOnHost(_ context.Context, vm, _ string) error { + f.record(vm, "host") + f.mu.Lock() + defer f.mu.Unlock() + f.attempts[vm]++ + if f.hostFails < 0 || f.attempts[vm] <= f.hostFails { + return fmt.Errorf("host down (attempt %d)", f.attempts[vm]) + } + return nil +} + +func (f *scriptedFleet) DestroyUnpublishIngress(_ context.Context, vm string) error { + f.record(vm, "ingress") + return nil +} + +func (f *scriptedFleet) DestroyFinish(_ context.Context, vm, _ string) error { + f.record(vm, "finish") + f.mu.Lock() + defer f.mu.Unlock() + delete(f.tracked, vm) + return nil +} + +func (f *scriptedFleet) DestroyFailed(_ context.Context, vm, _, lastErr string) error { + f.record(vm, "dead_letter") + f.mu.Lock() + defer f.mu.Unlock() + f.failed[vm] = lastErr + return nil +} + +// serveFleet serves fleet behind the restate-server named by the test env, +// registers it, and returns a client. skips without restate-server. +func serveFleet(t *testing.T, fleet Fleet, policy RetryPolicy) *Client { + t.Helper() + admin := os.Getenv("FUSE_RESTATE_TEST_ADMIN_URL") + ingressURL := os.Getenv("FUSE_RESTATE_TEST_INGRESS_URL") + if admin == "" || ingressURL == "" { + t.Skip("set FUSE_RESTATE_TEST_ADMIN_URL and FUSE_RESTATE_TEST_INGRESS_URL to run against restate-server") + } + host := os.Getenv("FUSE_RESTATE_TEST_HANDLER_HOST") + if host == "" { + host = "127.0.0.1" + } + ln, err := net.Listen("tcp", net.JoinHostPort(host, "0")) + if err != nil { + t.Fatal(err) + } + addr := ln.Addr().String() + _ = ln.Close() + + cfg := Config{Mode: ModeSelf, Listen: addr, IngressURL: ingressURL, AdminURL: admin, + Retry: map[string]RetryPolicy{EnvironmentObject: policy, HostObject: DefaultRetryPolicy}} + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { _ = NewEndpoint(cfg, fleet, testLogger).Start(ctx, addr) }() + waitListening(t, addr) + + client := NewClient(cfg) + if err := client.Register(ctx, "http://"+addr); err != nil { + t.Fatalf("register: %v", err) + } + return client +} + +var fastPolicy = RetryPolicy{InitialInterval: 20 * time.Millisecond, MaxInterval: 50 * time.Millisecond, MaxAttempts: 4} + +// TestDestroyThroughRestate: a host that fails twice is retried inside the +// host step until it succeeds, and the earlier steps are not run again. +func TestDestroyThroughRestate(t *testing.T) { + vm := fmt.Sprintf("fuse-retry-%d", time.Now().UnixNano()) + fleet := newScriptedFleet(2, vm) + client := serveFleet(t, fleet, fastPolicy) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := client.Destroy(ctx, vm, orchestrator.DestroyRequest{Reason: "manual_destroy"}); err != nil { + t.Fatalf("destroy: %v", err) + } + if got, want := fleet.steps(vm), "begin,egress,host,host,host,ingress,finish"; got != want { + t.Fatalf("steps %s, want %s", got, want) + } + + // a second destroy of the same vm finds it gone and does nothing more. + if err := client.Destroy(ctx, vm, orchestrator.DestroyRequest{Reason: "manual_destroy"}); err != nil { + t.Fatalf("second destroy: %v", err) + } + if got, want := fleet.steps(vm), "begin,egress,host,host,host,ingress,finish,begin"; got != want { + t.Fatalf("steps after the repeat %s, want %s", got, want) + } +} + +// TestDestroyDeadLettersThroughRestate: a host that never recovers ends in +// a dead letter with the last error, and the caller gets that error. +func TestDestroyDeadLettersThroughRestate(t *testing.T) { + vm := fmt.Sprintf("fuse-dead-%d", time.Now().UnixNano()) + fleet := newScriptedFleet(-1, vm) + client := serveFleet(t, fleet, fastPolicy) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + err := client.Destroy(ctx, vm, orchestrator.DestroyRequest{Reason: "manual_destroy"}) + if err == nil { + t.Fatal("destroy succeeded against a host that never recovers") + } + if errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("destroy never finished: %v", err) + } + want := "begin,egress," + strings.Repeat("host,", fastPolicy.MaxAttempts) + "dead_letter" + if got := fleet.steps(vm); got != want { + t.Fatalf("steps %s, want %s", got, want) + } + fleet.mu.Lock() + lastErr := fleet.failed[vm] + fleet.mu.Unlock() + if !strings.Contains(lastErr, "host down") { + t.Fatalf("dead letter reason %q, want the host error", lastErr) + } +} + +// TestDestroyCleanupOfUntrackedVM: a create cleanup names the host, so a vm +// the fleet already dropped is still destroyed there. +func TestDestroyCleanupOfUntrackedVM(t *testing.T) { + vm := fmt.Sprintf("fuse-cleanup-%d", time.Now().UnixNano()) + fleet := newScriptedFleet(0) // not tracked + client := serveFleet(t, fleet, fastPolicy) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := client.Destroy(ctx, vm, orchestrator.DestroyRequest{Reason: "create_cleanup", HostID: "h1"}); err != nil { + t.Fatalf("destroy: %v", err) + } + if got, want := fleet.steps(vm), "begin,egress,host,ingress,finish"; got != want { + t.Fatalf("steps %s, want %s", got, want) + } +} diff --git a/internal/durable/server.go b/internal/durable/server.go index af482fcb..ed727e88 100644 --- a/internal/durable/server.go +++ b/internal/durable/server.go @@ -34,6 +34,7 @@ var services = []string{EnvironmentObject, HostObject} type Fleet interface { GetVM(vmID string) (orchestrator.VMInfo, bool) GetHost(hostID string) (orchestrator.Host, bool) + Destroyer } // PingResult is what Ping returns. it never carries more than whether the @@ -54,6 +55,10 @@ func NewEndpoint(cfg Config, fleet Fleet, logger *slog.Logger) *server.Restate { _, ok := fleet.GetVM(key) return ok }) + })). + Handler("Destroy", restate.NewObjectHandler( + func(ctx restate.ObjectContext, req orchestrator.DestroyRequest) (restate.Void, error) { + return restate.Void{}, destroy(ctx, fleet, servicePolicy(cfg, EnvironmentObject), req) })) host := restate.NewObject(HostObject, retryOption(cfg, HostObject)). @@ -93,10 +98,7 @@ func ping(ctx restate.ObjectSharedContext, object string, known func(string) boo // retryOption turns the service's configured policy into its restate option. func retryOption(cfg Config, service string) restate.ServiceDefinitionOption { - p, ok := cfg.Retry[service] - if !ok { - p = DefaultRetryPolicy - } + p := servicePolicy(cfg, service) return restate.WithInvocationRetryPolicy( restate.WithInitialRetryInterval(p.InitialInterval), restate.WithMaxRetryInterval(p.MaxInterval), @@ -104,6 +106,14 @@ func retryOption(cfg Config, service string) restate.ServiceDefinitionOption { ) } +// servicePolicy is the service's configured retry policy, or the default. +func servicePolicy(cfg Config, service string) RetryPolicy { + if p, ok := cfg.Retry[service]; ok { + return p + } + return DefaultRetryPolicy +} + // Serve runs the handler endpoint until ctx is cancelled: on cfg.Listen, or // over the outbound tunnel in mode cloud with FUSE_RESTATE_CONNECT=tunnel. // mode cloud also registers the deployment; in mode self the installer does. diff --git a/internal/durable/server_test.go b/internal/durable/server_test.go index f0320eeb..23f913ed 100644 --- a/internal/durable/server_test.go +++ b/internal/durable/server_test.go @@ -5,6 +5,7 @@ import ( "crypto/ed25519" "encoding/json" "errors" + "fmt" "io" "log/slog" "net" @@ -33,6 +34,15 @@ func (fakeFleet) GetHost(id string) (orchestrator.Host, bool) { return orchestrator.Host{ID: id, Token: "secret"}, id == "host-known" } +func (fakeFleet) DestroyBegin(context.Context, string, string) (orchestrator.DestroyTarget, error) { + return orchestrator.DestroyTarget{}, nil +} +func (fakeFleet) DestroyReleaseEgress(context.Context, string) error { return nil } +func (fakeFleet) DestroyOnHost(context.Context, string, string) error { return nil } +func (fakeFleet) DestroyUnpublishIngress(context.Context, string) error { return nil } +func (fakeFleet) DestroyFinish(context.Context, string, string) error { return nil } +func (fakeFleet) DestroyFailed(context.Context, string, string, string) error { return nil } + var testLogger = slog.New(slog.NewTextHandler(io.Discard, nil)) // discovery is the manifest restate-server reads on registration. checking it @@ -84,8 +94,16 @@ func TestDiscoveryManifest(t *testing.T) { if s.Ty != "VIRTUAL_OBJECT" { t.Errorf("%s: type %q, want VIRTUAL_OBJECT", s.Name, s.Ty) } - if len(s.Handlers) != 1 || s.Handlers[0].Name != "Ping" || s.Handlers[0].Ty != "SHARED" { - t.Errorf("%s: handlers %+v, want one shared Ping", s.Name, s.Handlers) + handlers := map[string]string{} + for _, h := range s.Handlers { + handlers[h.Name] = h.Ty + } + want := map[string]string{"Ping": "SHARED"} + if s.Name == EnvironmentObject { + want["Destroy"] = "EXCLUSIVE" + } + if fmt.Sprint(handlers) != fmt.Sprint(want) { + t.Errorf("%s: handlers %v, want %v", s.Name, handlers, want) } var p RetryPolicy if s.RetryPolicyMaxAttempts != nil { diff --git a/internal/entspike/ent/migrate/schema.go b/internal/entspike/ent/migrate/schema.go index acdfd7a1..d59d261d 100644 --- a/internal/entspike/ent/migrate/schema.go +++ b/internal/entspike/ent/migrate/schema.go @@ -59,7 +59,7 @@ var ( {Name: "vm_id", Type: field.TypeString}, {Name: "host_id", Type: field.TypeString, Default: ""}, {Name: "network_host", Type: field.TypeString, Default: ""}, - {Name: "state", Type: field.TypeEnum, Enums: []string{"provisioning", "running", "draining", "destroying", "destroyed", "failed"}}, + {Name: "state", Type: field.TypeEnum, Enums: []string{"provisioning", "running", "draining", "destroying", "destroy_failed", "destroyed", "failed"}}, {Name: "url", Type: field.TypeString, Default: ""}, {Name: "task_id", Type: field.TypeString, Default: ""}, {Name: "tenant_id", Type: field.TypeString, Default: ""}, diff --git a/internal/entspike/ent/schema/vm.go b/internal/entspike/ent/schema/vm.go index 12424e52..b5376c44 100644 --- a/internal/entspike/ent/schema/vm.go +++ b/internal/entspike/ent/schema/vm.go @@ -42,7 +42,7 @@ func (VM) Fields() []ent.Field { field.String("host_id").Default(""), field.String("network_host").Default(""), field.Enum("state").Values( - "provisioning", "running", "draining", "destroying", + "provisioning", "running", "draining", "destroying", "destroy_failed", // terminal states written by the event path "destroyed", "failed", ), diff --git a/internal/entspike/ent/vm/vm.go b/internal/entspike/ent/vm/vm.go index f8ea3ef7..2164e224 100644 --- a/internal/entspike/ent/vm/vm.go +++ b/internal/entspike/ent/vm/vm.go @@ -156,12 +156,13 @@ type State string // State values. const ( - StateProvisioning State = "provisioning" - StateRunning State = "running" - StateDraining State = "draining" - StateDestroying State = "destroying" - StateDestroyed State = "destroyed" - StateFailed State = "failed" + StateProvisioning State = "provisioning" + StateRunning State = "running" + StateDraining State = "draining" + StateDestroying State = "destroying" + StateDestroyFailed State = "destroy_failed" + StateDestroyed State = "destroyed" + StateFailed State = "failed" ) func (s State) String() string { @@ -171,7 +172,7 @@ func (s State) String() string { // StateValidator is a validator for the "state" field enum values. It is called by the builders before save. func StateValidator(s State) error { switch s { - case StateProvisioning, StateRunning, StateDraining, StateDestroying, StateDestroyed, StateFailed: + case StateProvisioning, StateRunning, StateDraining, StateDestroying, StateDestroyFailed, StateDestroyed, StateFailed: return nil default: return fmt.Errorf("vm: invalid enum value for state field: %q", s) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 74e1e631..7bfce7a7 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -40,6 +40,10 @@ type PrometheusMetrics struct { egressProvisionFailed *prometheus.CounterVec egressTeardownFailed *prometheus.CounterVec + // durable destroys that ran out of retries, through + // DestroyDeadLetterMetrics. + destroyDeadLetters prometheus.Counter + // lifecycle operations, through OperationMetrics. operations *prometheus.CounterVec operationDuration *prometheus.HistogramVec @@ -169,6 +173,11 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics { Name: "teardown_failures_total", Help: "Total proxy egress backends that failed to release on teardown, by provider.", }, []string{"provider"}), + destroyDeadLetters: prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: "orchestrator", + Name: "destroy_dead_letters_total", + Help: "Total durable destroys that ran out of retries and left their environment in destroy_failed.", + }), operations: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "orchestrator", @@ -240,6 +249,7 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics { m.egressEndpoints, m.egressProvisionFailed, m.egressTeardownFailed, + m.destroyDeadLetters, m.operations, m.operationDuration, m.stepDuration, @@ -284,6 +294,11 @@ func (m *PrometheusMetrics) EgressProvisionFailed(provider string) { m.egressProvisionFailed.WithLabelValues(provider).Inc() } +// DestroyDeadLettered implements orchestrator.DestroyDeadLetterMetrics. +func (m *PrometheusMetrics) DestroyDeadLettered() { + m.destroyDeadLetters.Inc() +} + // EgressTeardownFailed implements orchestrator.EgressMetrics. func (m *PrometheusMetrics) EgressTeardownFailed(provider string) { m.egressTeardownFailed.WithLabelValues(provider).Inc() diff --git a/internal/orchestrator/destroy_durable.go b/internal/orchestrator/destroy_durable.go new file mode 100644 index 00000000..48ece7b0 --- /dev/null +++ b/internal/orchestrator/destroy_durable.go @@ -0,0 +1,323 @@ +package orchestrator + +import ( + "context" + "fmt" + "time" +) + +// destroy reasons carried on a durable destroy. they become the reason on +// the vm.destroying and vm.removed events. +const ( + DestroyReasonManual = "manual_destroy" + DestroyReasonTaskCompleted = "task_completed" + DestroyReasonCreateCleanup = "create_cleanup" + DestroyReasonRetry = "teardown_retry" + DestroyReasonExpired = "expired" +) + +// destroyStepTimeout bounds one attempt at the host destroy, the same bound +// destroyAndRemove puts on the whole teardown. +const destroyStepTimeout = 60 * time.Second + +// maxDestroySends bounds the one-way destroy sends in flight, so a restate +// that accepts connections and never answers holds at most this many. +const maxDestroySends = 8 + +// DestroyRequest is the input of the durable Environment/{vm_id}.Destroy +// handler. +type DestroyRequest struct { + Reason string `json:"reason"` + // HostID names the host to destroy on when the fleet no longer tracks + // the vm, as after a failed create, which drops the vm before cleanup. + HostID string `json:"host_id,omitempty"` +} + +// DestroyTarget is what DestroyBegin found. +type DestroyTarget struct { + // Tracked is false when the fleet has no record of the vm. + Tracked bool `json:"tracked"` + HostID string `json:"host_id"` + // Replaced is set on a create cleanup that finds the vm id tracked + // again: a retried create reused it, and the vm there is not the one + // the cleanup was sent for. the destroy stops without touching it. + Replaced bool `json:"replaced,omitempty"` +} + +// DurableDestroyer runs a destroy as the restate Environment/{vm_id}.Destroy +// handler. *durable.Client satisfies it. Destroy waits for the result, Send +// only enqueues it. +type DurableDestroyer interface { + Destroy(ctx context.Context, vmID string, req DestroyRequest) error + SendDestroy(ctx context.Context, vmID string, req DestroyRequest) error +} + +// DestroyDeadLetterMetrics is an optional extension of ReconcileMetrics, +// counting durable destroys that ran out of retries. +type DestroyDeadLetterMetrics interface { + DestroyDeadLettered() +} + +// SetDurableDestroyer routes every destroy through restate. nil (mode off, +// the default) keeps the in-process teardown. set it before Start. +func (fm *FleetManager) SetDurableDestroyer(d DurableDestroyer) { + fm.mu.Lock() + fm.durableDestroy = d + if fm.destroySendSlots == nil { + fm.destroySendSlots = make(chan struct{}, maxDestroySends) + } + fm.mu.Unlock() +} + +func (fm *FleetManager) durableDestroyer() DurableDestroyer { + fm.mu.RLock() + defer fm.mu.RUnlock() + return fm.durableDestroy +} + +// destroyAsync starts a teardown without waiting for it: a one-way send to +// restate, or the in-process destroyAndRemove in mode off. the send runs off +// the caller's goroutine, at most maxDestroySends at a time, so a slow restate +// never holds up reconcile. a send that fails leaves the vm in destroying, +// where reconcile re-sends it. +func (fm *FleetManager) destroyAsync(vmID, reason string) { + d := fm.durableDestroyer() + if d == nil { + go fm.destroyAndRemove(vmID) + return + } + fm.destroySends.Add(1) + go func() { + defer fm.destroySends.Done() + fm.destroySendSlots <- struct{}{} + defer func() { <-fm.destroySendSlots }() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := d.SendDestroy(ctx, vmID, DestroyRequest{Reason: reason}); err != nil { + fm.logger.Error("send durable destroy failed, reconcile will retry", "vm", vmID, "reason", reason, "err", err) + } + }() +} + +// cleanupFailedCreate tears down a vm a failed create already dropped from +// the fleet. v is the dropped vm when its egress needs releasing, or nil. +func (fm *FleetManager) cleanupFailedCreate(provider Provider, v *vm, vmID, hostID string) { + if d := fm.durableDestroyer(); d != nil { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := d.SendDestroy(ctx, vmID, DestroyRequest{Reason: DestroyReasonCreateCleanup, HostID: hostID}); err != nil { + fm.logger.Error("send durable cleanup destroy failed", "vm", vmID, "host", hostID, "err", err) + } + return + } + // TODO: Track this goroutine so Stop() can wait for it to finish + // instead of leaking on shutdown. + go func() { + if v != nil { + fm.releaseEgress(context.Background(), v) + } + if err := provider.Destroy(context.Background(), vmID); err != nil { + fm.logger.Error("cleanup destroy failed", "vm", vmID, "err", err) + } + }() +} + +// durableDestroyVM is DestroyVM in mode self or cloud. it checks what the +// in-process path checks before changing anything, then runs the handler and +// waits for it. a caller that gives up does not stop the destroy. +func (fm *FleetManager) durableDestroyVM(ctx context.Context, d DurableDestroyer, vmID string) error { + fm.mu.RLock() + v, ok := fm.vms[vmID] + hostID := "" + if ok { + hostID = v.hostID + } + fm.mu.RUnlock() + if !ok { + return fmt.Errorf("%w: %s", ErrVMNotFound, vmID) + } + if _, err := fm.providerForVM(hostID); err != nil { + return err + } + if err := d.Destroy(ctx, vmID, DestroyRequest{Reason: DestroyReasonManual}); err != nil { + return fmt.Errorf("destroy vm %s: %w", vmID, err) + } + return nil +} + +// the methods below are the steps of the durable destroy, in order. each runs +// inside a journaled restate step and must be safe to run again: a crash +// after a step's effect and before its journal entry replays it. + +// DestroyBegin marks the vm destroying and fails its task. a vm the fleet does +// not track is reported untracked, not an error. +func (fm *FleetManager) DestroyBegin(ctx context.Context, vmID, reason string) (DestroyTarget, error) { + fm.mu.RLock() + v, ok := fm.vms[vmID] + if !ok { + fm.mu.RUnlock() + return DestroyTarget{}, nil + } + // a failed create drops its vm before sending the cleanup, so a tracked + // vm here is a later create's. + if reason == DestroyReasonCreateCleanup { + fm.mu.RUnlock() + fm.logger.InfoContext(ctx, "create cleanup skipped, vm id reused by a later create", "vm", vmID) + return DestroyTarget{Replaced: true}, nil + } + target := DestroyTarget{Tracked: true, HostID: v.hostID} + priorTaskID := v.taskID + assignedAt := v.createdAt + // CompleteTask and the expiry detectors mark the vm destroying and emit + // the event themselves before they send the destroy. + already := v.state == VMStateDestroying + fm.mu.RUnlock() + + // the task is failed before the vm drops it, so a crash between the two + // writes replays with the task id still on the vm. + if priorTaskID != "" && fm.store != nil { + if err := fm.store.UpsertTask(ctx, TaskRecord{ + TaskID: priorTaskID, + VMID: vmID, + RunStatus: TaskRunFailed, + LastError: "vm force destroyed", + AssignedAt: assignedAt, + UpdatedAt: time.Now(), + }); err != nil { + return DestroyTarget{}, fmt.Errorf("persist task %s destroy state: %w", priorTaskID, err) + } + fm.appendEvent(ctx, "task", priorTaskID, "task.failed", map[string]any{"vm_id": vmID, "error": "vm force destroyed"}) + } + + fm.mu.Lock() + priorState := v.state + v.state = VMStateDestroying + v.taskID = "" + v.updatedAt = time.Now() + fm.mu.Unlock() + if err := fm.persistVMByID(ctx, vmID); err != nil { + // put the state back, so the retry still sees a vm that has not + // started destroying and sends the vm.destroying event. + fm.mu.Lock() + v.state = priorState + fm.mu.Unlock() + return DestroyTarget{}, fmt.Errorf("persist vm %s destroying state: %w", vmID, err) + } + if !already { + fm.appendEvent(ctx, "vm", vmID, "vm.destroying", map[string]any{"reason": reason}) + fm.publishStateChange(vmID, "") + } + fm.logger.InfoContext(ctx, "destroying vm", "vm", vmID, "reason", reason) + return target, nil +} + +// DestroyReleaseEgress releases the vm's proxy backend. best effort, like +// every teardown's egress release: it never fails the destroy. +func (fm *FleetManager) DestroyReleaseEgress(ctx context.Context, vmID string) error { + fm.mu.RLock() + v, ok := fm.vms[vmID] + fm.mu.RUnlock() + if ok { + fm.releaseEgress(ctx, v) + } else { + fm.releaseOrphanEgress(ctx, vmID) + } + return nil +} + +// DestroyOnHost destroys the vm on its host. a vm the host does not have is +// already destroyed. an attempt restarts the vm's teardown clock, so +// reconcile does not re-send a destroy that is still retrying. +func (fm *FleetManager) DestroyOnHost(ctx context.Context, vmID, hostID string) error { + fm.mu.Lock() + if v, ok := fm.vms[vmID]; ok { + v.updatedAt = time.Now() + } + fm.mu.Unlock() + provider, err := fm.providerForVM(hostID) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, destroyStepTimeout) + defer cancel() + if err := provider.Destroy(ctx, vmID); err != nil && !IsNotFound(err) { + fm.logger.WarnContext(ctx, "destroy attempt failed", "vm", vmID, "host", hostID, "err", err) + return err + } + return nil +} + +// DestroyUnpublishIngress removes the vm's ingress routes. +func (fm *FleetManager) DestroyUnpublishIngress(ctx context.Context, vmID string) error { + fm.unpublishIngress(ctx, vmID) + return nil +} + +// DestroyFinish returns the vm's host allocation, deletes its row and closes +// out what belonged to it: pending interventions, grants and scoped keys. +func (fm *FleetManager) DestroyFinish(ctx context.Context, vmID, reason string) error { + // the row goes first: a failed delete leaves the vm tracked, so the + // retry still finds it and sends the vm.removed and terminal events. + if fm.store != nil { + if err := fm.store.DeleteVM(ctx, vmID); err != nil { + return fmt.Errorf("delete vm %s row: %w", vmID, err) + } + } + fm.mu.Lock() + v, ok := fm.vms[vmID] + if ok { + if v.hostID != "" { + fm.deallocateOnHost(v.hostID, v.spec) + } + delete(fm.vms, vmID) + } + fm.mu.Unlock() + if ok { + fm.appendEvent(ctx, "vm", vmID, "vm.removed", map[string]any{"reason": reason}) + fm.publishTerminalEvent(vmID, VMStateDestroyed, "") + } + fm.cancelEnvInterventions(ctx, v, vmID, "environment destroyed") + fm.dropEnvGrants(ctx, vmID) + fm.revokeEnvKeys(ctx, vmID) + return nil +} + +// DestroyFailed records a destroy that ran out of retries: the vm moves to +// destroy_failed with the last error, and a dead letter is written. a later +// DELETE retries it. unlike reconcile's best-effort recordDeadLetter, a failed +// dead-letter write fails the step, so restate retries it. +func (fm *FleetManager) DestroyFailed(ctx context.Context, vmID, reason, lastErr string) error { + fm.mu.Lock() + v, ok := fm.vms[vmID] + if ok { + v.state = VMStateDestroyFailed + v.err = lastErr + v.updatedAt = time.Now() + } + fm.mu.Unlock() + if ok { + if err := fm.persistVMByID(ctx, vmID); err != nil { + return fmt.Errorf("persist vm %s destroy_failed state: %w", vmID, err) + } + fm.publishStateChange(vmID, "") + } + if fm.store != nil { + now := time.Now() + if err := fm.store.UpsertDeadLetter(ctx, DeadLetterRecord{ + Kind: DeadLetterDestroy, + EntityID: vmID, + Reason: lastErr, + Payload: dlPayload("reason", reason), + FirstSeenAt: now, + LastSeenAt: now, + }); err != nil { + return fmt.Errorf("record destroy dead letter for vm %s: %w", vmID, err) + } + } + fm.appendEvent(ctx, "vm", vmID, "vm.destroy_failed", map[string]any{"reason": reason, "error": lastErr}) + if m, ok := fm.metrics.(DestroyDeadLetterMetrics); ok { + m.DestroyDeadLettered() + } + fm.logger.ErrorContext(ctx, "destroy dead-lettered", "vm", vmID, "reason", reason, "err", lastErr) + return nil +} diff --git a/internal/orchestrator/destroy_durable_test.go b/internal/orchestrator/destroy_durable_test.go new file mode 100644 index 00000000..155d5568 --- /dev/null +++ b/internal/orchestrator/destroy_durable_test.go @@ -0,0 +1,437 @@ +package orchestrator + +import ( + "context" + "errors" + "fmt" + "net/http" + "sync" + "testing" + "time" +) + +// inlineDestroyer stands in for restate: it runs the destroy steps in order +// on the fleet, the way the Environment.Destroy handler does, and records +// every request it gets. +type inlineDestroyer struct { + fm *FleetManager + + mu sync.Mutex + sent []DestroyRequest + sync []DestroyRequest +} + +func (d *inlineDestroyer) run(ctx context.Context, vmID string, req DestroyRequest) error { + target, err := d.fm.DestroyBegin(ctx, vmID, req.Reason) + if err != nil { + return err + } + if target.Replaced { + return nil + } + hostID := target.HostID + if !target.Tracked { + if req.HostID == "" { + return nil + } + hostID = req.HostID + } + _ = d.fm.DestroyReleaseEgress(ctx, vmID) + if err := d.fm.DestroyOnHost(ctx, vmID, hostID); err != nil { + _ = d.fm.DestroyFailed(ctx, vmID, req.Reason, err.Error()) + return err + } + _ = d.fm.DestroyUnpublishIngress(ctx, vmID) + return d.fm.DestroyFinish(ctx, vmID, req.Reason) +} + +func (d *inlineDestroyer) Destroy(ctx context.Context, vmID string, req DestroyRequest) error { + d.mu.Lock() + d.sync = append(d.sync, req) + d.mu.Unlock() + return d.run(ctx, vmID, req) +} + +func (d *inlineDestroyer) SendDestroy(ctx context.Context, vmID string, req DestroyRequest) error { + d.mu.Lock() + d.sent = append(d.sent, req) + d.mu.Unlock() + return d.run(ctx, vmID, req) +} + +func (d *inlineDestroyer) requests() (sent, sync []DestroyRequest) { + d.mu.Lock() + defer d.mu.Unlock() + return append([]DestroyRequest(nil), d.sent...), append([]DestroyRequest(nil), d.sync...) +} + +// destroyErrProvider fails every Destroy with err. +type destroyErrProvider struct { + *mockProvider + err error +} + +func (p *destroyErrProvider) Destroy(context.Context, string) error { return p.err } + +func newDurableFleet(t *testing.T, p Provider) (*FleetManager, *inlineDestroyer, *MemoryStateStore) { + t.Helper() + store := NewMemoryStateStore() + fm := NewFleetManager(FleetConfig{Provider: p, StateStore: store, Prefix: "fuse-"}) + d := &inlineDestroyer{fm: fm} + fm.SetDurableDestroyer(d) + return fm, d, store +} + +func TestDurableDestroyVM(t *testing.T) { + p := newMockProvider() + fm, d, store := newDurableFleet(t, p) + ctx := context.Background() + if _, err := fm.ProvisionAndAssign(ctx, "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err != nil { + t.Fatal(err) + } + + if err := fm.DestroyVM(ctx, "fuse-task-1"); err != nil { + t.Fatalf("destroy: %v", err) + } + if _, sync := d.requests(); len(sync) != 1 || sync[0].Reason != DestroyReasonManual { + t.Fatalf("durable requests %+v, want one manual destroy", sync) + } + if p.count() != 0 || len(fm.ListFleet()) != 0 { + t.Fatalf("provider has %d envs, fleet %d vms; want both empty", p.count(), len(fm.ListFleet())) + } + vms, _ := store.ListVMs(ctx) + if len(vms) != 0 { + t.Fatalf("store still has %d vm rows", len(vms)) + } + tasks, _ := store.ListTasks(ctx) + if len(tasks) != 1 || tasks[0].RunStatus != TaskRunFailed { + t.Fatalf("tasks %+v, want task-1 failed", tasks) + } +} + +// blockingDestroyer holds every send until release is closed. +type blockingDestroyer struct { + inlineDestroyer + release chan struct{} + mu sync.Mutex + active int + peak int +} + +func (d *blockingDestroyer) SendDestroy(ctx context.Context, vmID string, req DestroyRequest) error { + d.mu.Lock() + d.active++ + if d.active > d.peak { + d.peak = d.active + } + d.mu.Unlock() + <-d.release + d.mu.Lock() + d.active-- + d.mu.Unlock() + return nil +} + +// a restate that never answers must not hold up the caller, and holds at +// most maxDestroySends sends at once. +func TestDestroyAsyncDoesNotBlockAndIsBounded(t *testing.T) { + fm := NewFleetManager(FleetConfig{Provider: newMockProvider(), Prefix: "fuse-"}) + d := &blockingDestroyer{release: make(chan struct{})} + fm.SetDurableDestroyer(d) + + done := make(chan struct{}) + go func() { + for i := 0; i < 3*maxDestroySends; i++ { + fm.destroyAsync(fmt.Sprintf("fuse-%d", i), DestroyReasonExpired) + } + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("destroyAsync blocked on a send") + } + time.Sleep(50 * time.Millisecond) + close(d.release) + fm.destroySends.Wait() + if d.peak > maxDestroySends { + t.Fatalf("%d sends in flight, want at most %d", d.peak, maxDestroySends) + } +} + +func TestDurableDestroyVMUnknownIsNotFound(t *testing.T) { + fm, d, _ := newDurableFleet(t, newMockProvider()) + err := fm.DestroyVM(context.Background(), "fuse-missing") + if !errors.Is(err, ErrVMNotFound) { + t.Fatalf("err %v, want ErrVMNotFound", err) + } + if _, sync := d.requests(); len(sync) != 0 { + t.Fatalf("an unknown vm reached restate: %+v", sync) + } +} + +func TestDurableCompleteTaskSendsDestroy(t *testing.T) { + p := newMockProvider() + fm, d, _ := newDurableFleet(t, p) + ctx := context.Background() + if _, err := fm.ProvisionAndAssign(ctx, "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err != nil { + t.Fatal(err) + } + if err := fm.CompleteTask("task-1"); err != nil { + t.Fatal(err) + } + fm.destroySends.Wait() + if sent, _ := d.requests(); len(sent) != 1 || sent[0].Reason != DestroyReasonTaskCompleted { + t.Fatalf("sent %+v, want one task_completed destroy", sent) + } + if p.count() != 0 || len(fm.ListFleet()) != 0 { + t.Fatal("vm survived the durable destroy") + } +} + +func TestDurableCreateCleanupCarriesHost(t *testing.T) { + p := newMockProvider() + p.createFn = func(context.Context, Spec) (Environment, error) { + return nil, errors.New("boom") + } + fm, d, _ := newDurableFleet(t, p) + if _, err := fm.ProvisionAndAssign(context.Background(), "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err == nil { + t.Fatal("provision succeeded, want an error") + } + sent, _ := d.requests() + if len(sent) != 1 || sent[0].Reason != DestroyReasonCreateCleanup { + t.Fatalf("sent %+v, want one create_cleanup destroy", sent) + } +} + +// a retried create reuses the vm id; the first attempt's cleanup arriving +// after it must not destroy the replacement. +func TestCreateCleanupSparesReplacement(t *testing.T) { + p := newMockProvider() + fm, d, _ := newDurableFleet(t, p) + ctx := context.Background() + if _, err := fm.ProvisionAndAssign(ctx, "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err != nil { + t.Fatal(err) + } + if err := d.SendDestroy(ctx, "fuse-task-1", DestroyRequest{Reason: DestroyReasonCreateCleanup, HostID: "h1"}); err != nil { + t.Fatal(err) + } + info, ok := fm.GetVM("fuse-task-1") + if !ok || info.State != VMStateRunning { + t.Fatalf("vm %+v (tracked %v), want the replacement still running", info, ok) + } + if p.count() != 1 { + t.Fatal("cleanup destroyed the replacement on the host") + } +} + +func TestDestroyOnHostTreatsNotFoundAsDone(t *testing.T) { + p := &destroyErrProvider{mockProvider: newMockProvider(), err: &HTTPStatusError{Code: http.StatusNotFound}} + fm, _, _ := newDurableFleet(t, p) + if err := fm.DestroyOnHost(context.Background(), "fuse-gone", ""); err != nil { + t.Fatalf("404 from the host: %v, want success", err) + } +} + +func TestDurableDestroyFailureDeadLetters(t *testing.T) { + mp := newMockProvider() + p := &destroyErrProvider{mockProvider: mp, err: errors.New("host down")} + fm, _, store := newDurableFleet(t, p) + ctx := context.Background() + if _, err := fm.ProvisionAndAssign(ctx, "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err != nil { + t.Fatal(err) + } + + if err := fm.DestroyVM(ctx, "fuse-task-1"); err == nil { + t.Fatal("destroy succeeded against a failing host") + } + info, ok := fm.GetVM("fuse-task-1") + if !ok || info.State != VMStateDestroyFailed { + t.Fatalf("vm %+v (tracked %v), want destroy_failed", info, ok) + } + dls, _ := store.ListDeadLetters(ctx) + if len(dls) != 1 || dls[0].Kind != DeadLetterDestroy || dls[0].Reason != "host down" { + t.Fatalf("dead letters %+v, want one destroy entry with the host error", dls) + } + + // the host recovers: a DELETE on the destroy_failed vm finishes it. + p.err = nil + if err := fm.DestroyVM(ctx, "fuse-task-1"); err != nil { + t.Fatalf("retry destroy: %v", err) + } + if _, ok := fm.GetVM("fuse-task-1"); ok { + t.Fatal("vm still tracked after the retried destroy") + } +} + +// deadLetterErrStore fails every dead-letter write. +type deadLetterErrStore struct{ *MemoryStateStore } + +func (deadLetterErrStore) UpsertDeadLetter(context.Context, DeadLetterRecord) error { + return errors.New("injected dead letter failure") +} + +// the dead letter is the record of a destroy that gave up, so a failed write +// fails the step for restate to retry instead of being dropped. +func TestDestroyFailedReturnsDeadLetterError(t *testing.T) { + fm := NewFleetManager(FleetConfig{ + Provider: newMockProvider(), + StateStore: deadLetterErrStore{NewMemoryStateStore()}, + Prefix: "fuse-", + }) + if err := fm.DestroyFailed(context.Background(), "fuse-gone", DestroyReasonManual, "host down"); err == nil { + t.Fatal("dead letter write failed but the step succeeded") + } +} + +func TestDestroyFinishIsIdempotent(t *testing.T) { + p := newMockProvider() + fm, _, _ := newDurableFleet(t, p) + ctx := context.Background() + if _, err := fm.ProvisionAndAssign(ctx, "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if err := fm.DestroyFinish(ctx, "fuse-task-1", DestroyReasonManual); err != nil { + t.Fatalf("finish %d: %v", i, err) + } + } + if _, ok := fm.GetVM("fuse-task-1"); ok { + t.Fatal("vm still tracked after finish") + } +} + +// failOnceStore fails the next DeleteVM or UpsertVM once. +type failOnceStore struct { + *MemoryStateStore + mu sync.Mutex + failDelete bool + failUpsert bool +} + +func (s *failOnceStore) DeleteVM(ctx context.Context, vmID string) error { + s.mu.Lock() + fail := s.failDelete + s.failDelete = false + s.mu.Unlock() + if fail { + return errors.New("injected delete failure") + } + return s.MemoryStateStore.DeleteVM(ctx, vmID) +} + +func (s *failOnceStore) UpsertVM(ctx context.Context, rec VMRecord) error { + s.mu.Lock() + fail := s.failUpsert + s.failUpsert = false + s.mu.Unlock() + if fail { + return errors.New("injected upsert failure") + } + return s.MemoryStateStore.UpsertVM(ctx, rec) +} + +// waitForState reads ch until an event in state want arrives. +func waitForState(t *testing.T, ch <-chan EnvironmentEvent, want VMState) { + t.Helper() + timeout := time.After(2 * time.Second) + for { + select { + case ev := <-ch: + if ev.State == want { + return + } + case <-timeout: + t.Fatalf("no %s event", want) + } + } +} + +// a step that fails a write and is retried must still send the events the +// first attempt never reached. +func TestDestroyStepRetriesKeepEvents(t *testing.T) { + store := &failOnceStore{MemoryStateStore: NewMemoryStateStore()} + fm := NewFleetManager(FleetConfig{Provider: newMockProvider(), StateStore: store, Prefix: "fuse-"}) + ctx := context.Background() + if _, err := fm.ProvisionAndAssign(ctx, "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err != nil { + t.Fatal(err) + } + ch, cancel := fm.SubscribeEnvironmentEvents("fuse-task-1") + defer cancel() + + store.mu.Lock() + store.failUpsert = true + store.mu.Unlock() + if _, err := fm.DestroyBegin(ctx, "fuse-task-1", DestroyReasonManual); err == nil { + t.Fatal("begin succeeded with a failing persist") + } + if _, err := fm.DestroyBegin(ctx, "fuse-task-1", DestroyReasonManual); err != nil { + t.Fatalf("retried begin: %v", err) + } + waitForState(t, ch, VMStateDestroying) + + store.mu.Lock() + store.failDelete = true + store.mu.Unlock() + if err := fm.DestroyFinish(ctx, "fuse-task-1", DestroyReasonManual); err == nil { + t.Fatal("finish succeeded with a failing delete") + } + if err := fm.DestroyFinish(ctx, "fuse-task-1", DestroyReasonManual); err != nil { + t.Fatalf("retried finish: %v", err) + } + waitForState(t, ch, VMStateDestroyed) +} + +func TestRetryFailedTeardownsResendsDurably(t *testing.T) { + p := newMockProvider() + fm, d, _ := newDurableFleet(t, p) + ctx := context.Background() + if _, err := fm.ProvisionAndAssign(ctx, "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err != nil { + t.Fatal(err) + } + // a vm left destroying long ago, e.g. by a mode off release before the + // upgrade, with nothing in restate working on it. + fm.mu.Lock() + v := fm.vms["fuse-task-1"] + v.state = VMStateDestroying + v.updatedAt = time.Now().Add(-2 * teardownRetryAfter) + fm.mu.Unlock() + + fm.retryFailedTeardowns(map[string]bool{"fuse-task-1": true}) + fm.destroySends.Wait() + if sent, _ := d.requests(); len(sent) != 1 || sent[0].Reason != DestroyReasonRetry { + t.Fatalf("sent %+v, want one teardown_retry destroy", sent) + } + if p.count() != 0 { + t.Fatal("re-sent destroy did not reach the host") + } +} + +// a destroy_failed vm that later vanishes from its host is reaped with its +// keys revoked: the failed destroy never reached DestroyFinish to do it. +func TestReapDestroyFailedRevokesKeys(t *testing.T) { + p := &destroyErrProvider{mockProvider: newMockProvider(), err: errors.New("host down")} + fm, _, _ := newDurableFleet(t, p) + keys := &fakeEnvKeys{} + fm.SetEnvKeyStore(keys) + ctx := context.Background() + if _, err := fm.ProvisionAndAssign(ctx, "task-1", Spec{}, []byte(`{}`), nil, BootOptions{}); err != nil { + t.Fatal(err) + } + if err := fm.DestroyVM(ctx, "fuse-task-1"); err == nil { + t.Fatal("destroy succeeded against a failing host") + } + if len(keys.revoked) != 0 { + t.Fatalf("keys revoked before the vm is gone: %v", keys.revoked) + } + + fm.reapDestroyedVMs(ctx, map[string]bool{}) + if _, ok := fm.GetVM("fuse-task-1"); ok { + t.Fatal("destroy_failed vm not reaped") + } + keys.mu.Lock() + defer keys.mu.Unlock() + if len(keys.revoked) != 1 || keys.revoked[0] != "fuse-task-1" { + t.Fatalf("revoked %v, want the reaped vm", keys.revoked) + } +} diff --git a/internal/orchestrator/fleet.go b/internal/orchestrator/fleet.go index 58c24749..50b11b08 100644 --- a/internal/orchestrator/fleet.go +++ b/internal/orchestrator/fleet.go @@ -140,7 +140,8 @@ var ( // VMState represents the lifecycle state of a managed VM. // // Stored states (recorded on a vm and persisted to the state store): -// - VMStateProvisioning, VMStateRunning, VMStateDraining, VMStateDestroying. +// - VMStateProvisioning, VMStateRunning, VMStateDraining, VMStateDestroying, +// VMStateDestroyFailed. // // Synthetic terminal states (emitted only over the wire as SSE events // — never stored on a vm record because the underlying record is gone @@ -161,8 +162,17 @@ const ( // supported state transition out of Draining (no "undrain"). VMStateDraining VMState = "draining" VMStateDestroying VMState = "destroying" + // VMStateDestroyFailed is a durable destroy that ran out of retries. the + // vm may still exist on its host and still holds its allocation; the last + // error is on the vm and in a dead letter, and a DELETE retries it. + VMStateDestroyFailed VMState = "destroy_failed" ) +// tearingDown reports whether s is a state a destroy has started on. +func (s VMState) tearingDown() bool { + return s == VMStateDestroying || s == VMStateDestroyFailed +} + // vm is the internal representation of a tracked VM. type vm struct { id string @@ -604,6 +614,13 @@ type FleetManager struct { publicURL string webhooks webhookSender + // durableDestroy runs destroys on restate; nil in mode off. see + // destroy_durable.go. destroySendSlots bounds the one-way sends in + // flight and destroySends tracks them. + durableDestroy DurableDestroyer + destroySendSlots chan struct{} + destroySends sync.WaitGroup + cancel context.CancelFunc done chan struct{} } @@ -1107,13 +1124,7 @@ func (fm *FleetManager) provisionAndAssign(ctx context.Context, taskID string, s fm.publishTerminalEvent(vmID, VMStateFailed, redactedErr) // Best-effort cleanup. - // TODO: Track this goroutine so Stop() can wait for it to finish - // instead of leaking on shutdown. - go func() { - if destroyErr := bootProvider.Destroy(context.Background(), vmID); destroyErr != nil { - fm.logger.Error("cleanup destroy failed", "vm", vmID, "err", destroyErr) - } - }() + fm.cleanupFailedCreate(bootProvider, nil, vmID, v.hostID) return nil, fmt.Errorf("provision vm %s: %w", vmID, err) } @@ -1178,12 +1189,7 @@ func (fm *FleetManager) provisionAndAssign(ctx context.Context, taskID string, s fm.deleteVMBackground(vmID) fm.publishTerminalEvent(vmID, VMStateFailed, errMsg) - go func() { - fm.releaseEgress(context.Background(), v) - if destroyErr := bootProvider.Destroy(context.Background(), vmID); destroyErr != nil { - fm.logger.Error("cleanup destroy failed after persist error", "vm", vmID, "err", destroyErr) - } - }() + fm.cleanupFailedCreate(bootProvider, v, vmID, v.hostID) return nil, fmt.Errorf("persist vm %s running state: %w", vmID, err) } @@ -1215,12 +1221,7 @@ func (fm *FleetManager) provisionAndAssign(ctx context.Context, taskID string, s fm.deleteVMBackground(vmID) fm.publishTerminalEvent(vmID, VMStateFailed, errMsg) - go func() { - fm.releaseEgress(context.Background(), v) - if destroyErr := bootProvider.Destroy(context.Background(), vmID); destroyErr != nil { - fm.logger.Error("cleanup destroy failed after persist error", "vm", vmID, "err", destroyErr) - } - }() + fm.cleanupFailedCreate(bootProvider, v, vmID, v.hostID) return nil, fmt.Errorf("persist task %s running state: %w", taskID, err) } @@ -1309,7 +1310,7 @@ func (fm *FleetManager) CompleteTask(taskID string) error { fm.publishStateChange(vmID, "") fm.logger.Info("completing task, destroying vm", "vm", found.id, "task", taskID) - go fm.destroyAndRemove(vmID) + fm.destroyAsync(vmID, DestroyReasonTaskCompleted) return nil } @@ -1317,7 +1318,12 @@ func (fm *FleetManager) CompleteTask(taskID string) error { // DestroyVM forcefully tears down a VM by ID. func (fm *FleetManager) DestroyVM(ctx context.Context, vmID string) error { ctx, end := fm.startOperation(ctx, "destroy", vmID) - err := fm.destroyVM(ctx, vmID) + var err error + if d := fm.durableDestroyer(); d != nil { + err = fm.durableDestroyVM(ctx, d, vmID) + } else { + err = fm.destroyVM(ctx, vmID) + } end(err) return err } @@ -1591,7 +1597,7 @@ func (fm *FleetManager) reconcile(ctx context.Context) { changedVMs := make([]string, 0) failedTasks := make([]TaskRecord, 0) for id, v := range fm.vms { - if v.state == VMStateProvisioning || v.state == VMStateDestroying { + if v.state == VMStateProvisioning || v.state.tearingDown() { continue } if !providerVMs[id] { @@ -1649,7 +1655,7 @@ func (fm *FleetManager) reconcile(ctx context.Context) { // sweep VMs that have been marked destroying and are no longer present // in the provider so they don't accumulate in memory or in the store. fm.retryFailedTeardowns(providerVMs) - fm.reapDestroyedVMs(providerVMs) + fm.reapDestroyedVMs(ctx, providerVMs) // Heal host allocation drift when an orphan is destroyed (#68). An orphan // is untracked by definition, so it has no persisted Spec for @@ -1715,12 +1721,15 @@ func (fm *FleetManager) destroyAndRemove(vmID string) { // reapDestroyedVMs removes tracked VMs that are in the Destroying state // and are no longer present in the provider (set provided by reconcile). // VM resources are returned to their host, the VM is removed from the -// in-memory fleet, and the persisted record is deleted. -func (fm *FleetManager) reapDestroyedVMs(providerVMs map[string]bool) { +// in-memory fleet, and the persisted record is deleted. what belonged to the +// vm is closed out as a finished destroy would: a destroy_failed vm has no +// destroy left running to do it. +func (fm *FleetManager) reapDestroyedVMs(ctx context.Context, providerVMs map[string]bool) { fm.mu.Lock() toRemove := make([]string, 0) + removed := make([]*vm, 0) for id, v := range fm.vms { - if v.state != VMStateDestroying { + if !v.state.tearingDown() { continue } if providerVMs[id] { @@ -1731,12 +1740,16 @@ func (fm *FleetManager) reapDestroyedVMs(providerVMs map[string]bool) { } delete(fm.vms, id) toRemove = append(toRemove, id) + removed = append(removed, v) } fm.mu.Unlock() - for _, id := range toRemove { + for i, id := range toRemove { fm.deleteVMBackground(id) fm.appendEventBackground("vm", id, "vm.removed", map[string]any{"reason": "destroyed"}) fm.publishTerminalEvent(id, VMStateDestroyed, "") + fm.cancelEnvInterventions(ctx, removed[i], id, "environment destroyed") + fm.dropEnvGrants(ctx, id) + fm.revokeEnvKeys(ctx, id) } } diff --git a/internal/orchestrator/interventions.go b/internal/orchestrator/interventions.go index ca796678..1ae0fa02 100644 --- a/internal/orchestrator/interventions.go +++ b/internal/orchestrator/interventions.go @@ -101,7 +101,7 @@ func (fm *FleetManager) RequestHuman(ctx context.Context, envID string, req Huma if !ok { return hitl.Intervention{}, fmt.Errorf("%w: %s", ErrVMNotFound, envID) } - if state == VMStateDestroying { + if state.tearingDown() { return hitl.Intervention{}, fmt.Errorf("%w: environment %s is being destroyed", ErrVMNotRunning, envID) } if spec == nil { diff --git a/internal/orchestrator/interventions_reconcile.go b/internal/orchestrator/interventions_reconcile.go index f52adf6d..b4c01162 100644 --- a/internal/orchestrator/interventions_reconcile.go +++ b/internal/orchestrator/interventions_reconcile.go @@ -35,7 +35,7 @@ func (fm *FleetManager) reconcileInterventions(ctx context.Context) { } fm.mu.RLock() v, ok := fm.vms[rec.EnvID] - gone := !ok || v.state == VMStateDestroying + gone := !ok || v.state.tearingDown() fm.mu.RUnlock() if gone { if _, err := fm.resolveIntervention(ctx, rec.ID, auditMeta{}, func(in hitl.Intervention, now time.Time) (hitl.Intervention, error) { diff --git a/internal/orchestrator/reconcile.go b/internal/orchestrator/reconcile.go index 9ce960cd..4c7b550a 100644 --- a/internal/orchestrator/reconcile.go +++ b/internal/orchestrator/reconcile.go @@ -409,7 +409,7 @@ func (fm *FleetManager) retryFailedTeardowns(providerVMs map[string]bool) { for _, id := range retry { fm.logger.Warn("vm still present after destroy, retrying teardown", "vm", id) fm.appendEventBackground("vm", id, "vm.destroy_retried", nil) - go fm.destroyAndRemove(id) + fm.destroyAsync(id, DestroyReasonRetry) } } @@ -523,7 +523,7 @@ func (fm *FleetManager) failExpiredVM(ctx context.Context, vmID, taskID string, "ceiling_s", fmt.Sprintf("%d", int(e.ceiling.Seconds())), ), }) - go fm.destroyAndRemove(vmID) + fm.destroyAsync(vmID, DestroyReasonExpired) return true } diff --git a/internal/orchestrator/state_store.go b/internal/orchestrator/state_store.go index d9b95477..dc15f836 100644 --- a/internal/orchestrator/state_store.go +++ b/internal/orchestrator/state_store.go @@ -230,6 +230,10 @@ const ( // DeadLetterDrainTimeout records an environment torn down because it // stayed in draining past the draining timeout with no DELETE. DeadLetterDrainTimeout DeadLetterKind = "drain_timeout" + + // DeadLetterDestroy records a durable destroy that ran out of retries + // and left its environment in destroy_failed. + DeadLetterDestroy DeadLetterKind = "destroy" ) // DeadLetterRecord is a failure the reconciler has given up on retrying. diff --git a/orchestrator/main.go b/orchestrator/main.go index b31dbc50..d794a1bc 100644 --- a/orchestrator/main.go +++ b/orchestrator/main.go @@ -563,6 +563,13 @@ func run() error { // the base url hitl webhook notifications link to, e.g. // https://fuse.example.com. empty leaves the link out. fm.SetPublicURL(strings.TrimRight(env("ORCH_PUBLIC_URL", ""), "/")) + // restate client, made before the fleet starts so the reconcile loop's + // first teardown already goes through restate. mode off makes none. + var restateClient *durable.Client + if durableCfg.Mode != durable.ModeOff { + restateClient = durable.NewClient(durableCfg) + fm.SetDurableDestroyer(restateClient) + } fm.Start(rootCtx) defer fm.Stop() @@ -572,9 +579,7 @@ func run() error { // restate handler endpoint. mode off starts nothing and leaves the // readiness check out. - var restateClient *durable.Client - if durableCfg.Mode != durable.ModeOff { - restateClient = durable.NewClient(durableCfg) + if restateClient != nil { go func() { if err := durable.Serve(rootCtx, durableCfg, fm, logger); err != nil { errCh <- fmt.Errorf("restate handler endpoint: %w", err) diff --git a/sdks/go/fuse_test.go b/sdks/go/fuse_test.go index 9e607327..24cce9bd 100644 --- a/sdks/go/fuse_test.go +++ b/sdks/go/fuse_test.go @@ -766,6 +766,7 @@ func TestStatePredicates(t *testing.T) { {StateRunning, false, true}, {StateDraining, false, false}, {StateDestroying, false, false}, + {StateDestroyFailed, false, false}, {StateDestroyed, true, true}, {StateFailed, true, true}, } diff --git a/sdks/go/types.go b/sdks/go/types.go index 320e2048..f0611928 100644 --- a/sdks/go/types.go +++ b/sdks/go/types.go @@ -338,8 +338,11 @@ const ( StateRunning = "running" StateDraining = "draining" StateDestroying = "destroying" - StateDestroyed = "destroyed" - StateFailed = "failed" + // StateDestroyFailed is a destroy that ran out of retries. the + // environment may still exist on its host; DELETE it again to retry. + StateDestroyFailed = "destroy_failed" + StateDestroyed = "destroyed" + StateFailed = "failed" ) // IsTerminalState reports whether state is a terminal lifecycle state. diff --git a/sdks/python/src/fuse/__init__.py b/sdks/python/src/fuse/__init__.py index 66482343..9b44b237 100644 --- a/sdks/python/src/fuse/__init__.py +++ b/sdks/python/src/fuse/__init__.py @@ -88,6 +88,7 @@ SCOPE_HITL_READ, SCOPE_HITL_SECRET, SCOPE_HITL_TAKEOVER, + STATE_DESTROY_FAILED, STATE_DESTROYED, STATE_DESTROYING, STATE_DRAINING, @@ -274,6 +275,7 @@ "STATE_RUNNING", "STATE_DRAINING", "STATE_DESTROYING", + "STATE_DESTROY_FAILED", "STATE_DESTROYED", "STATE_FAILED", "HEALTH_STARTING", diff --git a/sdks/python/src/fuse/types.py b/sdks/python/src/fuse/types.py index 11018258..ee9b7961 100644 --- a/sdks/python/src/fuse/types.py +++ b/sdks/python/src/fuse/types.py @@ -10,6 +10,9 @@ STATE_RUNNING = "running" STATE_DRAINING = "draining" STATE_DESTROYING = "destroying" +# a destroy that ran out of retries. the environment may still exist on its +# host; destroy it again to retry. +STATE_DESTROY_FAILED = "destroy_failed" STATE_DESTROYED = "destroyed" STATE_FAILED = "failed" diff --git a/sdks/python/tests/test_client.py b/sdks/python/tests/test_client.py index b3c90ee5..9240f5bb 100644 --- a/sdks/python/tests/test_client.py +++ b/sdks/python/tests/test_client.py @@ -325,6 +325,28 @@ def test_environments_exec_conflict() -> None: assert fuse.is_conflict(err) +@respx.mock +def test_environments_get_destroy_failed() -> None: + respx.get(f"{BASE_URL}/v1/environments/vm-1").mock( + return_value=httpx.Response( + 200, + json={ + "id": "vm-1", + "state": "destroy_failed", + "task_id": "", + "url": "u", + "error": "host down", + }, + ) + ) + with new_client() as client: + env = client.environments.get("vm-1") + + assert env.state == fuse.STATE_DESTROY_FAILED + # the environment still exists, so a wait for its end must keep going. + assert not fuse.is_terminal_state(env.state) + + @respx.mock def test_environments_get_with_endpoints() -> None: respx.get(f"{BASE_URL}/v1/environments/vm-1").mock( diff --git a/sdks/rust/src/types.rs b/sdks/rust/src/types.rs index 4bf4eb91..3b92a67c 100644 --- a/sdks/rust/src/types.rs +++ b/sdks/rust/src/types.rs @@ -34,6 +34,9 @@ string_enum! { Running => "running", Draining => "draining", Destroying => "destroying", + /// A destroy that ran out of retries. The environment may still + /// exist on its host; destroy it again to retry. + DestroyFailed => "destroy_failed", Destroyed => "destroyed", Failed => "failed", } diff --git a/sdks/rust/tests/wire.rs b/sdks/rust/tests/wire.rs index add96032..e7f62951 100644 --- a/sdks/rust/tests/wire.rs +++ b/sdks/rust/tests/wire.rs @@ -120,6 +120,12 @@ fn unknown_enum_values_pass_through() { assert!(!state.is_terminal()); assert_eq!(EnvironmentState::from("failed"), EnvironmentState::Failed); + // destroy_failed is known, and not terminal: the environment still exists. + assert_eq!( + EnvironmentState::from("destroy_failed"), + EnvironmentState::DestroyFailed + ); + assert!(!EnvironmentState::DestroyFailed.is_terminal()); assert!(EnvironmentState::Failed.is_terminal()); assert!(EnvironmentState::Running.is_settled()); assert!(!EnvironmentState::Provisioning.is_settled()); diff --git a/sdks/typescript/src/types.ts b/sdks/typescript/src/types.ts index 68d7047b..fef2c728 100644 --- a/sdks/typescript/src/types.ts +++ b/sdks/typescript/src/types.ts @@ -681,6 +681,8 @@ export const State = { Running: "running", Draining: "draining", Destroying: "destroying", + /** A destroy that ran out of retries. The environment may still exist on its host; destroy it again to retry. */ + DestroyFailed: "destroy_failed", Destroyed: "destroyed", Failed: "failed", } as const; diff --git a/sdks/typescript/test/environments.test.ts b/sdks/typescript/test/environments.test.ts index fd38e3b1..f9b6208b 100644 --- a/sdks/typescript/test/environments.test.ts +++ b/sdks/typescript/test/environments.test.ts @@ -1,6 +1,12 @@ import { afterEach, describe, expect, it } from "vitest"; import { pathOf, queryOf, readBody, serve, type TestServer } from "./server.js"; -import { FuseApiError, isConflict, isFuseApiError } from "../src/index.js"; +import { + FuseApiError, + isConflict, + isFuseApiError, + isTerminalState, + State, +} from "../src/index.js"; let current: TestServer | undefined; afterEach(async () => { @@ -113,6 +119,21 @@ describe("environments", () => { }); }); + it("get decodes destroy_failed as a live, non-terminal state", async () => { + current = await serve((req, res) => { + res.setHeader("Content-Type", "application/json"); + res.end( + `{"id":"vm-1","state":"destroy_failed","task_id":"","url":"u","error":"host down"}`, + ); + }); + + const env = await current.client.environments.get("vm-1"); + + expect(env.state).toBe(State.DestroyFailed); + // the environment still exists, so a wait for its end must keep going. + expect(isTerminalState(env.state)).toBe(false); + }); + it("get decodes an endpoint protocol, including one it does not know", async () => { // protocol is typed as string rather than a union, so a server that grows // a third transport does not break decoding. endpoints come back on every