Detect CRASHED Actors during Checkpoint and Restore (#353)

Related to #292 and #119

Changes:

1. ResumeActor:  
1. During `ResumeActor` we'll put the actor in CRASHED state if the
Resume failed due to any errors related to the snapshot itself, which
includes:
      1. Missing or corrupted snapshot when downloading from GCS  
      2. Runsc command fails.  
2. All other types of errors will not put the Actor in CRASHED state.
2. SuspendActor or PauseActor  
1. During `SuspendActor` or `PauseActor` we'll put the actor in CRASHED
state if we cannot produce a valid snapshot, which includes:
      1. cannot connect to Ateom after retries,  
      2. runsc command fails.  
3. cannot upload the snapshot to GCS (during suspend) ; or cannot save
snapshot locally (during pause).
   2. All other types of errors will not put the Actor in CRASHED state.

Author: zoez7

cmd/ateapi/internal/controlapi/crash.go

@@ -0,0 +1,67 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlapi
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/internal/ateerrors"
+	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// maybeCrashActor inspects err returned by an atelet RPC, it crashes
+// the actor if the err carries the actorCrashed=true metadata directive.
+func maybeCrashActor(ctx context.Context, st store.Interface, atespace, actorID string, err error, wrapMsg string) error {
+	if err == nil {
+		return nil
+	}
+
+	if ateerrors.ActorCrashRequested(err) {
+		slog.ErrorContext(ctx, "Setting Actor to crashed due to error", slog.Any("error", err))
+		if cerr := crashActor(ctx, st, atespace, actorID); cerr != nil {
+			slog.ErrorContext(ctx, "Failed to crash actor", slog.Any("cerr", cerr))
+			return cerr
+		}
+		return status.Errorf(codes.DataLoss, "actor %s crashed", actorID)
+	}
+	return fmt.Errorf("%s: %w", wrapMsg, err)
+}
+
+// crashActor moves the actor to CRASHED state.
+func crashActor(ctx context.Context, st store.Interface, atespace, actorID string) error {
+	actor, err := st.GetActor(ctx, atespace, actorID)
+	if err != nil {
+		return fmt.Errorf("while loading actor to crash: %w", err)
+	}
+	actor.Status = ateapipb.Actor_STATUS_CRASHED
+	// TODO(zoezhao):
+	// 1. If the Actor crashed because the worker is unhealthy,
+	//    free the worker and mark it as unhealthy(or delete it)
+	//    to prevent other actors from being scheduled on it.
+	// 2. If the Actor crashed while resuming from a Paused state,
+	//    we must preserve the Actor's assigned node VM in order
+	//    to support `ate actor dump` command.
+	// (https://github.com/agent-substrate/substrate/issues/119)
+	if err := st.UpdateActor(ctx, actor, actor.GetVersion()); err != nil {
+		return fmt.Errorf("while marking actor crashed: %w", err)
+	}
+
+	return nil
+}

cmd/ateapi/internal/controlapi/crash_test.go

@@ -0,0 +1,240 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlapi
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
+	"github.com/agent-substrate/substrate/internal/ateerrors"
+	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// seedActor stores a running actor with all worker-binding fields populated, so
+// tests can assert they are cleared when the actor crashes.
+func seedActor(t *testing.T, ctx context.Context, st store.Interface, atespace, id string) {
+	t.Helper()
+	if err := st.CreateActor(ctx, &ateapipb.Actor{
+		ActorId:           id,
+		Atespace:          atespace,
+		Status:            ateapipb.Actor_STATUS_RUNNING,
+		AteomPodNamespace: "ns",
+		AteomPodName:      "pod",
+		AteomPodIp:        "1.2.3.4",
+		AteomPodUid:       "uid",
+		WorkerPoolName:    "pool",
+	}); err != nil {
+		t.Fatalf("seed actor: %v", err)
+	}
+}
+
+// assertCrashed reloads the actor and verifies it is CRASHED.
+func assertCrashed(t *testing.T, ctx context.Context, st store.Interface, atespace, id string) {
+	t.Helper()
+	got, err := st.GetActor(ctx, atespace, id)
+	if err != nil {
+		t.Fatalf("GetActor(%q, %q) = %v, want nil", atespace, id, err)
+	}
+	if got.GetStatus() != ateapipb.Actor_STATUS_CRASHED {
+		t.Errorf("status = %v, want %v", got.GetStatus(), ateapipb.Actor_STATUS_CRASHED)
+	}
+}
+
+func TestCrashActor(t *testing.T) {
+	const (
+		atespace = "team-a"
+		actorID  = "actor-1"
+	)
+
+	tests := []struct {
+		name string
+		seed bool
+		// check inspects the returned error; nil-safe.
+		check func(t *testing.T, ctx context.Context, st store.Interface, err error)
+	}{
+		{
+			name: "crashes running actor",
+			seed: true,
+			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
+				if err != nil {
+					t.Fatalf("crashActor() = %v, want nil", err)
+				}
+				assertCrashed(t, ctx, st, atespace, actorID)
+			},
+		},
+		{
+			name: "actor not found",
+			seed: false,
+			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
+				if err == nil {
+					t.Fatal("crashActor() = nil, want error")
+				}
+				if !errors.Is(err, store.ErrNotFound) {
+					t.Errorf("crashActor() error = %v, want errors.Is(store.ErrNotFound)", err)
+				}
+				if !strings.Contains(err.Error(), "while loading actor to crash") {
+					t.Errorf("crashActor() error = %q, want it to contain %q", err, "while loading actor to crash")
+				}
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+			st, cleanup := storetest.SetupTestStore(t)
+			defer cleanup()
+
+			if tt.seed {
+				seedActor(t, ctx, st, atespace, actorID)
+			}
+
+			err := crashActor(ctx, st, atespace, actorID)
+			tt.check(t, ctx, st, err)
+		})
+	}
+}
+
+func TestMaybeCrashActor(t *testing.T) {
+	const (
+		atespace = "team-a"
+		actorID  = "actor-1"
+		wrapMsg  = "calling atelet"
+	)
+
+	crashErr := ateerrors.NewGRPCError(context.Background(), codes.NotFound, ateerrors.ReasonTerminalFileSystemError, ateerrors.ActorCrashedMetadata(), errors.New("boom"))
+	// A structured error carrying a reason but no actorCrashed directive must be
+	// wrapped, not crash the actor.
+	noCrashErr := ateerrors.NewGRPCError(context.Background(), codes.NotFound, ateerrors.ReasonFailedGetExternalObject, nil, errors.New("infra"))
+	plainErr := errors.New("transient")
+
+	tests := []struct {
+		name string
+		seed bool
+		err  error
+		// check inspects the returned error and store state.
+		check func(t *testing.T, ctx context.Context, st store.Interface, err error)
+	}{
+		{
+			name: "nil error returns nil",
+			seed: false,
+			err:  nil,
+			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
+				if err != nil {
+					t.Fatalf("maybeCrashActor() = %v, want nil", err)
+				}
+			},
+		},
+		{
+			name: "crash reason crashes actor",
+			seed: true,
+			err:  crashErr,
+			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
+				if err == nil {
+					t.Fatal("maybeCrashActor() = nil, want error")
+				}
+				if got := status.Code(err); got != codes.DataLoss {
+					t.Errorf("status code = %v, want %v", got, codes.DataLoss)
+				}
+				assertCrashed(t, ctx, st, atespace, actorID)
+			},
+		},
+		{
+			name: "crash reason but actor missing returns load error",
+			seed: false,
+			err:  crashErr,
+			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
+				if err == nil {
+					t.Fatal("maybeCrashActor() = nil, want error")
+				}
+				if got := status.Code(err); got == codes.DataLoss {
+					t.Errorf("status code = %v, want it not to be DataLoss", got)
+				}
+				if !errors.Is(err, store.ErrNotFound) {
+					t.Errorf("maybeCrashActor() error = %v, want errors.Is(store.ErrNotFound)", err)
+				}
+			},
+		},
+		{
+			name: "status error without crash directive is wrapped",
+			seed: true,
+			err:  noCrashErr,
+			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
+				if err == nil {
+					t.Fatal("maybeCrashActor() = nil, want error")
+				}
+				if !errors.Is(err, noCrashErr) {
+					t.Errorf("maybeCrashActor() error = %v, want errors.Is(noCrashErr)", err)
+				}
+				if !strings.HasPrefix(err.Error(), wrapMsg) {
+					t.Errorf("maybeCrashActor() error = %q, want prefix %q", err, wrapMsg)
+				}
+				// The actor must not have been crashed.
+				got, gerr := st.GetActor(ctx, atespace, actorID)
+				if gerr != nil {
+					t.Fatalf("GetActor() = %v, want nil", gerr)
+				}
+				if got.GetStatus() == ateapipb.Actor_STATUS_CRASHED {
+					t.Errorf("status = CRASHED, want it unchanged")
+				}
+			},
+		},
+		{
+			name: "non-crash error is wrapped",
+			seed: true,
+			err:  plainErr,
+			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
+				if err == nil {
+					t.Fatal("maybeCrashActor() = nil, want error")
+				}
+				if !errors.Is(err, plainErr) {
+					t.Errorf("maybeCrashActor() error = %v, want errors.Is(plainErr)", err)
+				}
+				if !strings.HasPrefix(err.Error(), wrapMsg) {
+					t.Errorf("maybeCrashActor() error = %q, want prefix %q", err, wrapMsg)
+				}
+				// The actor must not have been crashed.
+				got, gerr := st.GetActor(ctx, atespace, actorID)
+				if gerr != nil {
+					t.Fatalf("GetActor() = %v, want nil", gerr)
+				}
+				if got.GetStatus() == ateapipb.Actor_STATUS_CRASHED {
+					t.Errorf("status = CRASHED, want it unchanged")
+				}
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+			st, cleanup := storetest.SetupTestStore(t)
+			defer cleanup()
+
+			if tt.seed {
+				seedActor(t, ctx, st, atespace, actorID)
+			}
+
+			err := maybeCrashActor(ctx, st, atespace, actorID, tt.err, wrapMsg)
+			tt.check(t, ctx, st, err)
+		})
+	}
+}

