Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions internal/controller/worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,12 @@ func (w *Worker) processEvent(ctx context.Context, ev store.TaskEvent) error {
}
return fmt.Errorf("fetching task %s/%s: %w", ev.Atespace, ev.Name, err)
}
// A pending delete event owns this task now; reconciling would resume an actor
// that is about to be torn down and overwrite the Terminating phase.
if task.GetStatus().GetPhase() == v1alpha1.PhaseTerminating {
slog.Info("task is terminating, skipping reconcile", "atespace", ev.Atespace, "name", ev.Name)
return nil
}

// Resolve every bound workspace. A missing one is skipped so the task still
// runs; the runner creates an empty directory at its path.
Expand Down
58 changes: 58 additions & 0 deletions internal/controller/worker_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -173,3 +173,61 @@ func TestWorkerDeletion(t *testing.T) {
t.Errorf("expected template deleted, got %v", mockSrv.deletedTemplates)
}
}

func TestWorkerSkipsReconcileOfTerminatingTask(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

lis, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("failed to listen: %v", err)
}
defer lis.Close()

mockSrv := &mockControlServer{}
grpcServer := grpc.NewServer()
ateapipb.RegisterControlServer(grpcServer, mockSrv)
go grpcServer.Serve(lis)
defer grpcServer.Stop()

subClient, err := substrate.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
t.Fatalf("failed to create substrate client: %v", err)
}
defer subClient.Close()

reconciler := controller.NewTaskReconciler(subClient, "default-template", "ax-system")
reconciler.SecretResolver = noSecrets
reconciler.WorkspaceReadyTimeout = 200 * time.Millisecond

// Queue a reconcile and then a delete before the worker starts, as when a
// task is deleted while the controller is still busy with other events.
memStore := memory.NewStore()
task := &v1alpha1.Task{
Metadata: &v1alpha1.ObjectMeta{Name: "doomed", Atespace: "default"},
Spec: &v1alpha1.TaskSpec{Image: "ghcr.io/test/img"},
}
if err := memStore.SaveTask(ctx, task); err != nil {
t.Fatalf("failed to save task: %v", err)
}
if err := memStore.MarkTaskDeleting(ctx, "default", "doomed"); err != nil {
t.Fatalf("MarkTaskDeleting failed: %v", err)
}

worker := controller.NewWorker(memStore, reconciler, "test-group", "worker-1")
go func() { _ = worker.Run(ctx) }()

deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
if _, err := memStore.GetTask(ctx, "default", "doomed"); err != nil {
break
}
time.Sleep(50 * time.Millisecond)
}
if _, err := memStore.GetTask(ctx, "default", "doomed"); err == nil {
t.Fatalf("expected task record to be removed after cleanup")
}
if len(mockSrv.createdActors) != 0 || len(mockSrv.resumedActors) != 0 {
t.Errorf("terminating task was reconciled: created %v, resumed %v", mockSrv.createdActors, mockSrv.resumedActors)
}
}
Loading