feat(api): fix Volume File API with litestream restore

- Replace direct GCS meta.db download with litestream restore
- Convert journal mode WAL -> DELETE after restore
- Replace raw GCS upload with litestream replicate for writes
- Add 409 Conflict for writes when volume attached to sandbox
- Add 412 Precondition Failed for fresh volumes (not yet mounted)
- Add cache invalidation on sandbox start/stop with volume
- Install litestream and sqlite3 in API container

Fixes: Volume File API 500 errors (volume metadata not found)

Author: omins

packages/api/Dockerfile

@@ -54,6 +54,15 @@ FROM alpine:${ALPINE_VERSION}
 
 COPY --from=builder /build/api/bin/api .
 
+# Install sqlite3 for journal mode conversion (WAL -> DELETE after litestream restore)
+RUN apk add --no-cache sqlite
+
+# Install litestream from moru-ai fork (for volume metadata restore/replicate)
+ARG LITESTREAM_VERSION=v0.5.6-moru.1
+RUN wget -q -O /usr/local/bin/litestream \
+    "https://github.com/moru-ai/litestream/releases/download/${LITESTREAM_VERSION}/litestream-linux" && \
+    chmod +x /usr/local/bin/litestream
+
 # Set Gin server to the production mode
 ENV GIN_MODE=release
 

packages/api/internal/handlers/sandbox_create.go

@@ -268,6 +268,12 @@ func (a *APIStore) PostSandboxes(c *gin.Context) {
 		return
 	}
 
+	// Invalidate volume client cache when sandbox attaches a volume
+	// This ensures API sees fresh metadata after sandbox mounts the volume
+	if volumeConfig != nil && a.juicefsPool != nil {
+		a.juicefsPool.InvalidateVolume(volumeConfig.VolumeID)
+	}
+
 	c.JSON(http.StatusCreated, &sbx)
 }
 

packages/api/internal/handlers/store.go

@@ -149,12 +149,6 @@ func NewAPIStore(ctx context.Context, tel *telemetry.Client, config cfg.Config)
 	// Start the periodic sync of template builds statuses
 	go templateManager.BuildsStatusPeriodicalSync(ctx)
 
-	// Start sandbox runs consumer (writes sandbox events to PostgreSQL)
-	if redisClient != nil {
-		sandboxRunsConsumer := sandboxruns.NewConsumer(redisClient, sqlcDB)
-		go sandboxRunsConsumer.Run(ctx)
-	}
-
 	// Initialize volume events delivery for Redis Streams
 	var volEventsDelivery events.Delivery[events.VolumeEvent]
 	if redisClient != nil {
@@ -163,7 +157,7 @@ func NewAPIStore(ctx context.Context, tel *telemetry.Client, config cfg.Config)
 	}
 
 	// JuiceFS pool for volume file operations (list, download, upload, delete)
