diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 7746c9913d..ec1976c048 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -877,7 +877,7 @@ sequenceDiagram client-proxy resolves the node IP of a sandbox from one routing record in Redis, written by the orchestrator. Its JSON shape (`sandbox_catalog.SandboxInfo` in `packages/shared/pkg/sandbox-catalog`): `orchestrator_id`, `orchestrator_ip`, `execution_id`, `sandbox_started_at`, -`sandbox_max_length_in_hours`. +`sandbox_max_length_in_hours`, `lifecycle_id`. | Record | Key | Writer | Written | Deleted | |---|---|---|---|---| @@ -886,14 +886,25 @@ orchestrator. Its JSON shape (`sandbox_catalog.SandboxInfo` in `packages/shared/ **The orchestrator-owned record is the routing source.** The orchestrator writes `sandbox:routing:{id}` on `MarkRunning` and deletes it on `MarkStopping`. A failed write is logged and counted (`orchestrator.routing.publish.total{result=error}`); the sandbox keeps running. Build -sandboxes are skipped. The delete is guarded by `execution_id` in a Lua script, so a stale -lifecycle never removes the record of a newer execution. On a miss client-proxy goes to the -auto-resume path (`ResumeSandbox` gRPC to the API). +sandboxes are skipped. The publisher writes the actual Firecracker `lifecycle_id` for every +network mode. Its delete atomically compares all three nonempty owner fields: `execution_id`, +`lifecycle_id`, and `orchestrator_id` (the orchestrator process instance, not the API node ID). +This protects a replacement lifecycle even when a resume-fresh checkpoint retains the execution ID. +The Redis script leaves both the value and expiration unchanged on a mismatch or unverifiable +owner. Cleanup metrics distinguish deleted (`ok`), absent, superseded, unverifiable, and error. +On a miss client-proxy goes to the auto-resume path (`ResumeSandbox` gRPC to the API). + +The execution-only catalog delete remains for legacy callers, but refuses any record carrying +a `lifecycle_id` field. The full-owner delete refuses legacy records missing owner fields. Reads +remain compatible with both shapes. Older deployed binaries still contain the old deletion +script, so this protection requires upgrading publishers; a mixed-version fleet is not fenced +by the new library. Store/refresh writes remain unconditional. These ownership checks do not +prove that an old VM is dead and do not enable stale-node recovery or request replay. The TTL of the record is `sandbox_max_length_in_hours` from the write time. The record is deleted earlier in every normal stop path. -A routed sandbox adds `lifecycle_id` (its Firecracker process) and `network_placement` with the +A routed sandbox adds `network_placement` with the mode, Router, Portable IP and ingress endpoint (`ingress_address`, `ingress_port`). The endpoint is the Router for full-router and the worker for egress-router. The record carries location only. The orchestrator owns the Router configuration, which never travels through this record. An absent diff --git a/packages/orchestrator/pkg/routing/publisher.go b/packages/orchestrator/pkg/routing/publisher.go index 4e0d986b18..0a3ef66d44 100644 --- a/packages/orchestrator/pkg/routing/publisher.go +++ b/packages/orchestrator/pkg/routing/publisher.go @@ -29,7 +29,7 @@ var _ sandbox.MapSubscriber = (*Publisher)(nil) // Store is the subset of the Redis catalog the publisher needs. type Store interface { StoreSandbox(ctx context.Context, sandboxID string, info *catalog.SandboxInfo, expiration time.Duration) error - DeleteSandbox(ctx context.Context, sandboxID string, executionID string) error + DeleteSandboxIfOwner(ctx context.Context, sandboxID string, owner catalog.SandboxOwner) (catalog.DeleteOutcome, error) } var _ Store = (*catalog.RedisSandboxCatalog)(nil) @@ -120,6 +120,7 @@ func (p *Publisher) OnInsert(ctx context.Context, sbx *sandbox.Sandbox) { OrchestratorID: p.orchestratorID, OrchestratorIP: p.nodeIP, ExecutionID: sbx.Runtime.ExecutionID, + LifecycleID: sbx.LifecycleID, StartedAt: sbx.GetStartedAt(), MaxLengthInHours: sbx.Config.MaxSandboxLengthHours, } @@ -150,8 +151,8 @@ func (p *Publisher) OnInsert(ctx context.Context, sbx *sandbox.Sandbox) { } // OnStopping deletes the routing record this publisher wrote for the lifecycle. -// The delete is guarded by execution ID in Redis, so a stale lifecycle never -// removes the record of a newer execution. +// Redis compares the execution, VM lifecycle, and orchestrator process, so a +// stale stop cannot remove a replacement route within the same execution. func (p *Publisher) OnStopping(ctx context.Context, sbx *sandbox.Sandbox) { if sbx.Runtime.SandboxType != sandboxtypes.SandboxTypeSandbox { return @@ -177,8 +178,12 @@ func (p *Publisher) OnStopping(ctx context.Context, sbx *sandbox.Sandbox) { return } - err := telemetry.Observe0(ctx, tracer, "routing-delete", func(ctx context.Context) error { - return p.store.DeleteSandbox(ctx, sbx.Runtime.SandboxID, sbx.Runtime.ExecutionID) + outcome, err := telemetry.Observe1(ctx, tracer, "routing-delete", func(ctx context.Context) (catalog.DeleteOutcome, error) { + return p.store.DeleteSandboxIfOwner(ctx, sbx.Runtime.SandboxID, catalog.SandboxOwner{ + ExecutionID: sbx.Runtime.ExecutionID, + LifecycleID: sbx.LifecycleID, + OrchestratorID: p.orchestratorID, + }) }) if err != nil { p.deleteCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("result", "error"))) @@ -191,7 +196,11 @@ func (p *Publisher) OnStopping(ctx context.Context, sbx *sandbox.Sandbox) { return } - p.deleteCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("result", "ok"))) + result := string(outcome) + if outcome == catalog.DeleteOutcomeDeleted { + result = "ok" + } + p.deleteCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("result", result))) } // OnNetworkRelease is not used by the publisher. diff --git a/packages/orchestrator/pkg/routing/publisher_redis_test.go b/packages/orchestrator/pkg/routing/publisher_redis_test.go new file mode 100644 index 0000000000..9d8fd1dddf --- /dev/null +++ b/packages/orchestrator/pkg/routing/publisher_redis_test.go @@ -0,0 +1,50 @@ +package routing + +import ( + "testing" + + "github.com/stretchr/testify/require" + + redis_utils "github.com/e2b-dev/infra/packages/shared/pkg/redis" + catalog "github.com/e2b-dev/infra/packages/shared/pkg/sandbox-catalog" + "github.com/e2b-dev/infra/packages/shared/pkg/sandboxtypes" +) + +func TestPublisher_RedisPreservesNewLifecycleUntilItsOwnStop(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + store := catalog.NewRedisSandboxCatalog(client) + p := newPublisher(t, store) + ctx := t.Context() + oldSandbox := testSandbox(t, "sbx-1", "lc-old", sandboxtypes.SandboxTypeSandbox) + newSandbox := testSandbox(t, "sbx-1", "lc-new", sandboxtypes.SandboxTypeSandbox) + + // A resume-fresh checkpoint publishes a new VM under the same execution. + p.OnInsert(ctx, oldSandbox) + p.OnInsert(ctx, newSandbox) + got, err := store.GetSandbox(ctx, "sbx-1") + require.NoError(t, err) + require.Equal(t, "exec-1", got.ExecutionID) + require.Equal(t, "lc-new", got.LifecycleID) + require.Equal(t, "orch-1", got.OrchestratorID) + + key := "sandbox:routing:sbx-1" + value, err := client.Get(ctx, key).Result() + require.NoError(t, err) + expiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + + p.OnStopping(ctx, oldSandbox) + + retainedValue, err := client.Get(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, value, retainedValue, "the old lifecycle must preserve the replacement route") + retainedExpiry, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, expiresAt, retainedExpiry, "a stale stop must not refresh or shorten the replacement TTL") + + p.OnStopping(ctx, newSandbox) + _, err = store.GetSandbox(ctx, "sbx-1") + require.ErrorIs(t, err, catalog.ErrSandboxNotFound) +} diff --git a/packages/orchestrator/pkg/routing/publisher_test.go b/packages/orchestrator/pkg/routing/publisher_test.go index 6827de5739..38770acc6a 100644 --- a/packages/orchestrator/pkg/routing/publisher_test.go +++ b/packages/orchestrator/pkg/routing/publisher_test.go @@ -8,20 +8,25 @@ import ( "time" "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" noopmetric "go.opentelemetry.io/otel/metric/noop" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox" "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" catalog "github.com/e2b-dev/infra/packages/shared/pkg/sandbox-catalog" "github.com/e2b-dev/infra/packages/shared/pkg/sandboxtypes" + "github.com/e2b-dev/infra/packages/shared/pkg/telemetry" ) type memoryCatalog struct { - mu sync.Mutex - records map[string]*catalog.SandboxInfo - ttls map[string]time.Duration - storeErr error - deleteErr error + mu sync.Mutex + records map[string]*catalog.SandboxInfo + ttls map[string]time.Duration + storeErr error + storeAppliedErr error + deleteErr error // storeStarted is closed when StoreSandbox begins; storeRelease blocks it until closed. storeStarted chan struct{} @@ -63,22 +68,35 @@ func (c *memoryCatalog) StoreSandbox(_ context.Context, sandboxID string, info * c.records[sandboxID] = info c.ttls[sandboxID] = ttl - return nil + return c.storeAppliedErr } -func (c *memoryCatalog) DeleteSandbox(_ context.Context, sandboxID string, executionID string) error { +func (c *memoryCatalog) DeleteSandboxIfOwner(_ context.Context, sandboxID string, owner catalog.SandboxOwner) (catalog.DeleteOutcome, error) { + if owner.ExecutionID == "" || owner.LifecycleID == "" || owner.OrchestratorID == "" { + return "", catalog.ErrInvalidSandboxOwner + } + if c.deleteErr != nil { - return c.deleteErr + return "", c.deleteErr } c.mu.Lock() defer c.mu.Unlock() - if info, ok := c.records[sandboxID]; ok && info.ExecutionID == executionID { - delete(c.records, sandboxID) + info, ok := c.records[sandboxID] + if !ok { + return catalog.DeleteOutcomeAbsent, nil + } + if info.ExecutionID == "" || info.LifecycleID == "" || info.OrchestratorID == "" { + return catalog.DeleteOutcomeUnverifiable, nil } + if info.ExecutionID != owner.ExecutionID || info.LifecycleID != owner.LifecycleID || info.OrchestratorID != owner.OrchestratorID { + return catalog.DeleteOutcomeSuperseded, nil + } + + delete(c.records, sandboxID) - return nil + return catalog.DeleteOutcomeDeleted, nil } // has reports whether the record of the sandbox every test uses ("sbx-1") exists. @@ -142,6 +160,7 @@ func TestPublisher_OnInsertWritesRecord(t *testing.T) { require.Equal(t, "orch-1", got.OrchestratorID) require.Equal(t, "10.1.2.3", got.OrchestratorIP) require.Equal(t, "exec-1", got.ExecutionID) + require.Equal(t, "lc-1", got.LifecycleID) require.Equal(t, sbx.GetStartedAt(), got.StartedAt) require.EqualValues(t, 2, got.MaxLengthInHours) require.Equal(t, 2*time.Hour, c.ttls["sbx-1"]) @@ -161,6 +180,133 @@ func TestPublisher_OnStoppingDeletesRecord(t *testing.T) { require.ErrorIs(t, err, catalog.ErrSandboxNotFound) } +func TestPublisher_OnStoppingPreservesReplacementOwner(t *testing.T) { + t.Parallel() + + for _, test := range []struct { + name string + executionID string + lifecycleID string + orchestratorID string + }{ + { + name: "new execution", + executionID: "exec-2", + lifecycleID: "lc-1", + orchestratorID: "orch-1", + }, + { + name: "new lifecycle in same execution", + executionID: "exec-1", + lifecycleID: "lc-2", + orchestratorID: "orch-1", + }, + { + name: "new orchestrator process", + executionID: "exec-1", + lifecycleID: "lc-1", + orchestratorID: "orch-2", + }, + } { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + c := newMemoryCatalog() + p := newPublisher(t, c) + sbx := testSandbox(t, "sbx-1", "lc-1", sandboxtypes.SandboxTypeSandbox) + p.OnInsert(t.Context(), sbx) + + // A replacement route lands before this lifecycle's delayed stop. + replacement := &catalog.SandboxInfo{ + ExecutionID: test.executionID, + LifecycleID: test.lifecycleID, + OrchestratorID: test.orchestratorID, + OrchestratorIP: "10.2.3.4", + } + require.NoError(t, c.StoreSandbox(t.Context(), "sbx-1", replacement, time.Hour)) + + p.OnStopping(t.Context(), sbx) + + got, err := c.GetSandbox(t.Context(), "sbx-1") + require.NoError(t, err, "a stale stop must preserve the replacement route") + require.Equal(t, replacement, got) + }) + } +} + +func TestPublisher_OnStoppingReportsDeleteOutcome(t *testing.T) { + t.Parallel() + + for _, test := range []struct { + name string + replacement *catalog.SandboxInfo + absent bool + deleteErr error + wantResult string + }{ + {name: "deleted", wantResult: "ok"}, + {name: "absent", absent: true, wantResult: "absent"}, + { + name: "superseded", + replacement: &catalog.SandboxInfo{ + ExecutionID: "exec-1", LifecycleID: "lc-2", OrchestratorID: "orch-1", + }, + wantResult: "superseded", + }, + { + name: "unverifiable", + replacement: &catalog.SandboxInfo{ + ExecutionID: "exec-1", OrchestratorID: "orch-1", + }, + wantResult: "unverifiable", + }, + {name: "error", deleteErr: errors.New("redis down"), wantResult: "error"}, + } { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + t.Cleanup(func() { require.NoError(t, provider.Shutdown(context.WithoutCancel(t.Context()))) }) + c := newMemoryCatalog() + p, err := New(provider, c, "orch-1", "10.1.2.3") + require.NoError(t, err) + sbx := testSandbox(t, "sbx-1", "lc-1", sandboxtypes.SandboxTypeSandbox) + p.OnInsert(t.Context(), sbx) + + if test.absent { + delete(c.records, "sbx-1") + } + if test.replacement != nil { + require.NoError(t, c.StoreSandbox(t.Context(), "sbx-1", test.replacement, time.Hour)) + } + c.deleteErr = test.deleteErr + p.OnStopping(t.Context(), sbx) + + var metrics metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &metrics)) + for _, scope := range metrics.ScopeMetrics { + for _, metric := range scope.Metrics { + if metric.Name != string(telemetry.RoutingDeleteTotal) { + continue + } + + sum, ok := metric.Data.(metricdata.Sum[int64]) + require.True(t, ok) + require.Len(t, sum.DataPoints, 1) + require.EqualValues(t, 1, sum.DataPoints[0].Value) + result, ok := sum.DataPoints[0].Attributes.Value(attribute.Key("result")) + require.True(t, ok) + require.Equal(t, test.wantResult, result.AsString()) + + return + } + } + t.Fatal("routing delete outcome was not recorded") + }) + } +} + func TestPublisher_OnStoppingCollectsLifecycleState(t *testing.T) { t.Parallel() @@ -204,6 +350,24 @@ func TestPublisher_StoreErrorIsSwallowed(t *testing.T) { require.Empty(t, p.routes) } +func TestPublisher_StopAfterTimedOutStoreDeletesAppliedRecord(t *testing.T) { + t.Parallel() + + c := newMemoryCatalog() + c.storeAppliedErr = context.DeadlineExceeded + p := newPublisher(t, c) + sbx := testSandbox(t, "sbx-1", "lc-1", sandboxtypes.SandboxTypeSandbox) + + // Redis may apply the SET before the caller sees its timeout. + p.OnInsert(t.Context(), sbx) + _, err := c.GetSandbox(t.Context(), "sbx-1") + require.NoError(t, err) + + p.OnStopping(t.Context(), sbx) + _, err = c.GetSandbox(t.Context(), "sbx-1") + require.ErrorIs(t, err, catalog.ErrSandboxNotFound) +} + func TestPublisher_StopBeforeInsertLeavesTombstoneAndSkipsWrite(t *testing.T) { t.Parallel() diff --git a/packages/shared/pkg/sandbox-catalog/catalog.go b/packages/shared/pkg/sandbox-catalog/catalog.go index e4634c51ab..f881a72e6b 100644 --- a/packages/shared/pkg/sandbox-catalog/catalog.go +++ b/packages/shared/pkg/sandbox-catalog/catalog.go @@ -26,6 +26,23 @@ type SandboxInfo struct { NetworkPlacement *NetworkPlacement `json:"network_placement,omitempty"` } +// SandboxOwner identifies one VM lifecycle on one orchestrator process. +// Matching a routing record does not prove that the corresponding VM has stopped. +type SandboxOwner struct { + ExecutionID string + LifecycleID string + OrchestratorID string +} + +type DeleteOutcome string + +const ( + DeleteOutcomeAbsent DeleteOutcome = "absent" + DeleteOutcomeDeleted DeleteOutcome = "deleted" + DeleteOutcomeSuperseded DeleteOutcome = "superseded" + DeleteOutcomeUnverifiable DeleteOutcome = "unverifiable" +) + type SandboxesCatalog interface { GetSandbox(ctx context.Context, sandboxID string) (*SandboxInfo, error) StoreSandbox(ctx context.Context, sandboxID string, sandboxInfo *SandboxInfo, expiration time.Duration) error @@ -36,3 +53,5 @@ type SandboxesCatalog interface { var tracer = otel.Tracer("github.com/e2b-dev/infra/packages/shared/pkg/sandbox-catalog") var ErrSandboxNotFound = errors.New("sandbox not found") + +var ErrInvalidSandboxOwner = errors.New("sandbox owner requires execution, lifecycle, and orchestrator IDs") diff --git a/packages/shared/pkg/sandbox-catalog/catalog_owner_test.go b/packages/shared/pkg/sandbox-catalog/catalog_owner_test.go new file mode 100644 index 0000000000..66ddabfd84 --- /dev/null +++ b/packages/shared/pkg/sandbox-catalog/catalog_owner_test.go @@ -0,0 +1,260 @@ +package sandbox_catalog + +import ( + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/require" + + redis_utils "github.com/e2b-dev/infra/packages/shared/pkg/redis" +) + +func TestDeleteSandboxIfOwnerMatchesAndIsIdempotent(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + catalog := NewRedisSandboxCatalog(client) + ctx := t.Context() + id := "sbx-owner-delete" + owner := SandboxOwner{ExecutionID: "exec-1", LifecycleID: "life-1", OrchestratorID: "orch-A"} + + outcome, err := catalog.DeleteSandboxIfOwner(ctx, id, owner) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeAbsent, outcome) + + info := testSandboxInfo(owner.ExecutionID, owner.OrchestratorID) + info.LifecycleID = owner.LifecycleID + require.NoError(t, catalog.StoreSandbox(ctx, id, info, time.Minute)) + + outcome, err = catalog.DeleteSandboxIfOwner(ctx, id, owner) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeDeleted, outcome) + _, err = catalog.GetSandbox(ctx, id) + require.ErrorIs(t, err, ErrSandboxNotFound) + + outcome, err = catalog.DeleteSandboxIfOwner(ctx, id, owner) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeAbsent, outcome) +} + +func TestDeleteSandboxIfOwnerKeepsDifferentOwner(t *testing.T) { + t.Parallel() + + for name, owner := range map[string]SandboxOwner{ + "different execution": {ExecutionID: "exec-old", LifecycleID: "life-1", OrchestratorID: "orch-A"}, + "different lifecycle": {ExecutionID: "exec-1", LifecycleID: "life-old", OrchestratorID: "orch-A"}, + "different orchestrator": {ExecutionID: "exec-1", LifecycleID: "life-1", OrchestratorID: "orch-old"}, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + catalog := NewRedisSandboxCatalog(client) + ctx := t.Context() + id := "sbx-owner-mismatch" + key := catalog.getCatalogKey(id) + value := `{"execution_id":"exec-1","lifecycle_id":"life-1","orchestrator_id":"orch-A","unknown_field":"preserve me"}` + require.NoError(t, client.Set(ctx, key, value, time.Minute).Err()) + expiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + + outcome, err := catalog.DeleteSandboxIfOwner(ctx, id, owner) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeSuperseded, outcome) + + got, err := client.Get(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, value, got) + gotExpiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, expiresAt, gotExpiresAt) + }) + } +} + +func TestDeleteSandboxIfOwnerKeepsReplacementLifecycle(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + catalog := NewRedisSandboxCatalog(client) + ctx := t.Context() + id := "sbx-same-execution-aba" + key := catalog.getCatalogKey(id) + oldOwner := SandboxOwner{ExecutionID: "exec-1", LifecycleID: "life-1", OrchestratorID: "orch-A"} + oldInfo := testSandboxInfo(oldOwner.ExecutionID, oldOwner.OrchestratorID) + oldInfo.LifecycleID = oldOwner.LifecycleID + require.NoError(t, catalog.StoreSandbox(ctx, id, oldInfo, time.Minute)) + + // A resume-fresh checkpoint replaces the VM while retaining execution and node. + newOwner := SandboxOwner{ExecutionID: "exec-1", LifecycleID: "life-2", OrchestratorID: "orch-A"} + newInfo := testSandboxInfo(newOwner.ExecutionID, newOwner.OrchestratorID) + newInfo.LifecycleID = newOwner.LifecycleID + require.NoError(t, catalog.StoreSandbox(ctx, id, newInfo, 2*time.Minute)) + value, err := client.Get(ctx, key).Result() + require.NoError(t, err) + expiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + + // Delayed cleanup from the original VM must not remove the replacement. + outcome, err := catalog.DeleteSandboxIfOwner(ctx, id, oldOwner) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeSuperseded, outcome) + got, err := client.Get(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, value, got) + gotExpiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, expiresAt, gotExpiresAt) + + outcome, err = catalog.DeleteSandboxIfOwner(ctx, id, newOwner) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeDeleted, outcome) + _, err = catalog.GetSandbox(ctx, id) + require.ErrorIs(t, err, ErrSandboxNotFound) +} + +func TestDeleteSandboxIfOwnerKeepsUnverifiableRecord(t *testing.T) { + t.Parallel() + + cases := map[string]string{ + "legacy record": `{"execution_id":"exec-1","orchestrator_id":"orch-A"}`, + "malformed JSON": `{"execution_id":"exec-1"`, + "JSON null": `null`, + "JSON array": `[]`, + "JSON string": `"exec-1"`, + "JSON number": `1`, + "JSON boolean": `true`, + } + for _, field := range []string{"execution_id", "lifecycle_id", "orchestrator_id"} { + for name, invalidValue := range map[string]any{ + "missing": nil, + "empty": "", + "null": nil, + "number": 1, + "boolean": true, + "array": []string{"value"}, + "object": map[string]string{"id": "value"}, + } { + info := map[string]any{"execution_id": "exec-1", "lifecycle_id": "life-1", "orchestrator_id": "orch-A"} + if name == "missing" { + delete(info, field) + } else { + info[field] = invalidValue + } + value, err := json.Marshal(info) + require.NoError(t, err) + cases[field+" "+name] = string(value) + } + } + + for name, value := range cases { + t.Run(name, func(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + catalog := NewRedisSandboxCatalog(client) + ctx := t.Context() + id := "sbx-unverifiable-owner" + key := catalog.getCatalogKey(id) + require.NoError(t, client.Set(ctx, key, value, time.Minute).Err()) + expiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + + outcome, err := catalog.DeleteSandboxIfOwner(ctx, id, SandboxOwner{ + ExecutionID: "exec-1", LifecycleID: "life-1", OrchestratorID: "orch-A", + }) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeUnverifiable, outcome) + + got, err := client.Get(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, value, got) + gotExpiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, expiresAt, gotExpiresAt) + }) + } +} + +func TestDeleteSandboxIfOwnerRejectsIncompleteExpectedOwner(t *testing.T) { + t.Parallel() + + for name, owner := range map[string]SandboxOwner{ + "missing execution": {LifecycleID: "life-1", OrchestratorID: "orch-A"}, + "missing lifecycle": {ExecutionID: "exec-1", OrchestratorID: "orch-A"}, + "missing orchestrator": {ExecutionID: "exec-1", LifecycleID: "life-1"}, + "zero owner": {}, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + catalog := NewRedisSandboxCatalog(client) + ctx := t.Context() + id := "sbx-incomplete-expected-owner" + key := catalog.getCatalogKey(id) + value := `{"execution_id":"exec-1","lifecycle_id":"life-1","orchestrator_id":"orch-A"}` + require.NoError(t, client.Set(ctx, key, value, time.Minute).Err()) + expiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + + outcome, err := catalog.DeleteSandboxIfOwner(ctx, id, owner) + require.ErrorIs(t, err, ErrInvalidSandboxOwner) + require.Empty(t, outcome) + + got, err := client.Get(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, value, got) + gotExpiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, expiresAt, gotExpiresAt) + + outcome, err = catalog.DeleteSandboxIfOwner(ctx, "sbx-absent", owner) + require.ErrorIs(t, err, ErrInvalidSandboxOwner) + require.Empty(t, outcome) + }) + } +} + +func TestDeleteSandboxIfOwnerKeepsLegacyRecordReadableAndLegacyDeletable(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + catalog := NewRedisSandboxCatalog(client) + ctx := t.Context() + id := "sbx-legacy-compatibility" + key := catalog.getCatalogKey(id) + value := `{"orchestrator_id":"orch-A","orchestrator_ip":"10.0.0.1","execution_id":"exec-1"}` + require.NoError(t, client.Set(ctx, key, value, time.Minute).Err()) + + outcome, err := catalog.DeleteSandboxIfOwner(ctx, id, SandboxOwner{ + ExecutionID: "exec-1", LifecycleID: "life-1", OrchestratorID: "orch-A", + }) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeUnverifiable, outcome) + info, err := catalog.GetSandbox(ctx, id) + require.NoError(t, err) + require.Equal(t, "exec-1", info.ExecutionID) + require.Equal(t, "orch-A", info.OrchestratorID) + require.Empty(t, info.LifecycleID) + require.Nil(t, info.NetworkPlacement) + + require.NoError(t, catalog.DeleteSandbox(ctx, id, "exec-1")) + _, err = catalog.GetSandbox(ctx, id) + require.ErrorIs(t, err, ErrSandboxNotFound) +} + +func TestDeleteSandboxIfOwnerReturnsRedisError(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + require.NoError(t, client.Close()) + catalog := NewRedisSandboxCatalog(client) + + outcome, err := catalog.DeleteSandboxIfOwner(t.Context(), "sbx-closed", SandboxOwner{ + ExecutionID: "exec-1", LifecycleID: "life-1", OrchestratorID: "orch-A", + }) + require.Error(t, err) + require.Empty(t, outcome) +} diff --git a/packages/shared/pkg/sandbox-catalog/catalog_redis.go b/packages/shared/pkg/sandbox-catalog/catalog_redis.go index eb3f31ec22..58f344c5f1 100644 --- a/packages/shared/pkg/sandbox-catalog/catalog_redis.go +++ b/packages/shared/pkg/sandbox-catalog/catalog_redis.go @@ -17,26 +17,28 @@ import ( const ( catalogRedisTimeout = time.Second * 1 - // deleteIfSameExecution outcomes, distinguished for diagnostics. - catalogDeleteAbsent = 0 // no entry - catalogDeleteDeleted = 1 // entry deleted (execution matched) - catalogDeleteMismatch = 2 // entry kept — a different execution owns it now - catalogDeleteUnreadable = 3 // entry kept — value undecodable or missing execution_id + // Catalog delete script outcomes, distinguished for diagnostics. + catalogDeleteAbsent = 0 // no entry + catalogDeleteDeleted = 1 // entry deleted (expected identity matched) + catalogDeleteMismatch = 2 // entry kept — a different identity owns it now + catalogDeleteUnreadable = 3 // entry kept — undecodable value or unverifiable identity + catalogDeleteOwnerRequired = 4 // legacy deletion cannot verify a lifecycle-aware entry ) -// deleteIfSameExecution deletes the catalog entry only if its stored execution_id -// still matches the one passed in, as one atomic server-side script — so a stale -// teardown can't remove an entry a concurrent StoreSandbox wrote for a newer execution. -// Returns the catalogDelete* outcome. +// deleteIfSameExecution is the legacy execution-only delete. Records carrying +// any lifecycle_id field require the full owner comparison instead. var deleteIfSameExecution = redis.NewScript(` local v = redis.call('GET', KEYS[1]) if not v then return 0 end local ok, info = pcall(cjson.decode, v) -if not (ok and type(info) == 'table' and type(info.execution_id) == 'string') then +if not (ok and type(info) == 'table' and type(info.execution_id) == 'string' and info.execution_id ~= '' and ARGV[1] ~= '') then return 3 end +if info.lifecycle_id ~= nil then + return 4 +end if info.execution_id == ARGV[1] then redis.call('DEL', KEYS[1]) return 1 @@ -44,6 +46,29 @@ end return 2 `) +// deleteIfSameOwner compares all ownership dimensions and deletes in one Redis +// operation. Refused deletes leave the record and its expiration untouched. +var deleteIfSameOwner = redis.NewScript(` +local v = redis.call('GET', KEYS[1]) +if not v then + return 0 +end +local ok, info = pcall(cjson.decode, v) +if not (ok and type(info) == 'table') then + return 3 +end +for _, field in ipairs({'execution_id', 'lifecycle_id', 'orchestrator_id'}) do + if type(info[field]) ~= 'string' or info[field] == '' then + return 3 + end +end +if info.execution_id ~= ARGV[1] or info.lifecycle_id ~= ARGV[2] or info.orchestrator_id ~= ARGV[3] then + return 2 +end +redis.call('DEL', KEYS[1]) +return 1 +`) + // routingKeyPrefix is the orchestrator-owned routing record. const routingKeyPrefix = "sandbox:routing:" @@ -107,8 +132,9 @@ func (c *RedisSandboxCatalog) StoreSandbox(ctx context.Context, sandboxID string return nil } -// DeleteSandbox deletes the entry if its execution ID matches and returns the Redis error, if any. -// A mismatch, an unreadable value or an absent key are not errors. +// DeleteSandbox deletes a legacy entry with a matching nonempty execution ID. +// Lifecycle-aware entries must use DeleteSandboxIfOwner. A mismatch, an +// unverifiable value or an absent key are not errors. func (c *RedisSandboxCatalog) DeleteSandbox(ctx context.Context, sandboxID string, executionID string) error { spanCtx, span := tracer.Start(ctx, "sandbox-catalog-delete") defer span.End() @@ -136,6 +162,9 @@ func (c *RedisSandboxCatalog) DeleteSandbox(ctx context.Context, sandboxID strin logger.L().Warn(ctx, "sandbox catalog entry is unreadable; leaving it to expire via TTL", logger.WithSandboxID(sandboxID)) case catalogDeleteAbsent: span.SetAttributes(attribute.String("delete.outcome", "absent")) + case catalogDeleteOwnerRequired: + span.SetAttributes(attribute.String("delete.outcome", "owner-required")) + logger.L().Debug(ctx, "kept lifecycle-aware sandbox catalog entry requiring full owner comparison", logger.WithSandboxID(sandboxID)) default: span.SetAttributes(attribute.String("delete.outcome", "unexpected")) logger.L().Warn(ctx, "unexpected sandbox catalog delete outcome", logger.WithSandboxID(sandboxID), zap.Int("outcome", outcome)) @@ -144,6 +173,49 @@ func (c *RedisSandboxCatalog) DeleteSandbox(ctx context.Context, sandboxID strin return nil } +// DeleteSandboxIfOwner removes only the exact execution/lifecycle/orchestrator +// owner. Legacy records lacking any owner field are unverifiable, not wildcards. +func (c *RedisSandboxCatalog) DeleteSandboxIfOwner(ctx context.Context, sandboxID string, owner SandboxOwner) (DeleteOutcome, error) { + if owner.ExecutionID == "" || owner.LifecycleID == "" || owner.OrchestratorID == "" { + return "", ErrInvalidSandboxOwner + } + + spanCtx, span := tracer.Start(ctx, "sandbox-catalog-delete-if-owner") + defer span.End() + + ctx, ctxCancel := context.WithTimeout(spanCtx, catalogRedisTimeout) + defer ctxCancel() + + result, err := deleteIfSameOwner.Run(ctx, c.redisClient, []string{c.getCatalogKey(sandboxID)}, + owner.ExecutionID, owner.LifecycleID, owner.OrchestratorID).Int() + if err != nil { + span.RecordError(err) + + return "", fmt.Errorf("failed to delete sandbox info from redis: %w", err) + } + + var outcome DeleteOutcome + switch result { + case catalogDeleteAbsent: + outcome = DeleteOutcomeAbsent + case catalogDeleteDeleted: + outcome = DeleteOutcomeDeleted + case catalogDeleteMismatch: + outcome = DeleteOutcomeSuperseded + case catalogDeleteUnreadable: + outcome = DeleteOutcomeUnverifiable + default: + err := fmt.Errorf("unexpected sandbox catalog owner delete outcome: %d", result) + span.RecordError(err) + + return "", err + } + + span.SetAttributes(attribute.String("delete.outcome", string(outcome))) + + return outcome, nil +} + func (c *RedisSandboxCatalog) getCatalogKey(sandboxID string) string { return routingKeyPrefix + sandboxID } diff --git a/packages/shared/pkg/sandbox-catalog/catalog_redis_test.go b/packages/shared/pkg/sandbox-catalog/catalog_redis_test.go index 8a18a7d27b..077240ac34 100644 --- a/packages/shared/pkg/sandbox-catalog/catalog_redis_test.go +++ b/packages/shared/pkg/sandbox-catalog/catalog_redis_test.go @@ -64,7 +64,13 @@ func TestRedisSandboxCatalog(t *testing.T) { got, err := catalog.GetSandbox(ctx, id) require.NoError(t, err) require.Equal(t, want, got) - require.NoError(t, catalog.DeleteSandbox(ctx, id, "exec-1")) + outcome, err := catalog.DeleteSandboxIfOwner(ctx, id, SandboxOwner{ + ExecutionID: want.ExecutionID, LifecycleID: want.LifecycleID, OrchestratorID: want.OrchestratorID, + }) + require.NoError(t, err) + require.Equal(t, DeleteOutcomeDeleted, outcome) + _, err = catalog.GetSandbox(ctx, id) + require.ErrorIs(t, err, ErrSandboxNotFound) }) t.Run("get on an absent key returns ErrSandboxNotFound", func(t *testing.T) { @@ -124,6 +130,82 @@ func TestDeleteSandboxReturnsRedisError(t *testing.T) { require.Error(t, broken.DeleteSandbox(ctx, "sbx-closed", "exec-1")) } +func TestDeleteSandboxKeepsLifecycleRecord(t *testing.T) { + t.Parallel() + + for name, value := range map[string]string{ + "nonempty lifecycle": `{"execution_id":"exec-1","lifecycle_id":"lifecycle-2","orchestrator_id":"orch-B"}`, + "empty lifecycle": `{"execution_id":"exec-1","lifecycle_id":"","orchestrator_id":"orch-B"}`, + "null lifecycle": `{"execution_id":"exec-1","lifecycle_id":null,"orchestrator_id":"orch-B"}`, + "numeric lifecycle": `{"execution_id":"exec-1","lifecycle_id":2,"orchestrator_id":"orch-B"}`, + "object lifecycle": `{"execution_id":"exec-1","lifecycle_id":{},"orchestrator_id":"orch-B"}`, + "array lifecycle": `{"execution_id":"exec-1","lifecycle_id":[],"orchestrator_id":"orch-B"}`, + "boolean lifecycle": `{"execution_id":"exec-1","lifecycle_id":false,"orchestrator_id":"orch-B"}`, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + catalog := NewRedisSandboxCatalog(client) + ctx := t.Context() + id := "sbx-legacy-delete-lifecycle" + key := catalog.getCatalogKey(id) + require.NoError(t, client.Set(ctx, key, value, time.Minute).Err()) + expiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + + // Execution IDs survive a resume-fresh checkpoint. An old caller cannot + // prove it owns the new Firecracker lifecycle with that ID alone. + require.NoError(t, catalog.DeleteSandbox(ctx, id, "exec-1")) + + got, err := client.Get(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, value, got) + gotExpiresAt, err := client.PExpireTime(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, expiresAt, gotExpiresAt) + }) + } +} + +func TestDeleteSandboxKeepsUnverifiableExecution(t *testing.T) { + t.Parallel() + + for name, tc := range map[string]struct { + value string + executionID string + }{ + "malformed JSON": {value: `{"execution_id":"exec-1"`, executionID: "exec-1"}, + "missing execution": {value: `{"orchestrator_id":"orch-A"}`, executionID: "exec-1"}, + "null execution": {value: `{"execution_id":null}`, executionID: "exec-1"}, + "numeric execution": {value: `{"execution_id":1}`, executionID: "exec-1"}, + "empty execution": {value: `{"execution_id":""}`, executionID: ""}, + "empty expected": {value: `{"execution_id":"exec-1"}`, executionID: ""}, + "JSON null": {value: `null`, executionID: "exec-1"}, + "JSON array": {value: `[]`, executionID: "exec-1"}, + "JSON string": {value: `"exec-1"`, executionID: "exec-1"}, + "JSON number": {value: `1`, executionID: "exec-1"}, + "JSON boolean": {value: `true`, executionID: "exec-1"}, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + + client := redis_utils.SetupInstance(t) + catalog := NewRedisSandboxCatalog(client) + ctx := t.Context() + id := "sbx-legacy-delete-unverifiable" + key := catalog.getCatalogKey(id) + require.NoError(t, client.Set(ctx, key, tc.value, time.Minute).Err()) + + require.NoError(t, catalog.DeleteSandbox(ctx, id, tc.executionID)) + + got, err := client.Get(ctx, key).Result() + require.NoError(t, err) + require.Equal(t, tc.value, got) + }) + } +} + func TestDeleteIfSameExecutionOutcomes(t *testing.T) { t.Parallel() diff --git a/packages/shared/pkg/telemetry/meters.go b/packages/shared/pkg/telemetry/meters.go index f710e513f1..627c945d04 100644 --- a/packages/shared/pkg/telemetry/meters.go +++ b/packages/shared/pkg/telemetry/meters.go @@ -853,7 +853,7 @@ var counterDesc = map[CounterType]string{ CmuxErrorsTotal: "Total number of cmux connection multiplexer errors", RoutingPublishTotal: "Total number of sandbox routing record writes on MarkRunning (result=ok/error)", - RoutingDeleteTotal: "Total number of sandbox routing record deletes on MarkStopping (result=ok/error)", + RoutingDeleteTotal: "Total number of sandbox routing record cleanup attempts on MarkStopping (result=ok/absent/superseded/unverifiable/error; ok means deleted)", SandboxFCNetFails: "Total Firecracker VMM errors transmitting or receiving data (direction=tx/rx)", SandboxFCNetNoAvailBuffer: "Total Firecracker VMM events where no virtqueue buffer was available (direction=tx/rx)",