cmd/ateapi/internal/controlapi/functional_test.go

@@ -237,6 +237,7 @@ type testContext struct {
 	k8sClient           kubernetes.Interface
 	substrateClient     versioned.Interface
 	persistence         *ateredis.Persistence
+	workerCache         *workercache.Cache
 	fakeAtelet          *FakeAteletServer
 	cleanup             func()
 	actorTemplateLister listersv1alpha1.ActorTemplateLister
@@ -369,6 +370,7 @@ func setupTest(t *testing.T, ns string) *testContext {
 		k8sClient:           k8sClient,
 		substrateClient:     substrateClient,
 		persistence:         persistence,
+		workerCache:         wc,
 		fakeAtelet:          fakeAtelet,
 		cleanup:             cleanup,
 		actorTemplateLister: actorTemplateLister,
@@ -617,6 +619,23 @@ func createWorkerPod(t *testing.T, tc *testContext, ns string, name string, node
 	if err != nil {
 		t.Fatalf("failed to wait for worker to be registered: %v", err)
 	}
+
+	// Wait for the worker to appear in worker cache.
+	err = wait.PollUntilContextTimeout(context.Background(), 10*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) {
+		workers, err := tc.workerCache.Workers()
+		if err != nil {
+			return false, nil // Cache not ready yet; retry.
+		}
+		for _, w := range workers {
+			if w.GetWorkerNamespace() == ns && w.GetWorkerPod() == name {
+				return true, nil
+			}
+		}
+		return false, nil
+	})
+	if err != nil {
+		t.Fatalf("failed to wait for worker to appear in worker cache: %v", err)
+	}
 }
 
 func deleteWorkerPod(t *testing.T, tc *testContext, ns string, name string) {
@@ -642,6 +661,22 @@ func deleteWorkerPod(t *testing.T, tc *testContext, ns string, name string) {
 	if err != nil {
 		t.Fatalf("failed to wait for worker to be removed: %v", err)
 	}
+
+	err = wait.PollUntilContextTimeout(context.Background(), 10*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) {
+		workers, err := tc.workerCache.Workers()
+		if err != nil {
+			return false, nil // Cache not ready yet; retry.
+		}
+		for _, w := range workers {
+			if w.GetWorkerNamespace() == ns && w.GetWorkerPod() == name {
+				return false, nil // Still there
+			}
+		}
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("failed to wait for worker to be removed from worker cache: %v", err)
+	}
 }
 
 // TestCreateActor_Success tests the happy path for creating an actor.

cmd/ateapi/internal/controlapi/syncer.go

@@ -193,7 +193,9 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
 	if actor.GetAteomPodNamespace() != namespace || actor.GetAteomPodName() != podName {
 		return nil
 	}
-	actor.Status = ateapipb.Actor_STATUS_SUSPENDED
+	if actor.Status != ateapipb.Actor_STATUS_CRASHED {
+		actor.Status = ateapipb.Actor_STATUS_SUSPENDED
+	}
 	actor.AteomPodNamespace = ""
 	actor.AteomPodName = ""
 	actor.AteomPodIp = ""