Add standby compression start delay

cmd/api/api/instances.go

@@ -542,6 +542,16 @@ func (s *ApiService) StandbyInstance(ctx context.Context, request oapi.StandbyIn
 		}
 		standbyReq.Compression = compression
 	}
+	if request.Body != nil && request.Body.CompressionDelay != nil {
+		compressionDelay, err := parseOptionalDuration(*request.Body.CompressionDelay, "compression_delay")
+		if err != nil {
+			return oapi.StandbyInstance400JSONResponse{
+				Code:    "invalid_compression_delay",
+				Message: err.Error(),
+			}, nil
+		}
+		standbyReq.CompressionDelay = compressionDelay
+	}
 
 	result, err := s.InstanceManager.StandbyInstance(ctx, inst.Id, standbyReq)
 	if err != nil {
@@ -1152,6 +1162,13 @@ func toInstanceSnapshotPolicy(policy oapi.SnapshotPolicy) (*instances.SnapshotPo
 		}
 		out.Compression = compression
 	}
+	if policy.StandbyCompressionDelay != nil {
+		delay, err := parseOptionalDuration(*policy.StandbyCompressionDelay, "standby_compression_delay")
+		if err != nil {
+			return nil, err
+		}
+		out.StandbyCompressionDelay = delay
+	}
 	return out, nil
 }
 
@@ -1176,5 +1193,20 @@ func toOAPISnapshotPolicy(policy instances.SnapshotPolicy) oapi.SnapshotPolicy {
 		compression := toOAPISnapshotCompressionConfig(*policy.Compression)
 		out.Compression = &compression
 	}
+	if policy.StandbyCompressionDelay != nil {
+		delay := policy.StandbyCompressionDelay.String()
+		out.StandbyCompressionDelay = &delay
+	}
 	return out
 }
+
+func parseOptionalDuration(value string, field string) (*time.Duration, error) {
+	duration, err := time.ParseDuration(value)
+	if err != nil {
+		return nil, fmt.Errorf("%s must be a valid duration: %w", field, err)
+	}
+	if duration < 0 {
+		return nil, fmt.Errorf("%s cannot be negative", field)
+	}
+	return &duration, nil
+}

cmd/api/api/instances_test.go

@@ -480,10 +480,85 @@ func TestCreateInstance_MapsNetworkEgressEnforcementMode(t *testing.T) {
 	assert.Equal(t, instances.EgressEnforcementModeHTTPHTTPSOnly, mockMgr.lastReq.NetworkEgress.EnforcementMode)
 }
 
