fix(orchestrator): guard route cleanup by lifecycle owner

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

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.

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)
+}

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()