Apply backpressure to live exec output subscribers (#464)

* Apply backpressure to live exec output subscribers

* Use integer range loops in exec backpressure test

* Preserve reconnectable exec subscriber isolation

---------

Co-authored-by: Yibo Zhuang <yzhuang@openai.com>

.github/workflows/ci.yml

@@ -25,7 +25,7 @@ jobs:
           cache: true
       - uses: golangci/golangci-lint-action@v9
         with:
-          version: v2.12.0
+          version: v2.13.0
           only-new-issues: true
 
   test-linux:

.golangci.yml

@@ -9,7 +9,6 @@ formatters:
     - gofmt
     - gofumpt
     - goimports
-    - golines
     - swaggo
 
 linters:
@@ -41,6 +40,7 @@ linters:
 
     # Style linters that are total nuts.
     - wsl
+    - wsl_v5
     - funlen
 
     # Enough parallelism for now.
@@ -70,6 +70,18 @@ linters:
     # Not all errors need to be checked
     - errcheck
 
+    # It's OK to not initialize some struct fields
+    - exhaustruct
+
+    # We'll control the variable name length ourselves
+    - varnamelen
+
+    # Inline error handling keeps assignment and checking together
+    - noinlineerr
+
+    # Avoid unrelated style churn for now
+    - funcorder
+
 issues:
   # Don't hide multiple issues that belong to one class since GitHub annotations can handle them all nicely.
   max-issues-per-linter: 0

api/openapi.yaml

@@ -791,7 +791,9 @@ components:
             model: macstudio
         hostDirs:
           type: array
-          description: Directories on the Orchard Worker host to mount to a VM
+          description: |
+            Directories on the Orchard Worker host to mount to a VM.
+            Requires running Orchard Controller with `--insecure-allow-host-dirs`.
           items:
             type: object
             properties:

go.mod

@@ -1,6 +1,6 @@
 module github.com/cirruslabs/orchard
 
-go 1.25.1
+go 1.27
 
 // Work around https://github.com/gin-gonic/gin/issues/4372
 replace github.com/gin-gonic/gin v1.11.0 => github.com/gin-gonic/gin v1.10.0

internal/command/controller/run.go

@@ -30,6 +30,7 @@ var addressPprof string
 var debug bool
 var noTLS bool
 var sshNoClientAuth bool
+var insecureAllowHostDirs bool
 var experimentalRPCV2 bool
 var noExperimentalRPCV2 bool
 var experimentalPingInterval time.Duration
@@ -74,6 +75,8 @@ func newRunCommand() *cobra.Command {
 	cmd.Flags().BoolVar(&sshNoClientAuth, "insecure-ssh-no-client-auth", false,
 		"allow SSH clients to connect to the controller's SSH server without authentication, "+
 			"thus only authenticating on the target worker/VM's SSH server")
+	cmd.Flags().BoolVar(&insecureAllowHostDirs, "insecure-allow-host-dirs", false,
+		"allow unsafe path-based local host directory sharing")
 	cmd.Flags().BoolVar(&experimentalRPCV2, "experimental-rpc-v2", false,
 		"enable experimental RPC v2 (https://github.com/cirruslabs/orchard/issues/235)")
 	_ = cmd.Flags().MarkHidden("experimental-rpc-v2")
@@ -166,6 +169,10 @@ func runController(cmd *cobra.Command, args []string) (err error) {
 		controllerOpts = append(controllerOpts, controller.WithSynthetic())
 	}
 
+	if insecureAllowHostDirs {
+		controllerOpts = append(controllerOpts, controller.WithInsecureAllowHostDirs())
+	}
+
 	var controllerCert tls.Certificate
 
 	if !noTLS {

internal/command/dev/dev.go

@@ -35,6 +35,7 @@ var experimentalRPCV2 bool
 var addressPprof string
 var synthetic bool
 var workers int
+var insecureAllowHostDirs bool
 
 func NewCommand() *cobra.Command {
 	command := &cobra.Command{
@@ -57,6 +58,8 @@ func NewCommand() *cobra.Command {
 	command.Flags().BoolVar(&synthetic, "synthetic", false,
 		"do not instantiate real Tart VM, use synthetic in-memory VMs suitable for load testing")
 	command.Flags().IntVar(&workers, "workers", 1, "number of workers to start")
+	command.Flags().BoolVar(&insecureAllowHostDirs, "insecure-allow-host-dirs", false,
+		"allow unsafe path-based local host directory sharing")
 
 	return command
 }
@@ -105,6 +108,10 @@ func runDev(cmd *cobra.Command, args []string) error {
 		additionalControllerOpts = append(additionalControllerOpts, controller.WithExperimentalRPCV2())
 	}
 
+	if insecureAllowHostDirs {
+		additionalControllerOpts = append(additionalControllerOpts, controller.WithInsecureAllowHostDirs())
+	}
+
 	group, ctx := errgroup.WithContext(cmd.Context())
 
 	var additionalWorkerOpts []worker.Option

internal/controller/api_vms.go

@@ -158,6 +158,10 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
 		return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid JSON was provided"))
 	}
 
+	if responder := controller.validateHostDirs(userVM.HostDirs); responder != nil {
+		return responder
+	}
+
 	name := ctx.Param("name")
 
 	return controller.storeUpdate(func(txn storepkg.Transaction) responder.Responder {
@@ -540,6 +544,14 @@ func (controller *Controller) validateHostDirs(hostDirs []v1.HostDir) responder.
 		return nil
 	}
 
+	if !controller.insecureAllowHostDirs {
+		return responder.JSON(
+			http.StatusBadRequest,
+			NewErrorResponse("host directory sharing is disabled; "+
+				"restart the controller with --insecure-allow-host-dirs to enable this unsafe feature"),
+		)
+	}
+
 	// Retrieve cluster settings
 	var clusterSettings *v1.ClusterSettings
 	var err error

internal/controller/controller.go

@@ -46,6 +46,7 @@ type Controller struct {
 	listener                           net.Listener
 	httpServer                         *http.Server
 	insecureAuthDisabled               bool
+	insecureAllowHostDirs              bool
 	scheduler                          *scheduler.Scheduler
 	store                              storepkg.Store
 	logger                             *zap.SugaredLogger
@@ -195,6 +196,14 @@ func New(opts ...Option) (*Controller, error) {
 		return nil, err
 	}
 
+	// When no "--insecure-allow-host-dirs" is present,
+	// fail the VMs that have "hostDirs" set
+	if !controller.insecureAllowHostDirs {
+		if err := controller.failVMsWithHostDirs(); err != nil {
+			return nil, err
+		}
+	}
+
 	return controller, nil
 }
 
@@ -305,6 +314,34 @@ func (controller *Controller) DeleteServiceAccount(name string) error {
 	})
 }
 
+func (controller *Controller) failVMsWithHostDirs() error {
+	return controller.store.Update(func(txn storepkg.Transaction) error {
+		vms, err := txn.ListVMs()
+		if err != nil {
+			return err
+		}
+
+		for _, vm := range vms {
+			permanentlyFailed := vm.TerminalState() &&
+				vm.RestartPolicy == v1.RestartPolicyNever
+
+			if permanentlyFailed || len(vm.HostDirs) == 0 {
+				continue
+			}
+
+			vm.Status = v1.VMStatusFailed
+			vm.StatusMessage = "host directories are used, but host directory sharing is disabled"
+			vm.RestartPolicy = v1.RestartPolicyNever
+
+			if err := txn.SetVM(vm); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
 func (controller *Controller) Run(ctx context.Context) error {
 	// Run the scheduler so that each VM will eventually
 	// be assigned to a specific Worker

internal/controller/exec_sessions.go

@@ -15,14 +15,16 @@ import (
 const execSessionReplayBufferBytes = 4 * 1024 * 1024
 
 type execSessionPolicy struct {
-	closeOnDetach   bool
+	closeOnDetach                 bool
-	retainAfterExit bool
+	retainAfterExit               bool
-	replayEnabled   bool
+	replayEnabled                 bool
+	blockOnSubscriberBackpressure bool
 }
 
 var (
 	legacyExecSessionPolicy = execSessionPolicy{
-		closeOnDetach: true,
+		closeOnDetach:                 true,
+		blockOnSubscriberBackpressure: true,
 	}
 	reconnectableExecSessionPolicy = execSessionPolicy{
 		retainAfterExit: true,
@@ -247,10 +249,14 @@ func newExecSessionSubscriber() *execSessionSubscriber {
 	}
 }
 
-func (subscriber *execSessionSubscriber) enqueue(frame *execstream.Frame) bool {
+func (subscriber *execSessionSubscriber) enqueue(frame *execstream.Frame, block bool) bool {
 	subscriber.sendMu.Lock()
 	defer subscriber.sendMu.Unlock()
 
+	if block {
+		return subscriber.sendLocked(frame)
+	}
+
 	if subscriber.alreadySentLocked(frame) {
 		return true
 	}
@@ -629,7 +635,7 @@ func (session *execSession) recordFrame(frame *execstream.Frame) {
 	session.mu.Unlock()
 
 	for _, subscriber := range subscribers {
-		if !subscriber.enqueue(frame) {
+		if !subscriber.enqueue(frame, session.policy.blockOnSubscriberBackpressure) {
 			session.dropSubscriber(subscriber)
 		}
 	}

internal/controller/exec_sessions_test.go

@@ -266,6 +266,135 @@ func TestExecSessionHistoryReplayStreamsPastSubscriberBuffer(t *testing.T) {
 	}, time.Second, 10*time.Millisecond)
 }
 
+func TestExecSessionLiveOutputAppliesBackpressure(t *testing.T) {
+	session := newManualExecSessionForTest(execSessionKey{vmName: "vm", sessionID: "session"}, nil)
+	session.policy = legacyExecSessionPolicy
+	t.Cleanup(session.close)
+
+	subscriber, err := session.attach()
+	require.NoError(t, err)
+
+	const frameCount = 256
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for i := range frameCount {
+			session.recordFrame(&execstream.Frame{
+				Type:      execstream.FrameTypeStdout,
+				Data:      []byte{byte(i)},
+				Terminal:  nil,
+				Exit:      nil,
+				Error:     "",
+				Watermark: 0,
+			})
+		}
+		session.recordFrame(&execstream.Frame{
+			Type:      execstream.FrameTypeExit,
+			Data:      nil,
+			Terminal:  nil,
+			Exit:      &execstream.Exit{Code: 0},
+			Error:     "",
+			Watermark: 0,
+		})
+	}()
+
+	require.Eventually(t, func() bool {
+		return len(subscriber.frames) == cap(subscriber.frames)
+	}, time.Second, time.Millisecond)
+
+	for i := range frameCount {
+		frame, ok := <-subscriber.frames
+		require.True(t, ok, "subscriber closed before output frame %d", i)
+		require.Equal(t, execstream.FrameTypeStdout, frame.Type)
+		require.Equal(t, []byte{byte(i)}, frame.Data)
+	}
+
+	exitFrame, ok := <-subscriber.frames
+	require.True(t, ok, "subscriber closed before the exit frame")
+	require.Equal(t, execstream.FrameTypeExit, exitFrame.Type)
+	require.EqualValues(t, 0, exitFrame.Exit.Code)
+
+	require.Eventually(t, func() bool {
+		select {
+		case <-done:
+			return true
+		default:
+			return false
+		}
+	}, time.Second, time.Millisecond)
+}
+
+func TestReconnectableExecSessionDropsStalledSubscriber(t *testing.T) {
+	session := newManualExecSessionForTest(execSessionKey{vmName: "vm", sessionID: "session"}, nil)
+	t.Cleanup(session.close)
+
+	stalledSubscriber, err := session.attach()
+	require.NoError(t, err)
+
+	for i := range cap(stalledSubscriber.frames) {
+		session.recordFrame(&execstream.Frame{
+			Type:      execstream.FrameTypeStdout,
+			Data:      []byte{byte(i)},
+			Terminal:  nil,
+			Exit:      nil,
+			Error:     "",
+			Watermark: 0,
+		})
+	}
+
+	healthySubscriber, err := session.attach()
+	require.NoError(t, err)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		session.recordFrame(&execstream.Frame{
+			Type:      execstream.FrameTypeStdout,
+			Data:      []byte("still running"),
+			Terminal:  nil,
+			Exit:      nil,
+			Error:     "",
+			Watermark: 0,
+		})
+		session.recordFrame(&execstream.Frame{
+			Type:      execstream.FrameTypeExit,
+			Data:      nil,
+			Terminal:  nil,
+			Exit:      &execstream.Exit{Code: 0},
+			Error:     "",
+			Watermark: 0,
+		})
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("a stalled reconnectable subscriber blocked live output")
+	}
+
+	outputFrame := <-healthySubscriber.frames
+	require.Equal(t, execstream.FrameTypeStdout, outputFrame.Type)
+	require.Equal(t, []byte("still running"), outputFrame.Data)
+
+	exitFrame := <-healthySubscriber.frames
+	require.Equal(t, execstream.FrameTypeExit, exitFrame.Type)
+	require.EqualValues(t, 0, exitFrame.Exit.Code)
+
+	select {
+	case <-stalledSubscriber.closed:
+	default:
+		t.Fatal("the stalled reconnectable subscriber was not dropped")
+	}
+
+	reconnectedSubscriber, err := session.attach()
+	require.NoError(t, err)
+	session.sendHistory(reconnectedSubscriber, uint64(cap(stalledSubscriber.frames)))
+
+	require.Equal(t, execstream.FrameTypeStdout, (<-reconnectedSubscriber.frames).Type)
+	require.Equal(t, execstream.FrameTypeExit, (<-reconnectedSubscriber.frames).Type)
+	require.Equal(t, execstream.FrameTypeNoMoreHistory, (<-reconnectedSubscriber.frames).Type)
+}
+
 func TestExecSessionDetachKeepsProcessAlive(t *testing.T) {
 	registry := newExecSessionRegistry()
 	session := newManualExecSessionForTest(execSessionKey{vmName: "vm", sessionID: "session"}, registry)

internal/controller/option.go

@@ -48,6 +48,12 @@ func WithInsecureAuthDisabled() Option {
 	}
 }
 
+func WithInsecureAllowHostDirs() Option {
+	return func(controller *Controller) {
+		controller.insecureAllowHostDirs = true
+	}
+}
+
 func WithSwaggerDocs() Option {
 	return func(controller *Controller) {
 		controller.enableSwaggerDocs = true

internal/controller/scheduler/scheduler_capacity_test.go

@@ -0,0 +1,105 @@
+//nolint:testpackage // The regression exercises the unexported scheduler reconciliation loop.
+package scheduler
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cirruslabs/orchard/internal/controller/notifier"
+	storepkg "github.com/cirruslabs/orchard/internal/controller/store"
+	"github.com/cirruslabs/orchard/internal/controller/store/badger"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func TestSchedulingLoopSkipsOvercommittedWorker(t *testing.T) {
+	logger := zap.NewNop().Sugar()
+
+	store, err := badger.NewBadgerStore(t.TempDir(), true, logger)
+	require.NoError(t, err)
+
+	var worker v1.Worker
+	worker.Name = "worker-a"
+	worker.LastSeen = time.Now()
+	worker.MachineID = "machine-a"
+	worker.Resources = v1.Resources{v1.ResourceTartVMs: 2}
+	worker.Arch = v1.ArchitectureARM64
+	worker.Runtime = v1.RuntimeTart
+
+	newPendingVM := func(name string) v1.VM {
+		var vm v1.VM
+		vm.Name = name
+		vm.CreatedAt = time.Now()
+		vm.UID = name + "-uid"
+		vm.Status = v1.VMStatusPending
+		vm.Resources = v1.Resources{v1.ResourceTartVMs: 1}
+		vm.Arch = v1.ArchitectureARM64
+		vm.Runtime = v1.RuntimeTart
+		vm.PowerState = v1.PowerStateRunning
+		vm.Conditions = []v1.Condition{{
+			Type:  v1.ConditionTypeScheduled,
+			State: v1.ConditionStateFalse,
+		}}
+
+		return vm
+	}
+
+	assignedVM := func(name string, status v1.VMStatus) v1.VM {
+		vm := newPendingVM(name)
+		vm.Worker = worker.Name
+		vm.Status = status
+		vm.Conditions[0].State = v1.ConditionStateTrue
+
+		return vm
+	}
+
+	pending := newPendingVM("pending-vm")
+
+	var settings v1.ClusterSettings
+	settings.SchedulerProfile = v1.SchedulerProfileOptimizeUtilization
+
+	err = store.Update(func(txn storepkg.Transaction) error {
+		if err := txn.SetClusterSettings(settings); err != nil {
+			return err
+		}
+
+		if err := txn.SetWorker(worker); err != nil {
+			return err
+		}
+
+		vms := []v1.VM{
+			assignedVM("running-first", v1.VMStatusRunning),
+			assignedVM("running-second", v1.VMStatusRunning),
+			assignedVM("failed-third", v1.VMStatusFailed),
+			pending,
+		}
+
+		for _, vm := range vms {
+			if err := txn.SetVM(vm); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+	require.NoError(t, err)
+
+	scheduler, err := NewScheduler(store, notifier.NewNotifier(logger), time.Minute, logger)
+	require.NoError(t, err)
+
+	numWorkers, numVMs, err := scheduler.schedulingLoopIteration()
+	require.NoError(t, err)
+	require.Equal(t, 1, numWorkers)
+	require.Equal(t, 4, numVMs)
+
+	err = store.View(func(txn storepkg.Transaction) error {
+		currentVM, err := txn.GetVM(pending.Name)
+		require.NoError(t, err)
+		require.False(t, currentVM.IsScheduled())
+		require.Empty(t, currentVM.Worker)
+
+		return nil
+	})
+	require.NoError(t, err)
+}

internal/tests/integration_test.go

@@ -372,12 +372,26 @@ func TestVMGarbageCollection(t *testing.T) {
 	}), "failed to wait for the VM %s to be garbage-collected", vmName)
 }
 
+func TestHostDirsDisabledByDefault(t *testing.T) {
+	devClient, _, _ := devcontroller.StartIntegrationTestEnvironment(t)
+
+	err := devClient.VMs().Create(context.Background(), &v1.VM{
+		Meta:     v1.Meta{Name: "test-host-dirs-disabled"},
+		Image:    imageconstant.DefaultMacosImage,
+		HostDirs: []v1.HostDir{{Name: "src", Path: "/Users/ci/src"}},
+	})
+	require.Error(t, err)
+}
+
 func TestHostDirs(t *testing.T) {
 	if runtime.GOOS != "darwin" {
 		t.Skip("HostDirs is only supported on macOS with Tart")
 	}
 
-	devClient, _, _ := devcontroller.StartIntegrationTestEnvironment(t)
+	devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts(t,
+		false, []controller.Option{controller.WithInsecureAllowHostDirs()},
+		false, nil,
+	)
 
 	dirToMount := t.TempDir()
 
@@ -449,7 +463,10 @@ func TestHostDirsInvalidPolicy(t *testing.T) {
 		t.Skip("HostDirs is only supported on macOS with Tart")
 	}
 
-	devClient, _, _ := devcontroller.StartIntegrationTestEnvironment(t)
+	devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts(t,
+		false, []controller.Option{controller.WithInsecureAllowHostDirs()},
+		false, nil,
+	)
 
 	dirToMount := t.TempDir()
 

internal/worker/rpc.go

@@ -22,7 +22,7 @@ import (
 	"github.com/samber/lo"
 )
 
-func (worker *Worker) watchRPC(ctx context.Context) error {
+func (worker *Worker) watchRPC(ctx context.Context, operationCtx context.Context, onEstablished func()) error {
 	worker.logger.Infof("connecting to %s over gRPC", worker.client.GRPCTarget())
 
 	conn, err := grpc.NewClient(worker.client.GRPCTarget(),
@@ -40,11 +40,13 @@ func (worker *Worker) watchRPC(ctx context.Context) error {
 	client := rpc.NewControllerClient(conn)
 
 	ctxWithMetadata := metadata.NewOutgoingContext(ctx, worker.grpcMetadata())
+	operationCtxWithMetadata := metadata.NewOutgoingContext(operationCtx, worker.grpcMetadata())
 
 	stream, err := client.Watch(ctxWithMetadata, &emptypb.Empty{})
 	if err != nil {
 		return err
 	}
+	onEstablished()
 
 	worker.logger.Infof("running gRPC stream with the controller")
 
@@ -56,11 +58,11 @@ func (worker *Worker) watchRPC(ctx context.Context) error {
 
 		switch action := watchFromController.Action.(type) {
 		case *rpc.WatchInstruction_PortForwardAction:
-			go worker.handlePortForward(ctxWithMetadata, client, action.PortForwardAction)
+			go worker.handlePortForward(operationCtxWithMetadata, client, action.PortForwardAction)
 		case *rpc.WatchInstruction_SyncVmsAction:
 			worker.requestVMSyncing()
 		case *rpc.WatchInstruction_ResolveIpAction:
-			go worker.handleGetIP(ctxWithMetadata, client, action.ResolveIpAction)
+			go worker.handleGetIP(operationCtxWithMetadata, client, action.ResolveIpAction)
 		}
 	}
 }

internal/worker/rpcv2.go

@@ -11,21 +11,22 @@ import (
 	"github.com/samber/lo"
 )
 
-func (worker *Worker) watchRPCV2(ctx context.Context) error {
+func (worker *Worker) watchRPCV2(ctx context.Context, operationCtx context.Context, onEstablished func()) error {
 	watchInstructionCh, watchErrCh, err := worker.client.RPC().Watch(ctx, worker.name)
 	if err != nil {
 		return err
 	}
+	onEstablished()
 
 	for {
 		select {
 		case watchInstruction := <-watchInstructionCh:
 			if portForwardAction := watchInstruction.PortForwardAction; portForwardAction != nil {
-				go worker.handlePortForwardV2(ctx, portForwardAction)
+				go worker.handlePortForwardV2(operationCtx, portForwardAction)
 			} else if syncVMsAction := watchInstruction.SyncVMsAction; syncVMsAction != nil {
 				worker.requestVMSyncing()
 			} else if resolveIPAction := watchInstruction.ResolveIPAction; resolveIPAction != nil {
-				go worker.handleGetIPV2(ctx, resolveIPAction)
+				go worker.handleGetIPV2(operationCtx, resolveIPAction)
 			}
 		case watchErr := <-watchErrCh:
 			return watchErr

internal/worker/worker.go

@@ -11,7 +11,6 @@ import (
 
 	goruntime "runtime"
 
-	"github.com/avast/retry-go/v4"
 	"github.com/cirruslabs/orchard/internal/dialer"
 	"github.com/cirruslabs/orchard/internal/opentelemetry"
 	"github.com/cirruslabs/orchard/internal/worker/dhcpleasetime"
@@ -38,9 +37,18 @@ import (
 const (
 	pollInterval                 = 5 * time.Second
 	workerResourceUpdateInterval = 15 * time.Second
+	recoveredVMProtectionPeriod  = 30 * time.Second
+	rpcWatchReconnectInterval    = 100 * time.Millisecond
+	rpcWatchReconnectMaxInterval = 5 * time.Second
+	rpcWatchReconnectMultiplier  = 2
+	rpcWatchHealthyInterval      = time.Second
+	onDiskVMSyncTimeout          = 30 * time.Second
 )
 
-var ErrPollFailed = errors.New("failed to poll controller")
+var (
+	ErrPollFailed           = errors.New("failed to poll controller")
+	errRPCWatchDisconnected = errors.New("RPC watch disconnected")
+)
 
 type Worker struct {
 	name          string
@@ -49,6 +57,7 @@ type Worker struct {
 	vmm           *vmmanager.VMManager
 	client        *client.Client
 	pollTicker    *time.Ticker
+	recoveredVMs  map[ondiskname.OnDiskName]time.Time
 	resources     v1.Resources
 	labels        v1.Labels
 
@@ -68,6 +77,7 @@ func New(client *client.Client, opts ...Option) (*Worker, error) {
 	worker := &Worker{
 		client:        client,
 		pollTicker:    time.NewTicker(pollInterval),
+		recoveredVMs:  make(map[ondiskname.OnDiskName]time.Time),
 		vmm:           vmmanager.New(),
 		syncRequested: make(chan bool, 1),
 	}
@@ -147,8 +157,20 @@ func (worker *Worker) Run(ctx context.Context) error {
 		}
 	}
 
+	var reconnectBackoff rpcWatchReconnectBackoff
+	reconnectBackoff.reset()
+
 	for {
-		if err := worker.runNewSession(ctx); err != nil {
+		if err := worker.runNewSession(ctx, reconnectBackoff.reset); err != nil {
+			if errors.Is(err, errRPCWatchDisconnected) {
+				select {
+				case <-time.After(reconnectBackoff.next()):
+					continue
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+			}
+
 			return err
 		}
 
@@ -161,6 +183,29 @@ func (worker *Worker) Run(ctx context.Context) error {
 	}
 }
 
+type rpcWatchReconnectBackoff struct {
+	nextInterval time.Duration
+}
+
+func (backoff *rpcWatchReconnectBackoff) next() time.Duration {
+	interval := backoff.nextInterval
+	if interval <= 0 {
+		interval = rpcWatchReconnectInterval
+	}
+
+	if interval >= rpcWatchReconnectMaxInterval/rpcWatchReconnectMultiplier {
+		backoff.nextInterval = rpcWatchReconnectMaxInterval
+	} else {
+		backoff.nextInterval = interval * rpcWatchReconnectMultiplier
+	}
+
+	return min(interval, rpcWatchReconnectMaxInterval)
+}
+
+func (backoff *rpcWatchReconnectBackoff) reset() {
+	backoff.nextInterval = rpcWatchReconnectInterval
+}
+
 func (worker *Worker) Close() error {
 	var result error
 	for _, vm := range worker.vmm.List() {
@@ -175,10 +220,19 @@ func (worker *Worker) Close() error {
 	return result
 }
 
-func (worker *Worker) runNewSession(ctx context.Context) error {
+func (worker *Worker) runNewSession(ctx context.Context, onWatchHealthy func()) error {
 	subCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	// Check the runtime before advertising this worker, but do not touch local
+	// VMs until registration confirms that this worker belongs to this machine.
+	vmInfos, err := worker.listOnDiskVMs(subCtx)
+	if err != nil {
+		worker.logger.Errorf("failed to list on-disk VMs: %v", err)
+
+		return nil
+	}
+
 	if err := worker.registerWorker(subCtx); err != nil {
 		worker.logger.Warnf("failed to register worker: %v", err)
 
@@ -192,35 +246,25 @@ func (worker *Worker) runNewSession(ctx context.Context) error {
 		return nil
 	}
 
-	if info.Capabilities.Has(v1.ControllerCapabilityRPCV2) {
+	group, sessionCtx := errgroup.WithContext(subCtx)
-		worker.logger.Infof("using WebSocket-based v2 RPC")
+	worker.superviseRPCWatch(sessionCtx, ctx, group, info, onWatchHealthy)
-
-		go func() {
-			_ = retry.Do(func() error {
-				return worker.watchRPCV2(subCtx)
-			}, retry.OnRetry(func(n uint, err error) {
-				worker.logger.Warnf("failed to watch RPC v2: %v", err)
-			}), retry.Context(subCtx), retry.Attempts(0))
-		}()
-	} else {
-		worker.logger.Infof("using gRPC-based v1 RPC")
-
-		go func() {
-			_ = retry.Do(func() error {
-				return worker.watchRPC(subCtx)
-			}, retry.OnRetry(func(n uint, err error) {
-				worker.logger.Warnf("failed to watch RPC v1: %v", err)
-			}), retry.Context(subCtx), retry.Attempts(0))
-		}()
-	}
 
 	// Sync on-disk VMs
-	if err := worker.syncOnDiskVMs(ctx); err != nil {
+	if err := worker.syncOnDiskVMsWithInventory(sessionCtx, vmInfos); err != nil {
+		cancel()
+		watchErr := group.Wait()
+
 		worker.logger.Errorf("failed to sync on-disk VMs: %v", err)
 
+		if errors.Is(watchErr, errRPCWatchDisconnected) {
+			return watchErr
+		}
+
 		return nil
 	}
 
+	recoveredVMs := worker.trackRecoveredVMs(time.Now())
+
 	// Backward compatibility with for older Orchard Controllers
 	updateFuncInner := worker.client.VMs().UpdateState
 
@@ -239,17 +283,15 @@ func (worker *Worker) runNewSession(ctx context.Context) error {
 		return err
 	}
 
-	group, ctx := errgroup.WithContext(subCtx)
-
 	group.Go(func() error {
 		for {
-			if err := worker.updateWorker(ctx); err != nil {
+			if err := worker.updateWorker(sessionCtx); err != nil {
 				return fmt.Errorf("failed to update worker resource: %w", err)
 			}
 
 			select {
-			case <-ctx.Done():
+			case <-sessionCtx.Done():
-				return ctx.Err()
+				return sessionCtx.Err()
 			case <-time.After(workerResourceUpdateInterval):
 				// Proceed
 			}
@@ -258,13 +300,13 @@ func (worker *Worker) runNewSession(ctx context.Context) error {
 
 	group.Go(func() error {
 		for {
-			if err := worker.syncVMs(ctx, updateFunc); err != nil {
+			if err := worker.syncVMs(sessionCtx, updateFunc, recoveredVMs); err != nil {
 				return fmt.Errorf("failed to sync VMs: %w", err)
 			}
 
 			select {
-			case <-ctx.Done():
+			case <-sessionCtx.Done():
-				return ctx.Err()
+				return sessionCtx.Err()
 			case <-worker.syncRequested:
 			case <-worker.pollTicker.C:
 				// Proceed
@@ -274,11 +316,102 @@ func (worker *Worker) runNewSession(ctx context.Context) error {
 
 	if err := group.Wait(); err != nil {
 		worker.logger.Errorf("%v", err)
+
+		if errors.Is(err, errRPCWatchDisconnected) {
+			return err
+		}
 	}
 
 	return nil
 }
 
+func (worker *Worker) superviseRPCWatch(
+	sessionCtx context.Context,
+	operationCtx context.Context,
+	group *errgroup.Group,
+	info v1.ControllerInfo,
+	onWatchHealthy func(),
+) {
+	watchRPC := worker.watchRPC
+	rpcVersion := "v1"
+
+	if info.Capabilities.Has(v1.ControllerCapabilityRPCV2) {
+		worker.logger.Infof("using WebSocket-based v2 RPC")
+		watchRPC = worker.watchRPCV2
+		rpcVersion = "v2"
+	} else {
+		worker.logger.Infof("using gRPC-based v1 RPC")
+	}
+
+	watchEstablished := make(chan struct{})
+
+	group.Go(func() error {
+		if err := watchRPC(sessionCtx, operationCtx, func() { close(watchEstablished) }); err != nil {
+			if sessionCtx.Err() != nil {
+				return sessionCtx.Err()
+			}
+
+			return fmt.Errorf("%w: failed to watch RPC %s: %w", errRPCWatchDisconnected, rpcVersion, err)
+		}
+
+		return fmt.Errorf("%w: RPC %s watch closed unexpectedly", errRPCWatchDisconnected, rpcVersion)
+	})
+
+	group.Go(func() error {
+		return monitorRPCWatchHealth(sessionCtx, watchEstablished, rpcWatchHealthyInterval, onWatchHealthy)
+	})
+}
+
+func monitorRPCWatchHealth(
+	ctx context.Context,
+	established <-chan struct{},
+	healthyAfter time.Duration,
+	onHealthy func(),
+) error {
+	select {
+	case <-established:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	healthTimer := time.NewTimer(healthyAfter)
+	defer healthTimer.Stop()
+
+	select {
+	case <-healthTimer.C:
+		onHealthy()
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (worker *Worker) trackRecoveredVMs(now time.Time) map[ondiskname.OnDiskName]time.Time {
+	if worker.recoveredVMs == nil {
+		worker.recoveredVMs = make(map[ondiskname.OnDiskName]time.Time)
+	}
+
+	for onDiskName := range worker.recoveredVMs {
+		if !worker.vmm.Exists(onDiskName) {
+			delete(worker.recoveredVMs, onDiskName)
+		}
+	}
+
+	for _, vm := range worker.vmm.List() {
+		status := vm.Status()
+		if status != v1.VMStatusPending && status != v1.VMStatusRunning {
+			continue
+		}
+
+		onDiskName := vm.OnDiskName()
+		if _, alreadyTracked := worker.recoveredVMs[onDiskName]; !alreadyTracked {
+			worker.recoveredVMs[onDiskName] = now.Add(recoveredVMProtectionPeriod)
+		}
+	}
+
+	return worker.recoveredVMs
+}
+
 func (worker *Worker) registerWorker(ctx context.Context) error {
 	platformUUID, err := platform.MachineID()
 	if err != nil {
@@ -326,8 +459,12 @@ func (worker *Worker) updateWorker(ctx context.Context) error {
 	return nil
 }
 
-//nolint:nestif,gocognit // nested "if" and cognitive complexity is tolerable for now
+//nolint:gocognit // VM lifecycle branches are clearest in a single reconciliation loop.
-func (worker *Worker) syncVMs(ctx context.Context, updateVM func(context.Context, v1.VM) error) error {
+func (worker *Worker) syncVMs(
+	ctx context.Context,
+	updateVM func(context.Context, v1.VM) error,
+	recoveredVMs map[ondiskname.OnDiskName]time.Time,
+) error {
 	allKeys := mapset.NewSet[ondiskname.OnDiskName]()
 
 	remoteVMs, err := worker.client.VMs().FindForWorker(ctx, worker.name)
@@ -367,6 +504,8 @@ func (worker *Worker) syncVMs(ctx context.Context, updateVM func(context.Context
 	// we risk violating the scheduler resource assumptions
 	sortNonExistentAndFailedFirst(pairs)
 
+	hasUnaccountedRecoveredVM := false
+
 	for _, tuple := range pairs {
 		onDiskName, vmResource, vm := lo.Unpack3(tuple)
 
@@ -382,6 +521,14 @@ func (worker *Worker) syncVMs(ctx context.Context, updateVM func(context.Context
 			localConditions = vm.Conditions()
 		}
 
+		if shouldPreserveRecoveredVM(recoveredVMs, onDiskName, vmResource, localState, time.Now()) {
+			hasUnaccountedRecoveredVM = true
+			worker.logger.Warnf("preserving active VM %s missing from controller during worker session recovery",
+				onDiskName)
+
+			continue
+		}
+
 		action := transitions[remoteState][localState]
 
 		worker.logger.Debugf("processing VM: %s, remote state: %s, local state: %s, "+
@@ -391,6 +538,13 @@ func (worker *Worker) syncVMs(ctx context.Context, updateVM func(context.Context
 		switch action {
 		case ActionCreate:
 			// Remote VM was created, but not the local VM
+			if hasUnaccountedRecoveredVM {
+				worker.logger.Warnf("deferring VM %s while recovered VMs are missing from controller inventory",
+					onDiskName)
+
+				continue
+			}
+
 			worker.createVM(onDiskName, *vmResource)
 		case ActionMonitorPending:
 			if vmResource.StatusMessage != vm.StatusMessage() {
@@ -418,69 +572,22 @@ func (worker *Worker) syncVMs(ctx context.Context, updateVM func(context.Context
 				return err
 			}
 		case ActionMonitorRunning:
-			if vmResource.Generation != vm.Resource().Generation {
+			if err := worker.monitorRunningVM(ctx, vmResource, vm, updateVM); err != nil {
-				// VM specification changed, reboot the VM for the changes to take effect
+				return err
-				stoppingOrSuspending := v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeStopping) ||
-					v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeSuspending)
-
-				if v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) && !stoppingOrSuspending {
-					// VM is running, suspend or stop it first
-					shouldStop := vmResource.PowerState == v1.PowerStateStopped || !vm.Resource().Suspendable
-
-					if shouldStop {
-						vm.Stop()
-					} else {
-						vm.Suspend()
-					}
-				}
-
-				if v1.ConditionIsFalse(vm.Conditions(), v1.ConditionTypeRunning) && !stoppingOrSuspending {
-					// VM stopped, update its specification
-					vm.SetResource(*vmResource)
-
-					if vmResource.PowerState == v1.PowerStateRunning {
-						// Start the VM
-						eventStreamer := worker.client.VMs().StreamEvents(vmResource.Name)
-						vm.Start(eventStreamer)
-					}
-				}
-			}
-
-			var updateNeeded bool
-
-			if vmResource.StatusMessage != vm.StatusMessage() {
-				vmResource.StatusMessage = vm.StatusMessage()
-
-				updateNeeded = true
-			}
-
-			if vmResource.ObservedGeneration != vm.Resource().ObservedGeneration {
-				vmResource.ObservedGeneration = vm.Resource().ObservedGeneration
-
-				updateNeeded = true
-			}
-
-			// Propagate VM's conditions to the Orchard Controller
-			for _, condition := range vm.Conditions() {
-				if v1.ConditionsSet(&vmResource.Conditions, condition) {
-					updateNeeded = true
-				}
-			}
-
-			if updateNeeded {
-				if err := updateVM(ctx, *vmResource); err != nil {
-					return err
-				}
 			}
 		case ActionStop:
 			// VM has failed on the remote side, stop it locally to prevent incorrect
 			// worker's resources calculation in the Controller's scheduler
-			vm.Stop()
+			if err := waitForVMStop(ctx, vm); err != nil {
+				return fmt.Errorf("failed to stop VM: %w", err)
+			}
 		case ActionFail, ActionLostTrack, ActionImpossible:
 			// VM has failed on the local side, stop it before reporting as failed to prevent incorrect
 			// worker's resources calculation in the Controller's scheduler
 			if vm != nil {
-				vm.Stop()
+				if err := waitForVMStop(ctx, vm); err != nil {
+					return fmt.Errorf("failed to stop VM: %w", err)
+				}
 			}
 
 			var statusMessage string
@@ -513,10 +620,135 @@ func (worker *Worker) syncVMs(ctx context.Context, updateVM func(context.Context
 	return nil
 }
 
-//nolint:nestif,gocognit // complexity is tolerable for now
+func (worker *Worker) monitorRunningVM(
-func (worker *Worker) syncOnDiskVMs(ctx context.Context) error {
+	ctx context.Context,
+	vmResource *v1.VM,
+	vm vmmanager.VM,
+	updateVM func(context.Context, v1.VM) error,
+) error {
+	worker.reconcileRunningVM(vmResource, vm) //nolint:contextcheck // Event streams outlive sync sessions.
+
+	var updateNeeded bool
+
+	if vmResource.StatusMessage != vm.StatusMessage() {
+		vmResource.StatusMessage = vm.StatusMessage()
+
+		updateNeeded = true
+	}
+
+	if vmResource.ObservedGeneration != vm.Resource().ObservedGeneration {
+		vmResource.ObservedGeneration = vm.Resource().ObservedGeneration
+
+		updateNeeded = true
+	}
+
+	// Propagate VM's conditions to the Orchard Controller
+	for _, condition := range vm.Conditions() {
+		if v1.ConditionsSet(&vmResource.Conditions, condition) {
+			updateNeeded = true
+		}
+	}
+
+	if updateNeeded {
+		return updateVM(ctx, *vmResource)
+	}
+
+	return nil
+}
+
+func (worker *Worker) reconcileRunningVM(vmResource *v1.VM, vm vmmanager.VM) {
+	if vmResource.Generation == vm.Resource().Generation {
+		return
+	}
+
+	stoppingOrSuspending := v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeStopping) ||
+		v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeSuspending)
+	if stoppingOrSuspending {
+		return
+	}
+
+	if v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) {
+		// VM is running, suspend or stop it first.
+		shouldStop := vmResource.PowerState == v1.PowerStateStopped || !vm.Resource().Suspendable
+
+		if shouldStop {
+			vm.Stop()
+		} else {
+			vm.Suspend()
+		}
+	}
+
+	if v1.ConditionIsFalse(vm.Conditions(), v1.ConditionTypeRunning) {
+		// VM stopped, update its specification.
+		vm.SetResource(*vmResource)
+
+		if vmResource.PowerState == v1.PowerStateRunning {
+			// Start the VM.
+			eventStreamer := worker.client.VMs().StreamEvents(vmResource.Name)
+			vm.Start(eventStreamer)
+		}
+	}
+}
+
+func shouldPreserveRecoveredVM(
+	recoveredVMs map[ondiskname.OnDiskName]time.Time,
+	onDiskName ondiskname.OnDiskName,
+	remoteVM *v1.VM,
+	localState mo.Option[v1.VMStatus],
+	now time.Time,
+) bool {
+	deadline, recovered := recoveredVMs[onDiskName]
+	if !recovered {
+		return false
+	}
+
+	active := localState == mo.Some(v1.VMStatusPending) || localState == mo.Some(v1.VMStatusRunning)
+	if remoteVM == nil && active && now.Before(deadline) {
+		return true
+	}
+
+	// Once the controller recognizes a recovered VM, or the bounded recovery
+	// window expires, user-requested deletion follows the normal lifecycle.
+	delete(recoveredVMs, onDiskName)
+
+	return false
+}
+
+func (worker *Worker) listOnDiskVMs(ctx context.Context) ([]vmmanager.VMInfo, error) {
 	if worker.runtime.Synthetic() {
 		// There's no on-disk VMs when using synthetic VMs
+		return nil, nil
+	}
+
+	worker.logger.Infof("listing on-disk VMs...")
+
+	runtimeCtx, cancelRuntime := context.WithTimeout(ctx, onDiskVMSyncTimeout)
+	defer cancelRuntime()
+
+	vmInfos, err := worker.runtime.ListVMs(runtimeCtx, worker.logger)
+	if err != nil {
+		if errors.Is(runtimeCtx.Err(), context.DeadlineExceeded) {
+			return nil, fmt.Errorf("timed out listing on-disk VMs: %w", context.DeadlineExceeded)
+		}
+
+		return nil, err
+	}
+
+	return vmInfos, nil
+}
+
+func (worker *Worker) syncOnDiskVMs(ctx context.Context) error {
+	vmInfos, err := worker.listOnDiskVMs(ctx)
+	if err != nil {
+		return err
+	}
+
+	return worker.syncOnDiskVMsWithInventory(ctx, vmInfos)
+}
+
+//nolint:nestif,gocognit // complexity is tolerable for now
+func (worker *Worker) syncOnDiskVMsWithInventory(ctx context.Context, vmInfos []vmmanager.VMInfo) error {
+	if worker.runtime.Synthetic() {
 		return nil
 	}
 
@@ -531,11 +763,6 @@ func (worker *Worker) syncOnDiskVMs(ctx context.Context) error {
 
 	worker.logger.Infof("syncing on-disk VMs...")
 
-	vmInfos, err := worker.runtime.ListVMs(ctx, worker.logger)
-	if err != nil {
-		return err
-	}
-
 	for _, vmInfo := range vmInfos {
 		onDiskName, err := ondiskname.Parse(vmInfo.Name)
 		if err != nil {
@@ -582,6 +809,15 @@ func (worker *Worker) syncOnDiskVMs(ctx context.Context) error {
 	return nil
 }
 
+func waitForVMStop(ctx context.Context, vm vmmanager.VM) error {
+	select {
+	case err := <-vm.Stop():
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
 func (worker *Worker) deleteVM(vm vmmanager.VM) error {
 	<-vm.Stop()
 

internal/worker/worker_stop_test.go

@@ -0,0 +1,135 @@
+package worker //nolint:testpackage // The regression test exercises unexported worker reconciliation.
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
+	"github.com/cirruslabs/orchard/internal/worker/vmmanager"
+	"github.com/cirruslabs/orchard/pkg/client"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+var errLocalVMFailed = errors.New("local VM failed")
+
+type delayedStopVM struct {
+	vmmanager.VM
+
+	resource    v1.VM
+	status      v1.VMStatus
+	stopStarted chan struct{}
+	stopResult  chan error
+}
+
+func (vm *delayedStopVM) OnDiskName() ondiskname.OnDiskName {
+	return ondiskname.NewFromResource(vm.resource)
+}
+
+func (vm *delayedStopVM) Status() v1.VMStatus { return vm.status }
+
+func (vm *delayedStopVM) Conditions() []v1.Condition { return nil }
+
+func (vm *delayedStopVM) Err() error { return errLocalVMFailed }
+
+func (vm *delayedStopVM) Stop() <-chan error {
+	close(vm.stopStarted)
+	return vm.stopResult
+}
+
+func TestSyncVMsWaitsForVMShutdown(t *testing.T) {
+	tests := []struct {
+		name         string
+		remoteStatus v1.VMStatus
+		localStatus  v1.VMStatus
+		update       bool
+	}{
+		{
+			name:         "remote failed VM stops before reconciliation continues",
+			remoteStatus: v1.VMStatusFailed,
+			localStatus:  v1.VMStatusRunning,
+		},
+		{
+			name:         "local failed VM stops before reporting failure",
+			remoteStatus: v1.VMStatusRunning,
+			localStatus:  v1.VMStatusFailed,
+			update:       true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resource := v1.VM{
+				Meta:   v1.Meta{Name: "test-vm"},
+				UID:    "00112233-4455-6677-8899-aabbccddeeff",
+				Worker: "test-worker",
+				Status: test.remoteStatus,
+			}
+
+			server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+				if request.Method != http.MethodGet || request.URL.Path != "/v1/vms" {
+					t.Errorf("unexpected request: %s %s", request.Method, request.URL.Path)
+					return
+				}
+
+				response.Header().Set("Content-Type", "application/json")
+				if err := json.NewEncoder(response).Encode([]map[string]any{{
+					"name":   resource.Name,
+					"uid":    resource.UID,
+					"worker": resource.Worker,
+					"status": resource.Status,
+				}}); err != nil {
+					t.Errorf("encode VM response: %v", err)
+				}
+			}))
+			defer server.Close()
+
+			apiClient, err := client.New(client.WithAddress(server.URL))
+			require.NoError(t, err)
+			vm := &delayedStopVM{
+				resource:    resource,
+				status:      test.localStatus,
+				stopStarted: make(chan struct{}),
+				stopResult:  make(chan error, 1),
+			}
+			manager := vmmanager.New()
+			manager.Put(vm.OnDiskName(), vm)
+			worker := &Worker{name: "test-worker", client: apiClient, vmm: manager, logger: zap.NewNop().Sugar()}
+			updated := make(chan v1.VM, 1)
+			finished := make(chan error, 1)
+
+			go func() {
+				finished <- worker.syncVMs(context.Background(), func(_ context.Context, updatedVM v1.VM) error {
+					updated <- updatedVM
+					return nil
+				}, nil)
+			}()
+
+			select {
+			case <-vm.stopStarted:
+			case <-time.After(time.Second):
+				t.Fatal("VM shutdown was not requested")
+			}
+
+			select {
+			case updatedVM := <-updated:
+				t.Fatalf("VM was reported %q before shutdown completed", updatedVM.Status)
+			case err := <-finished:
+				t.Fatalf("reconciliation continued before VM shutdown completed: %v", err)
+			case <-time.After(30 * time.Millisecond):
+			}
+
+			vm.stopResult <- nil
+			require.NoError(t, <-finished)
+			if test.update {
+				require.Equal(t, v1.VMStatusFailed, (<-updated).Status)
+			}
+		})
+	}
+}

internal/worker/worker_test.go

@@ -1,16 +1,917 @@
 package worker
 
 import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
+	"github.com/cirruslabs/orchard/internal/worker/runtime"
 	"github.com/cirruslabs/orchard/internal/worker/vmmanager"
 	"github.com/cirruslabs/orchard/internal/worker/vmmanager/tart"
+	"github.com/cirruslabs/orchard/pkg/client"
 	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/cirruslabs/orchard/rpc"
+	"github.com/coder/websocket"
 	"github.com/samber/lo"
+	"github.com/samber/mo"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/emptypb"
 )
 
+const (
+	recoveryTestWorkerName = "worker-a"
+	recoveryTestVMUID      = "running-vm-uid"
+	recoveryTestWorkerPath = "/v1/workers/" + recoveryTestWorkerName
+	recoveryTestVMsPath    = "/v1/vms"
+	recoveryTestWatchPath  = "/v1/rpc/watch"
+	recoveryTestInfoPath   = "/v1/controller/info"
+	startupTestWorkersPath = "/v1/workers"
+	startupTestVMUID       = "11111111-2222-4333-8444-555555555555"
+)
+
+func TestSyncOnDiskVMsCancelsStuckTartList(t *testing.T) {
+	worker := newWorkerWithFakeTart(t, "#!/bin/sh\nexec /bin/sleep 60\n")
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+
+	started := time.Now()
+	err := worker.syncOnDiskVMs(ctx)
+
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	require.ErrorContains(t, err, "timed out listing on-disk VMs")
+	require.Less(t, time.Since(started), time.Second)
+}
+
+func TestSyncOnDiskVMsPreservesTartPermissionError(t *testing.T) {
+	worker := newWorkerWithFakeTart(t,
+		"#!/bin/sh\necho 'Failed to perform garbage collection: NSCocoaErrorDomain Code=257' >&2\nexit 1\n",
+	)
+
+	err := worker.syncOnDiskVMs(context.Background())
+
+	require.Error(t, err)
+	require.NotErrorIs(t, err, context.DeadlineExceeded)
+	require.ErrorContains(t, err, "Code=257")
+}
+
+func TestRunNewSessionDoesNotRegisterWorkerWhenTartFails(t *testing.T) {
+	var registrations atomic.Int32
+	worker := newWorkerWithFakeTart(t,
+		"#!/bin/sh\necho 'Failed to perform garbage collection: NSCocoaErrorDomain Code=257' >&2\nexit 1\n",
+		func(_ http.ResponseWriter, request *http.Request) bool {
+			if request.Method == http.MethodPost && request.URL.Path == startupTestWorkersPath {
+				registrations.Add(1)
+			}
+
+			return false
+		},
+	)
+
+	require.NoError(t, worker.runNewSession(context.Background(), func() {}))
+	require.Zero(t, registrations.Load(), "workers with unusable Tart storage must not appear healthy")
+}
+
+func TestRunNewSessionDoesNotDeleteVMsWhenWorkerIdentityConflicts(t *testing.T) {
+	onDiskName := ondiskname.New("protected-vm", startupTestVMUID, 0).String()
+	script, commandsPath := fakeTartInventoryScript(t, onDiskName)
+	var registrations atomic.Int32
+
+	worker := newWorkerWithFakeTart(t, script,
+		func(writer http.ResponseWriter, request *http.Request) bool {
+			if request.Method != http.MethodPost || request.URL.Path != startupTestWorkersPath {
+				return false
+			}
+
+			registrations.Add(1)
+			writer.WriteHeader(http.StatusConflict)
+
+			return true
+		},
+	)
+
+	require.NoError(t, worker.runNewSession(context.Background(), func() {}))
+	require.Equal(t, int32(1), registrations.Load())
+
+	commands, err := readFakeTartCommands(commandsPath)
+	require.NoError(t, err)
+	require.Equal(t, "list\n", string(commands),
+		"registration conflicts must not stop or delete local VMs")
+}
+
+func TestRunNewSessionReconcilesVMsOnlyAfterWorkerRegistration(t *testing.T) {
+	onDiskName := ondiskname.New("orphaned-vm", startupTestVMUID, 0).String()
+	script, commandsPath := fakeTartInventoryScript(t, onDiskName)
+	var commandsAtRegistration atomic.Value
+
+	worker := newWorkerWithFakeTart(t, script,
+		func(writer http.ResponseWriter, request *http.Request) bool {
+			if request.Method != http.MethodPost || request.URL.Path != startupTestWorkersPath {
+				return false
+			}
+
+			commands, err := readFakeTartCommands(commandsPath)
+			if err != nil {
+				t.Errorf("failed to inspect Tart commands at worker registration: %v", err)
+				writer.WriteHeader(http.StatusInternalServerError)
+
+				return true
+			}
+			commandsAtRegistration.Store(string(commands))
+
+			var workerResource v1.Worker
+			if err := json.NewDecoder(request.Body).Decode(&workerResource); err != nil {
+				t.Errorf("failed to decode worker registration: %v", err)
+				writer.WriteHeader(http.StatusBadRequest)
+
+				return true
+			}
+
+			writeRecoveryTestJSON(t, writer, workerResource)
+
+			return true
+		},
+	)
+
+	require.NoError(t, worker.runNewSession(context.Background(), func() {}))
+	require.Equal(t, "list\n", commandsAtRegistration.Load(),
+		"local VMs must not be reconciled until worker registration succeeds")
+
+	commands, err := readFakeTartCommands(commandsPath)
+	require.NoError(t, err)
+	require.Equal(t, "list\nstop\ndelete\n", string(commands),
+		"successful registration should reconcile the original inventory without listing twice")
+}
+
+func fakeTartInventoryScript(t *testing.T, onDiskName string) (string, string) {
+	t.Helper()
+
+	commandsPath := filepath.Join(t.TempDir(), "tart-commands")
+	inventory, err := json.Marshal([]struct {
+		Name    string `json:"name"`
+		Running bool   `json:"running"`
+	}{{Name: onDiskName, Running: true}})
+	require.NoError(t, err)
+
+	script := fmt.Sprintf("#!/bin/sh\nprintf '%%s\\n' \"$1\" >> %q\n"+
+		"if [ \"$1\" = list ]; then\nprintf '%%s\\n' '%s'\nfi\n", commandsPath, inventory)
+
+	return script, commandsPath
+}
+
+func readFakeTartCommands(commandsPath string) ([]byte, error) {
+	return os.ReadFile(filepath.Clean(commandsPath))
+}
+
+func newWorkerWithFakeTart(
+	t *testing.T,
+	script string,
+	observeRequests ...func(http.ResponseWriter, *http.Request) bool,
+) *Worker {
+	t.Helper()
+
+	binDir := t.TempDir()
+	fakeTartPath := filepath.Join(binDir, "tart")
+	require.NoError(t, os.WriteFile(fakeTartPath, []byte(script), 0o600))
+	require.NoError(t, os.Chmod(fakeTartPath, 0o700)) //nolint:gosec // Fake Tart must be executable.
+	t.Setenv("PATH", binDir)
+
+	controller := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		for _, observeRequest := range observeRequests {
+			if observeRequest(writer, request) {
+				return
+			}
+		}
+
+		switch request.URL.Path {
+		case recoveryTestInfoPath:
+			writeRecoveryTestJSON(t, writer, v1.ControllerInfo{
+				Capabilities: v1.ControllerCapabilities{v1.ControllerCapabilityRPCV2},
+			})
+		case recoveryTestWatchPath:
+			connection, err := websocket.Accept(writer, request, nil)
+			if err != nil {
+				t.Errorf("failed to accept RPC watch: %v", err)
+
+				return
+			}
+			defer connection.CloseNow()
+
+			<-request.Context().Done()
+		case recoveryTestVMsPath:
+			writeRecoveryTestJSON(t, writer, []v1.VM{})
+		default:
+			http.NotFound(writer, request)
+		}
+	}))
+	t.Cleanup(controller.Close)
+
+	controllerClient, err := client.New(client.WithAddress(controller.URL))
+	require.NoError(t, err)
+
+	pollTicker := time.NewTicker(pollInterval)
+	t.Cleanup(pollTicker.Stop)
+
+	return &Worker{
+		name:          "worker-a",
+		client:        controllerClient,
+		vmm:           vmmanager.New(),
+		pollTicker:    pollTicker,
+		syncRequested: make(chan bool, 1),
+		runtime:       runtime.NewTart(),
+		logger:        zap.NewNop().Sugar(),
+	}
+}
+
+func TestWorkerRecoversControllerSessionWithoutDeletingRunningVM(t *testing.T) {
+	firstHeartbeat := make(chan struct{})
+	reregistered := make(chan struct{})
+	var firstHeartbeatOnce sync.Once
+	var registrations atomic.Int32
+	var watches atomic.Int32
+
+	vmResource := v1.VM{
+		Meta:   v1.Meta{Name: "running-vm"},
+		UID:    recoveryTestVMUID,
+		Worker: recoveryTestWorkerName,
+		Status: v1.VMStatusRunning,
+	}
+	recoveredVM := &recoveryTestVM{
+		resource:       vmResource,
+		conditionsSeen: make(chan struct{}),
+	}
+
+	controller := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		switch {
+		case request.Method == http.MethodPost && request.URL.Path == "/v1/workers":
+			var workerResource v1.Worker
+			if err := json.NewDecoder(request.Body).Decode(&workerResource); err != nil {
+				t.Errorf("failed to decode worker registration: %v", err)
+				return
+			}
+
+			if registrations.Add(1) == 2 {
+				close(reregistered)
+			}
+
+			writeRecoveryTestJSON(t, writer, workerResource)
+		case request.Method == http.MethodGet && request.URL.Path == recoveryTestInfoPath:
+			writeRecoveryTestJSON(t, writer, v1.ControllerInfo{
+				Capabilities: v1.ControllerCapabilities{v1.ControllerCapabilityRPCV2},
+			})
+		case request.Method == http.MethodGet && request.URL.Path == recoveryTestWorkerPath:
+			writeRecoveryTestJSON(t, writer, v1.Worker{Meta: v1.Meta{Name: recoveryTestWorkerName}})
+		case request.Method == http.MethodPut && request.URL.Path == recoveryTestWorkerPath:
+			var workerResource v1.Worker
+			if err := json.NewDecoder(request.Body).Decode(&workerResource); err != nil {
+				t.Errorf("failed to decode worker heartbeat: %v", err)
+				return
+			}
+
+			firstHeartbeatOnce.Do(func() { close(firstHeartbeat) })
+			writeRecoveryTestJSON(t, writer, workerResource)
+		case request.Method == http.MethodGet && request.URL.Path == recoveryTestVMsPath:
+			writeRecoveryTestJSON(t, writer, []v1.VM{})
+		case request.URL.Path == recoveryTestWatchPath:
+			handleRecoveryTestWatch(t, writer, request, &watches, firstHeartbeat, recoveredVM.conditionsSeen)
+		default:
+			http.NotFound(writer, request)
+		}
+	}))
+	t.Cleanup(controller.Close)
+
+	controllerClient, err := client.New(client.WithAddress(controller.URL))
+	require.NoError(t, err)
+
+	worker, err := New(controllerClient, WithName(recoveryTestWorkerName), WithSynthetic(), WithLogger(zap.NewNop()))
+	require.NoError(t, err)
+
+	t.Cleanup(worker.pollTicker.Stop)
+	worker.vmm.Put(ondiskname.NewFromResource(vmResource), recoveredVM)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	runResult := make(chan error, 1)
+	go func() {
+		runResult <- worker.Run(ctx)
+	}()
+
+	select {
+	case <-reregistered:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("worker did not re-register after its controller RPC watch disconnected (registrations=%d, watches=%d)",
+			registrations.Load(), watches.Load())
+	}
+
+	require.True(t, worker.vmm.Exists(ondiskname.NewFromResource(vmResource)))
+	require.False(t, recoveredVM.stopped.Load(), "controller recovery must not stop a running VM")
+	require.False(t, recoveredVM.deleted.Load(), "controller recovery must not delete a running VM")
+
+	cancel()
+	require.ErrorIs(t, <-runResult, context.Canceled)
+}
+
+func handleRecoveryTestWatch(
+	t *testing.T,
+	writer http.ResponseWriter,
+	request *http.Request,
+	watches *atomic.Int32,
+	firstHeartbeat <-chan struct{},
+	conditionsSeen <-chan struct{},
+) {
+	t.Helper()
+
+	connection, err := websocket.Accept(writer, request, nil)
+	if err != nil {
+		t.Errorf("failed to accept RPC watch: %v", err)
+		return
+	}
+	defer connection.CloseNow()
+
+	if watches.Add(1) == 1 {
+		select {
+		case <-firstHeartbeat:
+		case <-request.Context().Done():
+			return
+		}
+
+		select {
+		case <-conditionsSeen:
+		case <-request.Context().Done():
+			return
+		}
+
+		_ = connection.Close(websocket.StatusGoingAway, "controller rollout")
+		return
+	}
+
+	<-request.Context().Done()
+}
+
+func TestRPCWatchReconnectBackoff(t *testing.T) {
+	var backoff rpcWatchReconnectBackoff
+	backoff.reset()
+
+	expected := []time.Duration{
+		100 * time.Millisecond,
+		200 * time.Millisecond,
+		400 * time.Millisecond,
+		800 * time.Millisecond,
+		1600 * time.Millisecond,
+		3200 * time.Millisecond,
+		5 * time.Second,
+		5 * time.Second,
+	}
+	for _, interval := range expected {
+		require.Equal(t, interval, backoff.next())
+	}
+
+	backoff.reset()
+	require.Equal(t, 100*time.Millisecond, backoff.next(),
+		"a successfully connected RPC watch should restore the fast initial retry")
+}
+
+func TestMonitorRPCWatchHealth(t *testing.T) {
+	t.Run("closed watch is not healthy", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		established := make(chan struct{})
+		close(established)
+		var markedHealthy atomic.Bool
+		result := make(chan error, 1)
+
+		go func() {
+			result <- monitorRPCWatchHealth(ctx, established, time.Second, func() {
+				markedHealthy.Store(true)
+			})
+		}()
+
+		cancel()
+		require.ErrorIs(t, <-result, context.Canceled)
+		require.False(t, markedHealthy.Load())
+	})
+
+	t.Run("stable watch resets backoff", func(t *testing.T) {
+		established := make(chan struct{})
+		close(established)
+		var backoff rpcWatchReconnectBackoff
+		backoff.nextInterval = rpcWatchReconnectMaxInterval
+
+		require.NoError(t, monitorRPCWatchHealth(context.Background(), established,
+			10*time.Millisecond, backoff.reset))
+		require.Equal(t, rpcWatchReconnectInterval, backoff.next())
+	})
+}
+
+func TestWorkerBacksOffPersistentRPCWatchFailures(t *testing.T) {
+	testCases := []struct {
+		name       string
+		useRPCV2   bool
+		closeAfter bool
+	}{
+		{name: "HTTP rejection", useRPCV2: true},
+		{name: "WebSocket closes immediately", useRPCV2: true, closeAfter: true},
+		{name: "gRPC first receive fails"},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			testWorkerBacksOffPersistentRPCWatchFailure(t, testCase.useRPCV2, testCase.closeAfter)
+		})
+	}
+}
+
+func testWorkerBacksOffPersistentRPCWatchFailure(t *testing.T, useRPCV2 bool, closeAfter bool) {
+	t.Helper()
+
+	watchAttempts := make(chan time.Time, 4)
+	grpcServer := grpc.NewServer()
+	rpc.RegisterControllerServer(grpcServer, &failingRecoveryRPCServer{watchAttempts: watchAttempts})
+	t.Cleanup(grpcServer.Stop)
+
+	handler := http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		if request.Header.Get("Content-Type") == "application/grpc" {
+			grpcServer.ServeHTTP(writer, request)
+			return
+		}
+
+		switch {
+		case request.Method == http.MethodPost && request.URL.Path == "/v1/workers":
+			var workerResource v1.Worker
+			if err := json.NewDecoder(request.Body).Decode(&workerResource); err != nil {
+				t.Errorf("failed to decode worker registration: %v", err)
+				return
+			}
+			writeRecoveryTestJSON(t, writer, workerResource)
+		case request.Method == http.MethodGet && request.URL.Path == recoveryTestInfoPath:
+			info := v1.ControllerInfo{}
+			if useRPCV2 {
+				info.Capabilities = v1.ControllerCapabilities{v1.ControllerCapabilityRPCV2}
+			}
+			writeRecoveryTestJSON(t, writer, info)
+		case request.Method == http.MethodGet && request.URL.Path == recoveryTestWorkerPath:
+			writeRecoveryTestJSON(t, writer, v1.Worker{Meta: v1.Meta{Name: recoveryTestWorkerName}})
+		case request.Method == http.MethodPut && request.URL.Path == recoveryTestWorkerPath:
+			writeRecoveryTestJSON(t, writer, v1.Worker{Meta: v1.Meta{Name: recoveryTestWorkerName}})
+		case request.Method == http.MethodGet && request.URL.Path == recoveryTestVMsPath:
+			writeRecoveryTestJSON(t, writer, []v1.VM{})
+		case request.URL.Path == recoveryTestWatchPath:
+			select {
+			case watchAttempts <- time.Now():
+			default:
+			}
+
+			if closeAfter {
+				connection, err := websocket.Accept(writer, request, nil)
+				if err != nil {
+					t.Errorf("failed to accept rapidly closing RPC watch: %v", err)
+					return
+				}
+
+				_ = connection.Close(websocket.StatusGoingAway, "upstream unavailable")
+				return
+			}
+
+			writer.WriteHeader(http.StatusForbidden)
+		default:
+			http.NotFound(writer, request)
+		}
+	})
+	controller := httptest.NewServer(h2c.NewHandler(handler, &http2.Server{}))
+	t.Cleanup(controller.Close)
+
+	controllerClient, err := client.New(client.WithAddress(controller.URL))
+	require.NoError(t, err)
+	worker, err := New(controllerClient, WithName(recoveryTestWorkerName), WithSynthetic(), WithLogger(zap.NewNop()))
+	require.NoError(t, err)
+	t.Cleanup(worker.pollTicker.Stop)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	runResult := make(chan error, 1)
+	go func() {
+		runResult <- worker.Run(ctx)
+	}()
+
+	attempts := make([]time.Time, 0, 4)
+	for len(attempts) < 4 {
+		select {
+		case attemptedAt := <-watchAttempts:
+			attempts = append(attempts, attemptedAt)
+		case <-time.After(3 * time.Second):
+			t.Fatalf("worker made only %d RPC watch attempts", len(attempts))
+		}
+	}
+
+	expectedMinimums := []time.Duration{100 * time.Millisecond, 200 * time.Millisecond, 400 * time.Millisecond}
+	for index, minimum := range expectedMinimums {
+		actual := attempts[index+1].Sub(attempts[index])
+		require.GreaterOrEqual(t, actual, minimum-20*time.Millisecond,
+			"RPC watch retry %d did not increase its backoff", index+1)
+	}
+
+	cancel()
+	require.ErrorIs(t, <-runResult, context.Canceled)
+}
+
+type failingRecoveryRPCServer struct {
+	rpc.UnimplementedControllerServer
+
+	watchAttempts chan time.Time
+}
+
+func (server *failingRecoveryRPCServer) Watch(
+	_ *emptypb.Empty,
+	_ rpc.Controller_WatchServer,
+) error {
+	select {
+	case server.watchAttempts <- time.Now():
+	default:
+	}
+
+	return status.Error(codes.Unavailable, "RPC watch upstream unavailable")
+}
+
+func TestShouldPreserveRecoveredVM(t *testing.T) {
+	onDiskName := ondiskname.New("running-vm", recoveryTestVMUID, 0)
+	now := time.Unix(1_000, 0)
+	deadline := now.Add(recoveredVMProtectionPeriod)
+
+	t.Run("missing running VM stays protected", func(t *testing.T) {
+		recoveredVMs := map[ondiskname.OnDiskName]time.Time{onDiskName: deadline}
+
+		require.True(t, shouldPreserveRecoveredVM(recoveredVMs, onDiskName, nil,
+			mo.Some(v1.VMStatusRunning), now))
+		require.Contains(t, recoveredVMs, onDiskName)
+	})
+
+	t.Run("missing pending VM stays protected", func(t *testing.T) {
+		recoveredVMs := map[ondiskname.OnDiskName]time.Time{onDiskName: deadline}
+
+		require.True(t, shouldPreserveRecoveredVM(recoveredVMs, onDiskName, nil,
+			mo.Some(v1.VMStatusPending), now))
+		require.Contains(t, recoveredVMs, onDiskName)
+	})
+
+	t.Run("missing running VM loses protection after recovery deadline", func(t *testing.T) {
+		recoveredVMs := map[ondiskname.OnDiskName]time.Time{onDiskName: deadline}
+
+		require.False(t, shouldPreserveRecoveredVM(recoveredVMs, onDiskName, nil,
+			mo.Some(v1.VMStatusRunning), deadline))
+		require.NotContains(t, recoveredVMs, onDiskName)
+	})
+
+	t.Run("missing pending VM loses protection after recovery deadline", func(t *testing.T) {
+		recoveredVMs := map[ondiskname.OnDiskName]time.Time{onDiskName: deadline}
+
+		require.False(t, shouldPreserveRecoveredVM(recoveredVMs, onDiskName, nil,
+			mo.Some(v1.VMStatusPending), deadline))
+		require.NotContains(t, recoveredVMs, onDiskName)
+	})
+
+	t.Run("recognized VM returns to normal deletion behavior", func(t *testing.T) {
+		recoveredVMs := map[ondiskname.OnDiskName]time.Time{onDiskName: deadline}
+
+		require.False(t, shouldPreserveRecoveredVM(recoveredVMs, onDiskName,
+			&v1.VM{Status: v1.VMStatusRunning}, mo.Some(v1.VMStatusRunning), now))
+		require.NotContains(t, recoveredVMs, onDiskName)
+		require.False(t, shouldPreserveRecoveredVM(recoveredVMs, onDiskName, nil,
+			mo.Some(v1.VMStatusRunning), now))
+	})
+
+	t.Run("failed recovered VM is not protected", func(t *testing.T) {
+		recoveredVMs := map[ondiskname.OnDiskName]time.Time{onDiskName: deadline}
+
+		require.False(t, shouldPreserveRecoveredVM(recoveredVMs, onDiskName, nil,
+			mo.Some(v1.VMStatusFailed), now))
+		require.NotContains(t, recoveredVMs, onDiskName)
+	})
+
+	t.Run("new VM is not protected", func(t *testing.T) {
+		recoveredVMs := map[ondiskname.OnDiskName]time.Time{}
+
+		require.False(t, shouldPreserveRecoveredVM(recoveredVMs, onDiskName, nil,
+			mo.Some(v1.VMStatusRunning), now))
+	})
+}
+
+func TestTrackRecoveredVMsPreservesDeadlinesAcrossSessions(t *testing.T) {
+	firstVMResource := v1.VM{
+		Meta:   v1.Meta{Name: "first-running-vm"},
+		UID:    "first-running-vm-uid",
+		Worker: recoveryTestWorkerName,
+		Status: v1.VMStatusRunning,
+	}
+	secondVMResource := v1.VM{
+		Meta:   v1.Meta{Name: "second-pending-vm"},
+		UID:    "second-pending-vm-uid",
+		Worker: recoveryTestWorkerName,
+		Status: v1.VMStatusPending,
+	}
+
+	worker := &Worker{vmm: vmmanager.New()}
+	firstOnDiskName := ondiskname.NewFromResource(firstVMResource)
+	worker.vmm.Put(firstOnDiskName, &recoveryTestVM{resource: firstVMResource})
+
+	firstSession := time.Unix(1_000, 0)
+	firstDeadline := firstSession.Add(recoveredVMProtectionPeriod)
+	require.Equal(t, firstDeadline, worker.trackRecoveredVMs(firstSession)[firstOnDiskName])
+
+	secondSession := firstSession.Add(20 * time.Second)
+	secondOnDiskName := ondiskname.NewFromResource(secondVMResource)
+	worker.vmm.Put(secondOnDiskName, &recoveryTestVM{resource: secondVMResource})
+	recoveredVMs := worker.trackRecoveredVMs(secondSession)
+
+	require.Equal(t, firstDeadline, recoveredVMs[firstOnDiskName],
+		"controller reconnects must not extend an existing VM's protection")
+	require.Equal(t, secondSession.Add(recoveredVMProtectionPeriod), recoveredVMs[secondOnDiskName],
+		"newly recovered pending VMs receive their own bounded protection window")
+
+	worker.vmm.Delete(secondOnDiskName)
+	require.NotContains(t, worker.trackRecoveredVMs(secondSession.Add(time.Second)), secondOnDiskName)
+}
+
+func TestSyncVMsDeletesRecoveredVMAfterProtectionExpires(t *testing.T) {
+	controller := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		if request.URL.Path != recoveryTestVMsPath {
+			http.NotFound(writer, request)
+			return
+		}
+
+		writeRecoveryTestJSON(t, writer, []v1.VM{})
+	}))
+	t.Cleanup(controller.Close)
+
+	controllerClient, err := client.New(client.WithAddress(controller.URL))
+	require.NoError(t, err)
+
+	worker, err := New(controllerClient, WithName(recoveryTestWorkerName), WithSynthetic(), WithLogger(zap.NewNop()))
+	require.NoError(t, err)
+	t.Cleanup(worker.pollTicker.Stop)
+
+	vmResource := v1.VM{
+		Meta:   v1.Meta{Name: "deleted-vm"},
+		UID:    "deleted-vm-uid",
+		Worker: recoveryTestWorkerName,
+		Status: v1.VMStatusRunning,
+	}
+	deletedVM := &recoveryTestVM{
+		resource:       vmResource,
+		conditionsSeen: make(chan struct{}),
+	}
+	onDiskName := ondiskname.NewFromResource(vmResource)
+	worker.vmm.Put(onDiskName, deletedVM)
+
+	recoveredVMs := map[ondiskname.OnDiskName]time.Time{
+		onDiskName: time.Now().Add(-time.Second),
+	}
+	err = worker.syncVMs(context.Background(), func(context.Context, v1.VM) error {
+		return nil
+	}, recoveredVMs)
+
+	require.NoError(t, err)
+	require.True(t, deletedVM.stopped.Load(), "expired recovery protection must not prevent VM shutdown")
+	require.True(t, deletedVM.deleted.Load(), "expired recovery protection must not prevent VM deletion")
+	require.False(t, worker.vmm.Exists(onDiskName))
+	require.NotContains(t, recoveredVMs, onDiskName)
+}
+
+func TestSyncVMsDefersNewVMWhileRecoveredVMIsUnaccounted(t *testing.T) {
+	for _, existingStatus := range []v1.VMStatus{v1.VMStatusPending, v1.VMStatusRunning} {
+		t.Run(string(existingStatus), func(t *testing.T) {
+			existing := v1.VM{
+				Meta:      v1.Meta{Name: "missing-vm"},
+				UID:       "missing-vm-uid",
+				Worker:    recoveryTestWorkerName,
+				Status:    existingStatus,
+				Resources: v1.Resources{v1.ResourceTartVMs: 1},
+			}
+			replacement := v1.VM{
+				Meta:      v1.Meta{Name: "replacement-vm"},
+				UID:       "replacement-vm-uid",
+				Worker:    recoveryTestWorkerName,
+				Status:    v1.VMStatusPending,
+				Resources: v1.Resources{v1.ResourceTartVMs: 1},
+			}
+
+			controller := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+				if request.URL.Path != recoveryTestVMsPath {
+					http.NotFound(writer, request)
+					return
+				}
+
+				writeRecoveryTestJSON(t, writer, []v1.VM{replacement})
+			}))
+			t.Cleanup(controller.Close)
+
+			controllerClient, err := client.New(client.WithAddress(controller.URL))
+			require.NoError(t, err)
+			worker, err := New(controllerClient,
+				WithName(recoveryTestWorkerName),
+				WithSynthetic(),
+				WithResources(v1.Resources{v1.ResourceTartVMs: 1}),
+				WithLogger(zap.NewNop()),
+			)
+			require.NoError(t, err)
+			t.Cleanup(worker.pollTicker.Stop)
+
+			existingOnDiskName := ondiskname.NewFromResource(existing)
+			existingVM := &recoveryTestVM{
+				resource:       existing,
+				conditionsSeen: make(chan struct{}),
+			}
+			worker.vmm.Put(existingOnDiskName, existingVM)
+			recoveredVMs := map[ondiskname.OnDiskName]time.Time{
+				existingOnDiskName: time.Now().Add(recoveredVMProtectionPeriod),
+			}
+			updateVM := func(context.Context, v1.VM) error { return nil }
+
+			require.NoError(t, worker.syncVMs(context.Background(), updateVM, recoveredVMs))
+			require.True(t, worker.vmm.Exists(existingOnDiskName))
+			require.False(t, worker.vmm.Exists(ondiskname.NewFromResource(replacement)),
+				"a one-slot worker must not start another VM while an unaccounted VM is active")
+			require.Len(t, worker.vmm.List(), 1)
+
+			recoveredVMs[existingOnDiskName] = time.Now().Add(-time.Second)
+			require.NoError(t, worker.syncVMs(context.Background(), updateVM, recoveredVMs))
+			require.True(t, existingVM.stopped.Load())
+			require.True(t, existingVM.deleted.Load())
+			require.True(t, worker.vmm.Exists(ondiskname.NewFromResource(replacement)),
+				"the queued VM should start after the expired recovered VM is removed")
+			t.Cleanup(func() { require.NoError(t, worker.Close()) })
+		})
+	}
+}
+
+func TestWatchRPCV2PreservesActiveOperationsAfterWatchCloses(t *testing.T) {
+	operationStarted := make(chan struct{})
+	releaseOperation := make(chan struct{})
+	resolvedIP := make(chan string, 1)
+
+	controller := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		switch request.URL.Path {
+		case recoveryTestWatchPath:
+			connection, err := websocket.Accept(writer, request, nil)
+			if err != nil {
+				t.Errorf("failed to accept RPC watch: %v", err)
+				return
+			}
+			defer connection.CloseNow()
+
+			instruction := v1.WatchInstruction{ResolveIPAction: &v1.ResolveIPAction{
+				Session: "ip-session",
+				VMUID:   recoveryTestVMUID,
+			}}
+			payload, err := json.Marshal(instruction)
+			if err != nil {
+				t.Errorf("failed to encode RPC instruction: %v", err)
+				return
+			}
+			payload = append(payload, '\n')
+
+			if err := connection.Write(request.Context(), websocket.MessageBinary, payload); err != nil {
+				t.Errorf("failed to send RPC instruction: %v", err)
+				return
+			}
+
+			<-request.Context().Done()
+		case "/v1/rpc/resolve-ip":
+			resolvedIP <- request.URL.Query().Get("ip")
+			writer.WriteHeader(http.StatusOK)
+		default:
+			http.NotFound(writer, request)
+		}
+	}))
+	t.Cleanup(controller.Close)
+
+	controllerClient, err := client.New(client.WithAddress(controller.URL))
+	require.NoError(t, err)
+
+	worker, err := New(controllerClient, WithName(recoveryTestWorkerName), WithSynthetic(), WithLogger(zap.NewNop()))
+	require.NoError(t, err)
+	t.Cleanup(worker.pollTicker.Stop)
+
+	vmResource := v1.VM{Meta: v1.Meta{Name: "running-vm"}, UID: recoveryTestVMUID}
+	worker.vmm.Put(ondiskname.NewFromResource(vmResource), &recoveryIPTestVM{
+		recoveryTestVM: recoveryTestVM{resource: vmResource},
+		started:        operationStarted,
+		release:        releaseOperation,
+	})
+
+	operationCtx, cancelOperation := context.WithCancel(context.Background())
+	t.Cleanup(cancelOperation)
+	watchCtx, cancelWatch := context.WithCancel(operationCtx)
+	t.Cleanup(cancelWatch)
+
+	watchResult := make(chan error, 1)
+	go func() {
+		watchResult <- worker.watchRPCV2(watchCtx, operationCtx, func() {})
+	}()
+
+	select {
+	case <-operationStarted:
+	case err := <-watchResult:
+		t.Fatalf("RPC watch terminated before its operation started: %v", err)
+	case <-time.After(2 * time.Second):
+		t.Fatal("RPC operation did not start")
+	}
+
+	cancelWatch()
+	require.ErrorIs(t, <-watchResult, context.Canceled)
+	close(releaseOperation)
+
+	select {
+	case ip := <-resolvedIP:
+		require.Equal(t, "192.0.2.10", ip)
+	case <-time.After(2 * time.Second):
+		t.Fatal("active RPC operation was canceled with its watch session")
+	}
+}
+
+type recoveryTestVM struct {
+	vmmanager.VM
+
+	resource       v1.VM
+	conditionsSeen chan struct{}
+	conditionsOnce sync.Once
+	stopped        atomic.Bool
+	deleted        atomic.Bool
+}
+
+type recoveryIPTestVM struct {
+	recoveryTestVM
+
+	started chan struct{}
+	release chan struct{}
+}
+
+func (vm *recoveryIPTestVM) IP(ctx context.Context) (string, error) {
+	close(vm.started)
+
+	select {
+	case <-vm.release:
+		return "192.0.2.10", nil
+	case <-ctx.Done():
+		return "", ctx.Err()
+	}
+}
+
+func (vm *recoveryTestVM) Resource() v1.VM {
+	return vm.resource
+}
+
+func (vm *recoveryTestVM) OnDiskName() ondiskname.OnDiskName {
+	return ondiskname.NewFromResource(vm.resource)
+}
+
+func (vm *recoveryTestVM) Status() v1.VMStatus {
+	return vm.resource.Status
+}
+
+func (vm *recoveryTestVM) Conditions() []v1.Condition {
+	vm.conditionsOnce.Do(func() { close(vm.conditionsSeen) })
+
+	return nil
+}
+
+func (vm *recoveryTestVM) Stop() <-chan error {
+	vm.stopped.Store(true)
+
+	result := make(chan error, 1)
+	result <- nil
+
+	return result
+}
+
+func (vm *recoveryTestVM) Delete() error {
+	vm.deleted.Store(true)
+
+	return nil
+}
+
+func writeRecoveryTestJSON(t *testing.T, writer http.ResponseWriter, value any) {
+	t.Helper()
+
+	writer.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(writer).Encode(value); err != nil {
+		t.Errorf("failed to encode controller response: %v", err)
+	}
+}
+
 func TestSortNonExistentAndFailedFirst(t *testing.T) {
 	newVMTuple := func(name string, vmResource *v1.VM) lo.Tuple3[ondiskname.OnDiskName, *v1.VM, vmmanager.VM] {
 		return lo.T3[ondiskname.OnDiskName, *v1.VM, vmmanager.VM](

pkg/resource/v1/resources.go

@@ -62,16 +62,18 @@ func (resources Resources) Added(other Resources) Resources {
 
 func (resources Resources) Subtract(other Resources) {
 	for otherKey, otherValue := range other {
+		if otherValue >= resources[otherKey] {
+			resources[otherKey] = 0
+			continue
+		}
+
 		resources[otherKey] -= otherValue
 	}
 }
 
 func (resources Resources) Subtracted(other Resources) Resources {
 	result := resources.Copy()
-
+	result.Subtract(other)
-	for otherKey, otherValue := range other {
-		result[otherKey] -= otherValue
-	}
 
 	return result
 }

pkg/resource/v1/resources_test.go

@@ -55,6 +55,50 @@ func TestResourcesSubtracted(t *testing.T) {
 	}))
 }
 
+func TestResourcesSubtractSaturatesAtZero(t *testing.T) {
+	resources := v1.Resources{
+		v1.ResourceTartVMs:      2,
+		v1.ResourceLogicalCores: 8,
+	}
+
+	resources.Subtract(v1.Resources{
+		v1.ResourceTartVMs:      3,
+		v1.ResourceLogicalCores: 4,
+		v1.ResourceMemoryMiB:    1,
+	})
+
+	require.Equal(t, v1.Resources{
+		v1.ResourceTartVMs:      0,
+		v1.ResourceLogicalCores: 4,
+		v1.ResourceMemoryMiB:    0,
+	}, resources)
+	require.False(t, resources.CanFit(v1.Resources{v1.ResourceTartVMs: 1}))
+}
+
+func TestResourcesSubtractedSaturatesAtZeroWithoutMutatingOriginal(t *testing.T) {
+	resources := v1.Resources{
+		v1.ResourceTartVMs:      2,
+		v1.ResourceLogicalCores: 8,
+	}
+
+	remaining := resources.Subtracted(v1.Resources{
+		v1.ResourceTartVMs:      3,
+		v1.ResourceLogicalCores: 4,
+		v1.ResourceMemoryMiB:    1,
+	})
+
+	require.Equal(t, v1.Resources{
+		v1.ResourceTartVMs:      0,
+		v1.ResourceLogicalCores: 4,
+		v1.ResourceMemoryMiB:    0,
+	}, remaining)
+	require.Equal(t, v1.Resources{
+		v1.ResourceTartVMs:      2,
+		v1.ResourceLogicalCores: 8,
+	}, resources)
+	require.False(t, remaining.CanFit(v1.Resources{v1.ResourceTartVMs: 1}))
+}
+
 func TestResourcesCanFit(t *testing.T) {
 	resources := v1.Resources{
 		v1.ResourceTartVMs: 2,