Defer in-place updates while VMs are restarting (#483)

internal/worker/vmmanager/base/base.go

@@ -122,6 +122,15 @@ func (vm *VM) ConditionsSet() mapset.Set[v1.ConditionType] {
 	return vm.conditions
 }
 
+// Running reports whether the VM is running and neither stopping nor suspending.
+func (vm *VM) Running() bool {
+	// Snapshot the flags together so shutdown cannot mix old and new conditions
+	conditions := vm.conditions.Clone()
+
+	return conditions.ContainsOne(v1.ConditionTypeRunning) &&
+		!conditions.ContainsAny(v1.ConditionTypeStopping, v1.ConditionTypeSuspending)
+}
+
 func (vm *VM) Conditions() []v1.Condition {
 	// The worker must observe transitions before applying a new specification.
 	return []v1.Condition{

internal/worker/vmmanager/vmmanager.go

@@ -24,6 +24,7 @@ type VM interface {
 	StatusMessage() string
 	Err() error
 	Conditions() []v1.Condition
+	Running() bool
 
 	Start(eventStreamer *client.EventStreamer)
 	Suspend() <-chan error

internal/worker/worker.go

@@ -601,6 +601,7 @@ func (worker *Worker) syncVMs(
 			currentVMResource := vm.Resource()
 
 			if worker.softnetPolicyUpdates.OrElse(false) &&
+				vm.Running() &&
 				currentVMResource.SoftnetEnabled() && vmResource.SoftnetEnabled() &&
 				currentVMResource.SoftnetPolicyChanged(vmResource.VMSpec) {
 				if err := vm.UpdateSoftnetPolicy(ctx,
@@ -687,10 +688,11 @@ func (worker *Worker) monitorRunningVM(
 	// Try to apply the host process updates in-place or recover processes that exited unexpectedly
 	hostProcessesChanged := !currentVMResource.VMSpec.HostProcessesEqual(vmResource.VMSpec)
 	hostProcessesNeedRestart := currentVMResource.Generation == vmResource.Generation &&
-		v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) &&
 		!vm.HostProcessSet().Ready()
 
-	if vmResource.PowerState == v1.PowerStateRunning && (hostProcessesChanged || hostProcessesNeedRestart) {
+	if vmResource.PowerState == v1.PowerStateRunning &&
+		vm.Running() &&
+		(hostProcessesChanged || hostProcessesNeedRestart) {
 		if err := vm.HostProcessSet().Replace(ctx, vmResource.HostProcesses); err != nil {
 			worker.logger.Warnf("failed to update host processes for VM %q: %v",
 				vmResource.Name, err)
@@ -709,9 +711,7 @@ func (worker *Worker) monitorRunningVM(
 	)
 
 	if vmResource.PowerState == v1.PowerStateRunning &&
-		!v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeStopping) &&
-		!v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeSuspending) &&
-		v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) &&
+		vm.Running() &&
 		endpointsChanged {
 		currentVMResource.Endpoints = vmResource.Endpoints
 		appliedInPlace = true

internal/worker/worker_stop_test.go

@@ -41,6 +41,12 @@ func (vm *delayedStopVM) Status() v1.VMStatus { return vm.status }
 
 func (vm *delayedStopVM) Conditions() []v1.Condition { return vm.conditions }
 
+func (vm *delayedStopVM) Running() bool {
+	return v1.ConditionIsTrue(vm.conditions, v1.ConditionTypeRunning) &&
+		!v1.ConditionIsTrue(vm.conditions, v1.ConditionTypeStopping) &&
+		!v1.ConditionIsTrue(vm.conditions, v1.ConditionTypeSuspending)
+}
+
 func (vm *delayedStopVM) Resource() v1.VM { return vm.resource }
 
 func (vm *delayedStopVM) SetResource(resource v1.VM) {
@@ -130,6 +136,57 @@ func TestMonitorWaitsForStopBeforeApplyingGeneration(t *testing.T) {
 	}
 }
 
+func TestMonitorPreservesRestartAfterHostProcessUpdate(t *testing.T) {
+	events := make(chan struct{}, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) {
+		response.WriteHeader(http.StatusOK)
+		events <- struct{}{}
+	}))
+	defer server.Close()
+	apiClient, err := client.New(client.WithAddress(server.URL))
+	require.NoError(t, err)
+	worker := &Worker{client: apiClient, runtime: runtime.NewSynthetic()}
+	update := func(context.Context, v1.VM) error { return nil }
+
+	// An earlier specification change has already initiated a restart
+	vm := &delayedStopVM{
+		VM: &synthetic.VM{VM: base.NewVM(v1.VM{}, ondiskname.OnDiskName{}, zap.NewNop().Sugar())},
+		resource: v1.VM{
+			Name:          "test-vm",
+			PowerState:    v1.PowerStateRunning,
+			HostProcesses: []v1.HostProcess{{Name: "removed", Program: "unused"}},
+		},
+		conditions: []v1.Condition{
+			{Type: v1.ConditionTypeRunning, State: v1.ConditionStateFalse},
+			{Type: v1.ConditionTypeStopping, State: v1.ConditionStateTrue},
+		},
+	}
+	desired := vm.resource
+	desired.Generation = 2
+	desired.HostProcesses = nil
+
+	// Removing the host process must leave the generation pending while stopping
+	require.NoError(t, worker.monitorRunningVM(t.Context(), &desired, vm, update))
+	require.Zero(t, vm.resource.Generation)
+	require.Zero(t, desired.ObservedGeneration)
+	require.Zero(t, vm.starts)
+
+	// After stopping, the pending generation must still cause the VM to start
+	v1.ConditionsSet(&vm.conditions, v1.Condition{
+		Type: v1.ConditionTypeStopping, State: v1.ConditionStateFalse,
+	})
+	require.NoError(t, worker.monitorRunningVM(t.Context(), &desired, vm, update))
+	require.Equal(t, desired.Generation, vm.resource.Generation)
+	require.Equal(t, desired.Generation, desired.ObservedGeneration)
+	require.Equal(t, 1, vm.starts)
+	require.Empty(t, vm.resource.HostProcesses)
+	select {
+	case <-events:
+	case <-time.After(time.Second):
+		t.Fatal("restart event stream did not close")
+	}
+}
+
 func TestSyncVMsWaitsForVMShutdown(t *testing.T) {
 	tests := []struct {
 		name         string

internal/worker/worker_test.go

@@ -927,6 +927,8 @@ func (vm *recoveryTestVM) Conditions() []v1.Condition {
 	return nil
 }
 
+func (vm *recoveryTestVM) Running() bool { return false }
+
 func (vm *recoveryTestVM) Stop() <-chan error {
 	vm.stopped.Store(true)