Add optional snapshot compression defaults and standby integration (#149)

* Add configurable snapshot compression with async standby support

* Add CH snapshot compression restore coverage

* Skip transient compression temp files during snapshot copy

* Clarify async snapshot compression restore behavior

* Reduce snapshot compression test races

* Restore compression test parallelism

* Unify snapshot restore cancellation behavior

* Apply suggestions from code review

Co-authored-by: Steven Miller <sjmiller609@gmail.com>

* Add snapshot compression metrics

* Fix snapshot compression review feedback

* Update Stainless model config

* Serialize shared initrd rebuilds

* Fix disabled snapshot defaults fallback

* Make snapshot compression fully opt-in

* Fix snapshot compression restore races

* Fix optional standby body handling

* Fix standby snapshot compression races

* Normalize standby snapshot compression copies

* Handle optional standby bodies outside generated code

* Fix snapshot compression cleanup races

* Clarify snapshot compression metrics state

* Return bad request for invalid standby input

* Use native-first snapshot codecs with Go fallback

* Normalize snapshot compression algorithms case-insensitively

* Tighten standby compression validation handling

* Reduce compression levels in integration tests to avoid CI timeout

The compression integration tests were using zstd level 19 and lz4 level 9,
which are very slow for compressing ~1GB memory files. After merging main
(which added more integration tests to lib/instances), the total package
test time exceeded the 20-minute CI timeout.

Reduce to level 3 for both zstd and lz4 high-level cases. The tests still
exercise the full compression/decompression pipeline across both algorithms
and multiple levels (1 and 3 for zstd, 0 and 3 for lz4).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Reduce compression integration test cycles to fit CI timeout

Five compression cycles (each involving VM standby + compress + restore +
boot + exec readiness) consistently exceed the 20-minute CI timeout after
merging main. Reduce to three cycles: one in-flight zstd, one completed
zstd, and one completed lz4. This still exercises both algorithms and
both the in-flight/completed code paths.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Address PR review feedback: add OpenAPI descriptions, fix dst.Close() error handling

- Add descriptions to snapshot_policy and compression fields in openapi.yaml
  per Steven's review comments
- Check dst.Close() errors in runGoCompression and runGoDecompression to
  prevent silently corrupt snapshot files on delayed write failures

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Address review feedback: add server-side compression validation, log metadata errors

- Validate algorithm (zstd/lz4) and per-algorithm level ranges in
  toDomainSnapshotCompressionConfig instead of passing through unchecked
- Log metadata update errors in compression jobs instead of silently
  discarding them
- Normalize algorithm to lowercase in config struct after validation
- Fix misleading test name (OmitsLevel -> PreservesLevel)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Fix snapshot compression review follow-ups

* Fix standby fork compression race

---------

Co-authored-by: sjmiller609 <7516283+sjmiller609@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: sjmiller609

cmd/api/api/api_test.go

@@ -43,7 +43,7 @@ func newTestService(t *testing.T) *ApiService {
 	limits := instances.ResourceLimits{
 		MaxOverlaySize: 100 * 1024 * 1024 * 1024, // 100GB
 	}
-	instanceMgr := instances.NewManager(p, imageMgr, systemMgr, networkMgr, deviceMgr, volumeMgr, limits, "", nil, nil)
+	instanceMgr := instances.NewManager(p, imageMgr, systemMgr, networkMgr, deviceMgr, volumeMgr, limits, "", instances.SnapshotPolicy{}, nil, nil)
 
 	// Initialize network manager (creates bridge for network-enabled tests)
 	if err := networkMgr.Initialize(ctx(), nil); err != nil {

cmd/api/api/instances.go

@@ -17,6 +17,7 @@ import (
 	"github.com/kernel/hypeman/lib/network"
 	"github.com/kernel/hypeman/lib/oapi"
 	"github.com/kernel/hypeman/lib/resources"
+	"github.com/kernel/hypeman/lib/snapshot"
 	"github.com/kernel/hypeman/lib/vm_metrics"
 	"github.com/samber/lo"
 )
@@ -301,6 +302,16 @@ func (s *ApiService) CreateInstance(ctx context.Context, request oapi.CreateInst
 		SkipKernelHeaders:        request.Body.SkipKernelHeaders != nil && *request.Body.SkipKernelHeaders,
 		SkipGuestAgent:           request.Body.SkipGuestAgent != nil && *request.Body.SkipGuestAgent,
 	}
+	if request.Body.SnapshotPolicy != nil {
+		snapshotPolicy, err := toInstanceSnapshotPolicy(*request.Body.SnapshotPolicy)
+		if err != nil {
+			return oapi.CreateInstance400JSONResponse{
+				Code:    "invalid_snapshot_policy",
+				Message: err.Error(),
+			}, nil
+		}
+		domainReq.SnapshotPolicy = snapshotPolicy
+	}
 
 	inst, err := s.InstanceManager.CreateInstance(ctx, domainReq)
 	if err != nil {
@@ -438,9 +449,26 @@ func (s *ApiService) StandbyInstance(ctx context.Context, request oapi.StandbyIn
 	}
 	log := logger.FromContext(ctx)
 
-	result, err := s.InstanceManager.StandbyInstance(ctx, inst.Id)
+	standbyReq := instances.StandbyInstanceRequest{}
+	if request.Body != nil && request.Body.Compression != nil {
+		compression, err := toDomainSnapshotCompressionConfig(*request.Body.Compression)
+		if err != nil {
+			return oapi.StandbyInstance400JSONResponse{
+				Code:    "invalid_snapshot_compression",
+				Message: err.Error(),
+			}, nil
+		}
+		standbyReq.Compression = compression
+	}
+
+	result, err := s.InstanceManager.StandbyInstance(ctx, inst.Id, standbyReq)
 	if err != nil {
 		switch {
+		case errors.Is(err, instances.ErrInvalidRequest):
+			return oapi.StandbyInstance400JSONResponse{
+				Code:    "invalid_request",
+				Message: err.Error(),
+			}, nil
 		case errors.Is(err, instances.ErrInvalidState):
 			return oapi.StandbyInstance409JSONResponse{
 				Code:    "invalid_state",
@@ -951,6 +979,10 @@ func instanceToOAPI(inst instances.Instance) oapi.Instance {
 	if len(inst.Tags) > 0 {
 		oapiInst.Tags = toOAPITags(inst.Tags)
 	}
+	if inst.SnapshotPolicy != nil {
+		oapiPolicy := toOAPISnapshotPolicy(*inst.SnapshotPolicy)
+		oapiInst.SnapshotPolicy = &oapiPolicy
+	}
 
 	// Convert volume attachments
 	if len(inst.Volumes) > 0 {
@@ -985,3 +1017,73 @@ func instanceToOAPI(inst instances.Instance) oapi.Instance {
 
 	return oapiInst
 }
+
+func toDomainSnapshotCompressionConfig(cfg oapi.SnapshotCompressionConfig) (*snapshot.SnapshotCompressionConfig, error) {
+	out := &snapshot.SnapshotCompressionConfig{
+		Enabled: cfg.Enabled,
+	}
+	if cfg.Algorithm != nil {
+		algo := snapshot.SnapshotCompressionAlgorithm(strings.ToLower(string(*cfg.Algorithm)))
+		switch algo {
+		case snapshot.SnapshotCompressionAlgorithmZstd, snapshot.SnapshotCompressionAlgorithmLz4:
+		default:
+			return nil, fmt.Errorf("algorithm must be one of zstd or lz4, got %q", *cfg.Algorithm)
+		}
+		out.Algorithm = algo
+	}
+	if cfg.Level != nil {
+		level := *cfg.Level
+		algo := out.Algorithm
+		if algo == "" {
+			algo = snapshot.SnapshotCompressionAlgorithmZstd
+		}
+		switch algo {
+		case snapshot.SnapshotCompressionAlgorithmZstd:
+			if level < snapshot.MinSnapshotCompressionZstdLevel || level > snapshot.MaxSnapshotCompressionZstdLevel {
+				return nil, fmt.Errorf("level must be between %d and %d for zstd, got %d", snapshot.MinSnapshotCompressionZstdLevel, snapshot.MaxSnapshotCompressionZstdLevel, level)
+			}
+		case snapshot.SnapshotCompressionAlgorithmLz4:
+			if level < snapshot.MinSnapshotCompressionLz4Level || level > snapshot.MaxSnapshotCompressionLz4Level {
+				return nil, fmt.Errorf("level must be between %d and %d for lz4, got %d", snapshot.MinSnapshotCompressionLz4Level, snapshot.MaxSnapshotCompressionLz4Level, level)
+			}
+		}
+		out.Level = &level
+	}
+	return out, nil
+}
+
+func toInstanceSnapshotPolicy(policy oapi.SnapshotPolicy) (*instances.SnapshotPolicy, error) {
+	out := &instances.SnapshotPolicy{}
+	if policy.Compression != nil {
+		compression, err := toDomainSnapshotCompressionConfig(*policy.Compression)
+		if err != nil {
+			return nil, err
+		}
+		out.Compression = compression
+	}
+	return out, nil
+}
+
+func toOAPISnapshotCompressionConfig(cfg snapshot.SnapshotCompressionConfig) oapi.SnapshotCompressionConfig {
+	out := oapi.SnapshotCompressionConfig{
+		Enabled: cfg.Enabled,
+	}
+	if cfg.Algorithm != "" {
+		algo := oapi.SnapshotCompressionConfigAlgorithm(cfg.Algorithm)
+		out.Algorithm = &algo
+	}
+	if cfg.Level != nil {
+		level := *cfg.Level
+		out.Level = &level
+	}
+	return out
+}
+
+func toOAPISnapshotPolicy(policy instances.SnapshotPolicy) oapi.SnapshotPolicy {
+	out := oapi.SnapshotPolicy{}
+	if policy.Compression != nil {
+		compression := toOAPISnapshotCompressionConfig(*policy.Compression)
+		out.Compression = &compression
+	}
+	return out
+}

cmd/api/api/instances_test.go

@@ -204,6 +204,14 @@ type captureForkManager struct {
 	err     error
 }
 
+type captureStandbyManager struct {
+	instances.Manager
+	lastID  string
+	lastReq *instances.StandbyInstanceRequest
+	result  *instances.Instance
+	err     error
+}
+
 type captureUpdateManager struct {
 	instances.Manager
 	lastID  string
@@ -222,6 +230,16 @@ func (m *captureForkManager) ForkInstance(ctx context.Context, id string, req in
 	return m.result, nil
 }
 
+func (m *captureStandbyManager) StandbyInstance(ctx context.Context, id string, req instances.StandbyInstanceRequest) (*instances.Instance, error) {
+	reqCopy := req
+	m.lastID = id
+	m.lastReq = &reqCopy
+	if m.err != nil {
+		return nil, m.err
+	}
+	return m.result, nil
+}
+
 func (m *captureUpdateManager) UpdateInstance(ctx context.Context, id string, req instances.UpdateInstanceRequest) (*instances.Instance, error) {
 	reqCopy := req
 	m.lastID = id
@@ -677,6 +695,46 @@ func TestForkInstance_InvalidRequest(t *testing.T) {
 	assert.Equal(t, "invalid_request", badReq.Code)
 }
 
+func TestStandbyInstance_InvalidRequest(t *testing.T) {
+	t.Parallel()
+	svc := newTestService(t)
+
+	source := instances.Instance{
+		StoredMetadata: instances.StoredMetadata{
+			Id:             "standby-src",
+			Name:           "standby-src",
+			Image:          "docker.io/library/alpine:latest",
+			CreatedAt:      time.Now(),
+			HypervisorType: hypervisor.TypeCloudHypervisor,
+		},
+		State: instances.StateStopped,
+	}
+
+	mockMgr := &captureStandbyManager{
+		Manager: svc.InstanceManager,
+		err:     fmt.Errorf("%w: invalid snapshot compression level", instances.ErrInvalidRequest),
+	}
+	svc.InstanceManager = mockMgr
+
+	resp, err := svc.StandbyInstance(
+		mw.WithResolvedInstance(ctx(), source.Id, source),
+		oapi.StandbyInstanceRequestObject{
+			Id: source.Id,
+			Body: &oapi.StandbyInstanceRequest{
+				Compression: &oapi.SnapshotCompressionConfig{
+					Enabled: true,
+				},
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	badReq, ok := resp.(oapi.StandbyInstance400JSONResponse)
+	require.True(t, ok, "expected 400 response")
+	assert.Equal(t, "invalid_request", badReq.Code)
+	assert.Contains(t, badReq.Message, "invalid snapshot compression level")
+}
+
 func TestForkInstance_FromRunningFlagForwarded(t *testing.T) {
 	t.Parallel()
 	svc := newTestService(t)

cmd/api/api/optional_standby_body.go

@@ -0,0 +1,47 @@
+package api
+
+import (
+	"io"
+	"net/http"
+	"strings"
+)
+
+// NormalizeOptionalStandbyBody rewrites empty standby POST bodies to "{}"
+// so the generated strict handler can decode them without special casing.
+func NormalizeOptionalStandbyBody(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			next.ServeHTTP(w, r)
+			return
+		}
+		if isStandbyRoutePath(r.URL.Path) && requestBodyIsEmpty(r) {
+			r.Body = io.NopCloser(strings.NewReader(`{}`))
+			r.ContentLength = 2
+			if r.Header.Get("Content-Type") == "" {
+				r.Header.Set("Content-Type", "application/json")
+			}
+		}
+
+		next.ServeHTTP(w, r)
+	})
+}
+
+func isStandbyRoutePath(path string) bool {
+	if !strings.HasPrefix(path, "/instances/") || !strings.HasSuffix(path, "/standby") {
+		return false
+	}
+
+	instanceID := strings.TrimPrefix(path, "/instances/")
+	instanceID = strings.TrimSuffix(instanceID, "/standby")
+	return instanceID != "" && !strings.Contains(instanceID, "/")
+}
+
+func requestBodyIsEmpty(r *http.Request) bool {
+	if r == nil {
+		return true
+	}
+	if r.Body == nil || r.Body == http.NoBody {
+		return true
+	}
+	return r.ContentLength == 0 && len(r.TransferEncoding) == 0
+}

cmd/api/api/optional_standby_body_test.go

@@ -0,0 +1,108 @@
+package api
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNormalizeOptionalStandbyBody(t *testing.T) {
+	t.Parallel()
+
+	t.Run("empty standby body becomes empty JSON object", func(t *testing.T) {
+		t.Parallel()
+
+		var gotBody []byte
+		var gotContentType string
+		next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var err error
+			gotBody, err = io.ReadAll(r.Body)
+			require.NoError(t, err)
+			gotContentType = r.Header.Get("Content-Type")
+			w.WriteHeader(http.StatusNoContent)
+		})
+
+		req := httptest.NewRequest(http.MethodPost, "/instances/test/standby", nil)
+		rec := httptest.NewRecorder()
+
+		NormalizeOptionalStandbyBody(next).ServeHTTP(rec, req)
+
+		assert.Equal(t, http.StatusNoContent, rec.Code)
+		assert.Equal(t, []byte(`{}`), gotBody)
+		assert.Equal(t, "application/json", gotContentType)
+	})
+
+	t.Run("existing standby body is preserved", func(t *testing.T) {
+		t.Parallel()
+
+		var gotBody []byte
+		next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var err error
+			gotBody, err = io.ReadAll(r.Body)
+			require.NoError(t, err)
+			w.WriteHeader(http.StatusNoContent)
+		})
+
+		req := httptest.NewRequest(http.MethodPost, "/instances/test/standby", bytes.NewBufferString(`{"compression":{"enabled":true}}`))
+		rec := httptest.NewRecorder()
+
+		NormalizeOptionalStandbyBody(next).ServeHTTP(rec, req)
+
+		assert.Equal(t, http.StatusNoContent, rec.Code)
+		assert.Equal(t, []byte(`{"compression":{"enabled":true}}`), gotBody)
+	})
+
+	t.Run("non-standby route is untouched", func(t *testing.T) {
+		t.Parallel()
+
+		var gotBody []byte
+		next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var err error
+			gotBody, err = io.ReadAll(r.Body)
+			require.NoError(t, err)
+			w.WriteHeader(http.StatusNoContent)
+		})
+
+		req := httptest.NewRequest(http.MethodPost, "/instances/test/start", nil)
+		rec := httptest.NewRecorder()
+
+		NormalizeOptionalStandbyBody(next).ServeHTTP(rec, req)
+
+		assert.Equal(t, http.StatusNoContent, rec.Code)
+		assert.Empty(t, gotBody)
+	})
+
+	t.Run("non-post request skips standby normalization", func(t *testing.T) {
+		t.Parallel()
+
+		var gotBody []byte
+		next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var err error
+			gotBody, err = io.ReadAll(r.Body)
+			require.NoError(t, err)
+			w.WriteHeader(http.StatusNoContent)
+		})
+
+		req := httptest.NewRequest(http.MethodGet, "/instances/test/standby", nil)
+		rec := httptest.NewRecorder()
+
+		NormalizeOptionalStandbyBody(next).ServeHTTP(rec, req)
+
+		assert.Equal(t, http.StatusNoContent, rec.Code)
+		assert.Empty(t, gotBody)
+	})
+
+	t.Run("standby route matcher only accepts single path segment ids", func(t *testing.T) {
+		t.Parallel()
+
+		assert.True(t, isStandbyRoutePath("/instances/test/standby"))
+		assert.False(t, isStandbyRoutePath("/instances/test/start"))
+		assert.False(t, isStandbyRoutePath("/instances/test/standby/extra"))
+		assert.False(t, isStandbyRoutePath("/instances/test/nested/standby"))
+	})
+}