-	// Uses SQLite metadata downloaded from GCS for each volume
+	// Uses litestream restore to get SQLite metadata from GCS for each volume
 	var juicefsPool *juicefs.Pool
 	if config.VolumesBucket != "" {
 		juicefsPool = juicefs.NewPool(juicefs.Config{
@@ -175,6 +169,17 @@ func NewAPIStore(ctx context.Context, tel *telemetry.Client, config cfg.Config)
 		logger.L().Info(ctx, "Volume file operations disabled (no VOLUMES_BUCKET configured)")
 	}
 
+	// Start sandbox runs consumer (writes sandbox events to PostgreSQL)
+	// Pass juicefsPool so it can invalidate cache when sandbox with volume terminates
+	if redisClient != nil {
+		var consumerOpts []sandboxruns.ConsumerOption
+		if juicefsPool != nil {
+			consumerOpts = append(consumerOpts, sandboxruns.WithVolumeInvalidator(juicefsPool.InvalidateVolume))
+		}
+		sandboxRunsConsumer := sandboxruns.NewConsumer(redisClient, sqlcDB, consumerOpts...)
+		go sandboxRunsConsumer.Run(ctx)
+	}
+
 	a := &APIStore{
 		config:               config,
 		Healthy:              false,

packages/api/internal/handlers/volume_files.go

@@ -17,6 +17,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/moru-ai/sandbox-infra/packages/api/internal/api"
+	"github.com/moru-ai/sandbox-infra/packages/api/internal/juicefs"
 	"github.com/moru-ai/sandbox-infra/packages/db/queries"
 )
 
@@ -92,6 +93,11 @@ func (a *APIStore) GetVolumesVolumeIDFiles(c *gin.Context, volumeID string, para
 	// Note: redisDB parameter is deprecated, passing 0 (code won't reach here due to nil check above)
 	client, err := a.juicefsPool.Get(ctx, volume.ID, 0)
 	if err != nil {
+		// Handle fresh volumes that haven't been mounted yet
+		if errors.Is(err, juicefs.ErrVolumeNotInitialized) {
+			a.sendAPIStoreError(c, http.StatusPreconditionFailed, "Volume not initialized - mount to a sandbox first")
+			return
+		}
 		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to connect to volume: "+err.Error())
 		return
 	}
@@ -176,6 +182,11 @@ func (a *APIStore) GetVolumesVolumeIDFilesDownload(c *gin.Context, volumeID stri
 	// Note: redisDB parameter is deprecated, passing 0 (code won't reach here due to nil check above)
 	client, err := a.juicefsPool.Get(ctx, volume.ID, 0)
 	if err != nil {
+		// Handle fresh volumes that haven't been mounted yet
+		if errors.Is(err, juicefs.ErrVolumeNotInitialized) {
+			a.sendAPIStoreError(c, http.StatusPreconditionFailed, "Volume not initialized - mount to a sandbox first")
+			return
+		}
 		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to connect to volume: "+err.Error())
 		return
 	}
@@ -229,6 +240,17 @@ func (a *APIStore) PutVolumesVolumeIDFilesUpload(c *gin.Context, volumeID string
 		return
 	}
 
+	// Check if volume is attached to a running sandbox (write conflict)
+	isAttached, err := a.sqlcDB.IsVolumeAttached(ctx, &volume.ID)
+	if err != nil {
+		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to check volume status")
+		return
+	}
+	if isAttached {
+		a.sendAPIStoreError(c, http.StatusConflict, "Cannot modify volume while attached to sandbox")
+		return
+	}
+
 	// Validate path
 	if !strings.HasPrefix(params.Path, "/") {
 		a.sendAPIStoreError(c, http.StatusBadRequest, "Path must be absolute")
@@ -242,6 +264,11 @@ func (a *APIStore) PutVolumesVolumeIDFilesUpload(c *gin.Context, volumeID string
 	// Note: redisDB parameter is deprecated, passing 0 (code won't reach here due to nil check above)
 	client, err := a.juicefsPool.Get(ctx, volume.ID, 0)
 	if err != nil {
+		// Handle fresh volumes that haven't been mounted yet
+		if errors.Is(err, juicefs.ErrVolumeNotInitialized) {
+			a.sendAPIStoreError(c, http.StatusPreconditionFailed, "Volume not initialized - mount to a sandbox first")
+			return
+		}
 		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to connect to volume: "+err.Error())
 		return
 	}
@@ -293,6 +320,17 @@ func (a *APIStore) DeleteVolumesVolumeIDFiles(c *gin.Context, volumeID string, p
 		return
 	}
 
+	// Check if volume is attached to a running sandbox (write conflict)
+	isAttached, err := a.sqlcDB.IsVolumeAttached(ctx, &volume.ID)
+	if err != nil {
+		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to check volume status")
+		return
+	}
+	if isAttached {
+		a.sendAPIStoreError(c, http.StatusConflict, "Cannot modify volume while attached to sandbox")
+		return
+	}
+
 	// Validate path
 	if !strings.HasPrefix(params.Path, "/") {
 		a.sendAPIStoreError(c, http.StatusBadRequest, "Path must be absolute")
@@ -312,6 +350,11 @@ func (a *APIStore) DeleteVolumesVolumeIDFiles(c *gin.Context, volumeID string, p
 	// Note: redisDB parameter is deprecated, passing 0 (code won't reach here due to nil check above)
 	client, err := a.juicefsPool.Get(ctx, volume.ID, 0)
 	if err != nil {
+		// Handle fresh volumes that haven't been mounted yet
+		if errors.Is(err, juicefs.ErrVolumeNotInitialized) {
+			a.sendAPIStoreError(c, http.StatusPreconditionFailed, "Volume not initialized - mount to a sandbox first")
+			return
+		}
 		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to connect to volume: "+err.Error())
 		return
 	}

packages/api/internal/juicefs/client.go

@@ -16,7 +16,6 @@ import (
 	"syscall"
 	"time"
 
-	"cloud.google.com/go/storage"
 	"github.com/juicedata/juicefs/pkg/chunk"
 	"github.com/juicedata/juicefs/pkg/fs"
 	"github.com/juicedata/juicefs/pkg/meta"
@@ -60,57 +59,36 @@ type Client struct {
 	closed bool
 }
 
+// ErrVolumeNotInitialized is returned when a fresh volume has not been mounted to a sandbox yet.
+var ErrVolumeNotInitialized = fmt.Errorf("volume not initialized - mount to a sandbox first")
+
 // NewClient creates a new JuiceFS client for a volume.
-// Downloads SQLite metadata from GCS and initializes JuiceFS for file operations.
+// Uses litestream restore to reconstruct SQLite metadata from LTX files in GCS.
 func NewClient(volumeID string, _ int32, config Config) (*Client, error) {
 	ctx := context.Background()
 
-	// Create temporary directory for SQLite file
-	tmpDir, err := os.MkdirTemp("", "juicefs-client-*")
+	// Restore metadata from litestream
+	restoreResult, err := restoreMetaDB(ctx, volumeID, config.GCSBucket)
 	if err != nil {
-		return nil, fmt.Errorf("create temp dir: %w", err)
+		return nil, fmt.Errorf("restore metadata: %w", err)
 	}
 
-	sqlitePath := filepath.Join(tmpDir, "meta.db")
-
-	// Download SQLite metadata from GCS
-	gcsClient, err := storage.NewClient(ctx)
-	if err != nil {
-		os.RemoveAll(tmpDir)
-		return nil, fmt.Errorf("create GCS client: %w", err)
+	// Fresh volumes must be mounted to a sandbox first to initialize JuiceFS metadata
+	if restoreResult.IsFreshVolume {
+		cleanupVolumeDir(volumeID)
+		return nil, ErrVolumeNotInitialized
 	}
-	defer gcsClient.Close()
-
-	_, metaPrefix := gcsPathsForVolume(config.GCSBucket, volumeID)
-	bucket := gcsClient.Bucket(config.GCSBucket)
-	metaObj := bucket.Object(metaPrefix + "meta.db")
 
-	reader, err := metaObj.NewReader(ctx)
-	if err != nil {
-		os.RemoveAll(tmpDir)
-		if err == storage.ErrObjectNotExist {
-			return nil, fmt.Errorf("volume metadata not found: %s", volumeID)
-		}
-		return nil, fmt.Errorf("download metadata: %w", err)
-	}
-
-	sqliteFile, err := os.Create(sqlitePath)
-	if err != nil {
-		reader.Close()
-		os.RemoveAll(tmpDir)
-		return nil, fmt.Errorf("create sqlite file: %w", err)
-	}
+	sqlitePath := restoreResult.MetaDBPath
+	tmpDir := filepath.Dir(sqlitePath)
 
-	if _, err = io.Copy(sqliteFile, reader); err != nil {
-		sqliteFile.Close()
-		reader.Close()
-		os.RemoveAll(tmpDir)
-		return nil, fmt.Errorf("write sqlite file: %w", err)
+	// Convert journal mode from WAL to DELETE (required for JuiceFS)
+	if err := convertJournalMode(ctx, sqlitePath); err != nil {
+		cleanupVolumeDir(volumeID)
+		return nil, fmt.Errorf("convert journal mode: %w", err)
 	}
-	sqliteFile.Close()
-	reader.Close()
 
-	logger.L().Info(ctx, "Downloaded volume metadata",
+	logger.L().Info(ctx, "Restored volume metadata via litestream",
 		zap.String("volume_id", volumeID),
 		zap.String("path", sqlitePath))
 
@@ -251,50 +229,30 @@ func (c *Client) Close() error {
 	return nil
 }
 
-// SyncToGCS uploads the current SQLite metadata to GCS.
+// SyncToGCS syncs the current SQLite metadata to GCS via litestream.
 // This should be called after write operations to persist changes.
 func (c *Client) SyncToGCS() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	return c.syncToGCSLocked()
 }
 
-// syncToGCSLocked uploads SQLite metadata to GCS (must hold lock).
+// syncToGCSLocked syncs SQLite metadata to GCS via litestream (must hold lock).
+// Uses litestream replicate to ensure compatibility with sandbox's Litestream daemon.
 func (c *Client) syncToGCSLocked() error {
 	if c.sqlitePath == "" {
 		return nil
 	}
 
 	ctx := context.Background()
 
-	gcsClient, err := storage.NewClient(ctx)
-	if err != nil {
-		return fmt.Errorf("create GCS client: %w", err)
-	}
-	defer gcsClient.Close()
-
-	_, metaPrefix := gcsPathsForVolume(c.config.GCSBucket, c.volumeID)
-	bucket := gcsClient.Bucket(c.config.GCSBucket)
-	metaObj := bucket.Object(metaPrefix + "meta.db")
-
-	// Read local SQLite file
-	sqliteFile, err := os.Open(c.sqlitePath)
-	if err != nil {
-		return fmt.Errorf("open sqlite file: %w", err)
-	}
-	defer sqliteFile.Close()
-
-	// Upload to GCS
-	writer := metaObj.NewWriter(ctx)
-	if _, err = io.Copy(writer, sqliteFile); err != nil {
-		writer.Close()
-		return fmt.Errorf("upload metadata: %w", err)
-	}
-	if err = writer.Close(); err != nil {
-		return fmt.Errorf("close GCS writer: %w", err)
+	// Use litestream replicate to sync metadata to GCS
+	// This ensures compatibility with the sandbox's Litestream daemon
+	if err := syncViaLitestream(ctx, c.volumeID, c.sqlitePath, c.config.GCSBucket); err != nil {
+		return fmt.Errorf("litestream sync: %w", err)
 	}
 
-	logger.L().Debug(ctx, "Synced metadata to GCS",
+	logger.L().Debug(ctx, "Synced metadata to GCS via litestream",
 		zap.String("volume_id", c.volumeID))
 
 	return nil

packages/api/internal/juicefs/pool.go

@@ -7,10 +7,15 @@ import (
 	"fmt"
 	"sync"
 	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/moru-ai/sandbox-infra/packages/shared/pkg/logger"
 )
 
 // Pool manages a pool of JuiceFS clients, one per volume.
 // Clients are cached and reused to avoid repeated initialization.
+// Cache is invalidated when volume mount state changes (sandbox starts/stops).
 type Pool struct {
 	config Config
 
@@ -66,6 +71,27 @@ func (p *Pool) Get(ctx context.Context, volumeID string, _ int32) (*Client, erro
 	return client, nil
 }
 
+// InvalidateVolume removes a volume's cached client.
+// This should be called when a sandbox starts or stops with the volume attached,
+// as the volume's metadata may have changed.
+func (p *Pool) InvalidateVolume(volumeID string) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if pc, ok := p.clients[volumeID]; ok {
+		// Close the client (best effort - ignore errors during invalidation)
+		if err := pc.client.Close(); err != nil {
+			logger.L().Warn(context.Background(), "Error closing invalidated volume client",
+				zap.String("volume_id", volumeID),
+				zap.Error(err))
+		}
+		delete(p.clients, volumeID)
+
+		logger.L().Info(context.Background(), "Invalidated volume client cache",
+			zap.String("volume_id", volumeID))
+	}
+}
+
 // Config returns the pool's configuration.
 func (p *Pool) Config() Config {
 	return p.config