-func TestCreateInstance_MapsAutoStandbyPolicy(t *testing.T) {
+func TestCreateInstance_MapsStandbyCompressionDelayInSnapshotPolicy(t *testing.T) {
+	t.Parallel()
+
+	svc := newTestService(t)
+	origMgr := svc.InstanceManager
+	mockMgr := &captureCreateManager{Manager: origMgr}
+	svc.InstanceManager = mockMgr
+
+	delay := "2m30s"
+	resp, err := svc.CreateInstance(ctx(), oapi.CreateInstanceRequestObject{
+		Body: &oapi.CreateInstanceRequest{
+			Name:  "test-standby-compression-delay",
+			Image: "docker.io/library/alpine:latest",
+			SnapshotPolicy: &oapi.SnapshotPolicy{
+				StandbyCompressionDelay: &delay,
+			},
+		},
+	})
+	require.NoError(t, err)
+	_, ok := resp.(oapi.CreateInstance201JSONResponse)
+	require.True(t, ok, "expected 201 response")
+
+	require.NotNil(t, mockMgr.lastReq)
+	require.NotNil(t, mockMgr.lastReq.SnapshotPolicy)
+	require.NotNil(t, mockMgr.lastReq.SnapshotPolicy.StandbyCompressionDelay)
+	assert.Equal(t, 150*time.Second, *mockMgr.lastReq.SnapshotPolicy.StandbyCompressionDelay)
+}
+
+func TestCreateInstance_InvalidStandbyCompressionDelayInSnapshotPolicy(t *testing.T) {
 	t.Parallel()
+
 	svc := newTestService(t)
+	delay := "not-a-duration"
 
+	resp, err := svc.CreateInstance(ctx(), oapi.CreateInstanceRequestObject{
+		Body: &oapi.CreateInstanceRequest{
+			Name:  "test-invalid-standby-delay",
+			Image: "docker.io/library/alpine:latest",
+			SnapshotPolicy: &oapi.SnapshotPolicy{
+				StandbyCompressionDelay: &delay,
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	badReq, ok := resp.(oapi.CreateInstance400JSONResponse)
+	require.True(t, ok, "expected 400 response")
+	assert.Equal(t, "invalid_snapshot_policy", badReq.Code)
+	assert.Contains(t, badReq.Message, "standby_compression_delay")
+}
+
+func TestInstanceToOAPI_EmitsStandbyCompressionDelayInSnapshotPolicy(t *testing.T) {
+	t.Parallel()
+
+	delay := 90 * time.Second
+	inst := instances.Instance{
+		StoredMetadata: instances.StoredMetadata{
+			Id:             "inst-standby-delay",
+			Name:           "inst-standby-delay",
+			Image:          "docker.io/library/alpine:latest",
+			CreatedAt:      time.Now(),
+			HypervisorType: hypervisor.TypeCloudHypervisor,
+			SnapshotPolicy: &instances.SnapshotPolicy{
+				StandbyCompressionDelay: &delay,
+			},
+		},
+		State: instances.StateStandby,
+	}
+
+	oapiInst := instanceToOAPI(inst)
+	require.NotNil(t, oapiInst.SnapshotPolicy)
+	require.NotNil(t, oapiInst.SnapshotPolicy.StandbyCompressionDelay)
+	assert.Equal(t, "1m30s", *oapiInst.SnapshotPolicy.StandbyCompressionDelay)
+}
+
+func TestCreateInstance_MapsAutoStandbyPolicy(t *testing.T) {
+	t.Parallel()
+
+	svc := newTestService(t)
 	origMgr := svc.InstanceManager
 	mockMgr := &captureCreateManager{Manager: origMgr}
 	svc.InstanceManager = mockMgr
@@ -907,6 +982,81 @@ func TestStandbyInstance_InvalidRequest(t *testing.T) {
 	assert.Contains(t, badReq.Message, "invalid snapshot compression level")
 }
 
+func TestStandbyInstance_MapsCompressionDelay(t *testing.T) {
+	t.Parallel()
+
+	svc := newTestService(t)
+	now := time.Now()
+	source := instances.Instance{
+		StoredMetadata: instances.StoredMetadata{
+			Id:             "standby-delay-src",
+			Name:           "standby-delay-src",
+			Image:          "docker.io/library/alpine:latest",
+			CreatedAt:      now,
+			HypervisorType: hypervisor.TypeCloudHypervisor,
+		},
+		State: instances.StateRunning,
+	}
+
+	mockMgr := &captureStandbyManager{
+		Manager: svc.InstanceManager,
+		result:  &source,
+	}
+	svc.InstanceManager = mockMgr
+
+	delay := "45s"
+	resp, err := svc.StandbyInstance(
+		mw.WithResolvedInstance(ctx(), source.Id, source),
+		oapi.StandbyInstanceRequestObject{
+			Id: source.Id,
+			Body: &oapi.StandbyInstanceRequest{
+				CompressionDelay: &delay,
+			},
+		},
+	)
+	require.NoError(t, err)
+	_, ok := resp.(oapi.StandbyInstance200JSONResponse)
+	require.True(t, ok, "expected 200 response")
+
+	require.NotNil(t, mockMgr.lastReq)
+	require.NotNil(t, mockMgr.lastReq.CompressionDelay)
+	assert.Equal(t, 45*time.Second, *mockMgr.lastReq.CompressionDelay)
+}
+
+func TestStandbyInstance_InvalidCompressionDelay(t *testing.T) {
+	t.Parallel()
+
+	svc := newTestService(t)
+	now := time.Now()
+	source := instances.Instance{
+		StoredMetadata: instances.StoredMetadata{
+			Id:             "standby-invalid-delay-src",
+			Name:           "standby-invalid-delay-src",
+			Image:          "docker.io/library/alpine:latest",
+			CreatedAt:      now,
+			HypervisorType: hypervisor.TypeCloudHypervisor,
+		},
+		State: instances.StateRunning,
+	}
+
+	delay := "-5s"
+	resp, err := svc.StandbyInstance(
+		mw.WithResolvedInstance(ctx(), source.Id, source),
+		oapi.StandbyInstanceRequestObject{
+			Id: source.Id,
+			Body: &oapi.StandbyInstanceRequest{
+				CompressionDelay: &delay,
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	badReq, ok := resp.(oapi.StandbyInstance400JSONResponse)
+	require.True(t, ok, "expected 400 response")
+	assert.Equal(t, "invalid_compression_delay", badReq.Code)
+	assert.Contains(t, badReq.Message, "compression_delay")
+}
+
 func TestForkInstance_FromRunningFlagForwarded(t *testing.T) {
 	t.Parallel()
 	svc := newTestService(t)

lib/instances/compression_integration_linux_test.go

@@ -264,25 +264,46 @@ func waitForRunningAndExecReady(t *testing.T, ctx context.Context, mgr *manager,
 
 func writeGuestMarker(t *testing.T, ctx context.Context, inst *Instance, path string, value string) {
 	t.Helper()
-	execCtx, cancel := context.WithTimeout(ctx, integrationTestTimeout(compressionGuestExecTimeout))
-	defer cancel()
-
-	output, exitCode, err := execCommand(execCtx, inst, "sh", "-c", fmt.Sprintf("printf %q > %s && sync", value, path))
+	output, exitCode, err := execCommandWithRetry(ctx, inst, compressionGuestExecTimeout, "sh", "-c", fmt.Sprintf("printf %q > %s && sync", value, path))
 	require.NoError(t, err)
 	require.Equal(t, 0, exitCode, output)
 }
 
 func assertGuestMarker(t *testing.T, ctx context.Context, inst *Instance, path string, expected string) {
 	t.Helper()
-	execCtx, cancel := context.WithTimeout(ctx, integrationTestTimeout(compressionGuestExecTimeout))
-	defer cancel()
-
-	output, exitCode, err := execCommand(execCtx, inst, "cat", path)
+	output, exitCode, err := execCommandWithRetry(ctx, inst, compressionGuestExecTimeout, "cat", path)
 	require.NoError(t, err)
 	require.Equal(t, 0, exitCode, output)
 	assert.Equal(t, expected, output)
 }
 
+func execCommandWithRetry(ctx context.Context, inst *Instance, timeout time.Duration, command ...string) (string, int, error) {
+	deadline := time.Now().Add(integrationTestTimeout(timeout))
+	var lastOutput string
+	var lastExitCode int
+	var lastErr error
+
+	for {
+		execCtx, cancel := context.WithTimeout(ctx, integrationTestTimeout(5*time.Second))
+		output, exitCode, err := execCommand(execCtx, inst, command...)
+		cancel()
+
+		if err == nil {
+			return output, exitCode, nil
+		}
+
+		lastOutput = output
+		lastExitCode = exitCode
+		lastErr = err
+
+		if time.Now().After(deadline) {
+			return lastOutput, lastExitCode, lastErr
+		}
+
+		time.Sleep(500 * time.Millisecond)
+	}
+}
+
 func waitForCompressionJobStart(t *testing.T, mgr *manager, key string, timeout time.Duration) {
 	t.Helper()
 	deadline := time.Now().Add(timeout)

lib/instances/create.go

@@ -599,6 +599,11 @@ func validateCreateRequest(req *CreateInstanceRequest) error {
 			return err
 		}
 	}
+	if req.SnapshotPolicy != nil && req.SnapshotPolicy.StandbyCompressionDelay != nil {
+		if _, err := normalizeStandbyCompressionDelay(req.SnapshotPolicy.StandbyCompressionDelay); err != nil {
+			return err
+		}
+	}
 	normalizedAutoStandby, err := normalizeAutoStandbyPolicy(req.AutoStandby)
 	if err != nil {
 		return err

lib/instances/delete.go

@@ -39,8 +39,8 @@ func (m *manager) deleteInstance(
 	if err != nil {
 		return fmt.Errorf("wait for instance compression to stop: %w", err)
 	}
-	if target != nil {
-		m.recordSnapshotCompressionPreemption(ctx, snapshotCompressionPreemptionDeleteInstance, *target)
+	if target != nil && target.State == compressionJobStateRunning {
+		m.recordSnapshotCompressionPreemption(ctx, snapshotCompressionPreemptionDeleteInstance, target.Target)
 	}
 
 	// 2. Get network allocation BEFORE killing VMM (while we can still query it)

lib/instances/fork.go

@@ -269,7 +269,7 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin
 	}
 
 	now := time.Now()
-	forkMeta := cloneStoredMetadata(meta.StoredMetadata)
+	forkMeta := cloneStoredMetadataWithoutPendingStandbyCompression(meta.StoredMetadata)
 	forkMeta.Id = forkID
 	forkMeta.Name = req.Name
 	forkMeta.CreatedAt = now
@@ -472,6 +472,7 @@ func (m *manager) cleanupForkInstanceOnError(ctx context.Context, forkID string)
 
 func cloneStoredMetadata(src StoredMetadata) StoredMetadata {
 	dst := src
+	dst.PendingStandbyCompression = clonePendingStandbyCompression(src.PendingStandbyCompression)
 
 	if src.Env != nil {
 		dst.Env = make(map[string]string, len(src.Env))
@@ -536,3 +537,9 @@ func cloneStoredMetadata(src StoredMetadata) StoredMetadata {
 
 	return dst
 }
+
+func cloneStoredMetadataWithoutPendingStandbyCompression(src StoredMetadata) StoredMetadata {
+	dst := cloneStoredMetadata(src)
+	dst.PendingStandbyCompression = nil
+	return dst
+}

lib/instances/fork_test.go

@@ -282,9 +282,11 @@ func TestCloneStoredMetadataForFork_DeepCopiesReferenceFields(t *testing.T) {
 	t.Parallel()
 	startedAt := time.Now().Add(-2 * time.Minute)
 	stoppedAt := time.Now().Add(-1 * time.Minute)
+	notBefore := time.Now().Add(5 * time.Minute)
 	pid := 1234
 	exitCode := 17
 	compressionLevel := 5
+	pendingLevel := 3
 
 	src := StoredMetadata{
 		Env:           map[string]string{"A": "1"},
@@ -310,6 +312,14 @@ func TestCloneStoredMetadataForFork_DeepCopiesReferenceFields(t *testing.T) {
 				Level:     &compressionLevel,
 			},
 		},
+		PendingStandbyCompression: &PendingStandbyCompression{
+			Policy: snapshotstore.SnapshotCompressionConfig{
+				Enabled:   true,
+				Algorithm: snapshotstore.SnapshotCompressionAlgorithmZstd,
+				Level:     &pendingLevel,
+			},
+			NotBefore: notBefore,
+		},
 	}
 
 	cloned := cloneStoredMetadata(src)
@@ -327,6 +337,8 @@ func TestCloneStoredMetadataForFork_DeepCopiesReferenceFields(t *testing.T) {
 	cloned.AutoStandby.IgnoreDestinationPorts[0] = 443
 	*cloned.SnapshotPolicy.Compression.Level = 9
 	now := time.Now()
+	*cloned.PendingStandbyCompression.Policy.Level = 1
+	cloned.PendingStandbyCompression.NotBefore = now
 	*cloned.StartedAt = now
 	*cloned.StoppedAt = now
 
@@ -341,10 +353,34 @@ func TestCloneStoredMetadataForFork_DeepCopiesReferenceFields(t *testing.T) {
 	require.Equal(t, "10.0.0.0/8", src.AutoStandby.IgnoreSourceCIDRs[0])
 	require.Equal(t, uint16(22), src.AutoStandby.IgnoreDestinationPorts[0])
 	require.Equal(t, 5, *src.SnapshotPolicy.Compression.Level)
+	require.NotNil(t, src.PendingStandbyCompression)
+	require.NotNil(t, src.PendingStandbyCompression.Policy.Level)
+	require.Equal(t, 3, *src.PendingStandbyCompression.Policy.Level)
+	require.Equal(t, notBefore, src.PendingStandbyCompression.NotBefore)
 	require.Equal(t, startedAt, *src.StartedAt)
 	require.Equal(t, stoppedAt, *src.StoppedAt)
 }
 
+func TestCloneStoredMetadataWithoutPendingStandbyCompression_ClearsPendingPlan(t *testing.T) {
+	t.Parallel()
+
+	level := 4
+	src := StoredMetadata{
+		PendingStandbyCompression: &PendingStandbyCompression{
+			Policy: snapshotstore.SnapshotCompressionConfig{
+				Enabled:   true,
+				Algorithm: snapshotstore.SnapshotCompressionAlgorithmZstd,
+				Level:     &level,
+			},
+			NotBefore: time.Now().Add(2 * time.Minute),
+		},
+	}
+
+	cloned := cloneStoredMetadataWithoutPendingStandbyCompression(src)
+	assert.Nil(t, cloned.PendingStandbyCompression)
+	require.NotNil(t, src.PendingStandbyCompression)
+}
+
 func TestRotateSourceVsockForRestore_CloudHypervisorDoesNotPersistCIDRewrite(t *testing.T) {
 	t.Parallel()
 	manager, _ := setupTestManager(t)