From b817a5b772582fcd0d0b3173db11fc4cd6eca4d1 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 21 May 2026 22:22:08 -0500 Subject: [PATCH] Implemented shared S3 audio caching for prepare and restore --include-audio --- docs/cli.md | 2 + docs/config.md | 7 + docs/internal/command-restore.md | 4 + docs/internal/stage-prepare.md | 4 + docs/internal/workspace.md | 5 +- docs/operations.md | 2 + internal/adapters/storage/fake.go | 15 +- internal/app/commands_test.go | 4 + internal/app/restore_execute.go | 37 ++++ internal/app/restore_execution_test.go | 48 ++++++ internal/app/restore_plan.go | 38 +++++ internal/app/restore_plan_test.go | 27 +++ internal/artifacts/paths.go | 40 +++++ internal/artifacts/paths_model_test.go | 43 +++++ internal/audio/s3_audio.go | 223 +++++++++++++++++++++++++ internal/audio/s3_audio_test.go | 145 ++++++++++++++++ internal/config/cache_config_test.go | 67 ++++++++ internal/config/config.go | 7 + internal/config/defaults.go | 2 + internal/config/load.go | 13 ++ internal/config/validate.go | 10 ++ internal/manifest/manifest.go | 1 + internal/stage/prepare.go | 70 +++++--- internal/stage/prepare_test.go | 87 +++++++++- 24 files changed, 876 insertions(+), 25 deletions(-) create mode 100644 internal/audio/s3_audio.go create mode 100644 internal/audio/s3_audio_test.go create mode 100644 internal/config/cache_config_test.go diff --git a/docs/cli.md b/docs/cli.md index 4009bfb..ad85118 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -376,6 +376,8 @@ Common failure cases: - local conflicts without `--force`. - session lock conflict. +When `--include-audio` is set, S3 audio files are restored through the shared audio cache. Cache hits avoid re-downloading large audio objects. + ## Common Workflows Default-discovery run: diff --git a/docs/config.md b/docs/config.md index 6185ec6..6918e40 100644 --- a/docs/config.md +++ b/docs/config.md @@ -165,6 +165,10 @@ spool: root: /var/spool/narratio delete_audio_after_archive: true +cache: + root: /var/cache/narratio + s3_audio: true + archive: enabled: true upload_run: true @@ -261,6 +265,8 @@ Operational notes: | `pipeline.storage.s3.secret_access_key_env` | string | No | `OBJECT_STORAGE_KEY` | | `pipeline.spool.root` | string | No | `/var/spool/narratio` | | `pipeline.spool.delete_audio_after_archive` | bool | No | `false` | +| `pipeline.cache.root` | string | No | `/var/cache/narratio` | +| `pipeline.cache.s3_audio` | bool | No | `true` | | `pipeline.archive.enabled` | bool | No | `true` | | `pipeline.archive.upload_run` | bool | No | `true` | | `pipeline.archive.promote_artifacts[]` | list | No | trimmed transcript rule | @@ -401,6 +407,7 @@ Restore-related implications: - restore remote identity requires archive S3 identity to resolve (`pipeline.storage.s3.bucket` and session prefix derivation inputs). - restore scope considers committed current state and durable paths (`manifest.json`, `transcripts/**`, `artifacts/**`, `previous/**`, optional `audio/**`). +- S3 audio downloads use `pipeline.spool.root` for active downloads and `pipeline.cache.root` for reusable cached audio when `pipeline.cache.s3_audio` is true. ## 7. Full campaign reference diff --git a/docs/internal/command-restore.md b/docs/internal/command-restore.md index 3d1ed30..aea2102 100644 --- a/docs/internal/command-restore.md +++ b/docs/internal/command-restore.md @@ -34,6 +34,8 @@ Does not own: - `pipeline.workspace.root` (local restore target root). - `pipeline.storage.*` (remote backend + archive identity derivation). - `pipeline.storage.s3.*` identity components used by archive prefix helpers. +- `pipeline.spool.root` for active audio downloads. +- `pipeline.cache.root` and `pipeline.cache.s3_audio` for reusable S3 audio cache. - `session.session_id` - `session.campaign` @@ -48,6 +50,7 @@ Does not own: - `current/run_id.txt` must exist and be non-empty. - `current/manifest.json` must decode and match requested session/campaign. - Non-dry-run writes restore files to canonical session paths. +- With `--include-audio`, restore uses the shared S3 audio cache for `audio/**` objects. Cache hits avoid object downloads; cache misses download through spool, install the work file, and populate cache. - Manifest install behavior: - validated before replacement. - installed last among download actions. @@ -72,6 +75,7 @@ Restore path scope: ## Skip and resume behavior - Restore does not participate in stage skip/resume decisions. - Restore provides durable local state so subsequent stage commands can resume or rerun based on restored manifest state. +- Audio cache is outside the workspace and is reused across restore and prepare invocations. - Dry-run is read-only and returns plan output only. ## Failure behavior diff --git a/docs/internal/stage-prepare.md b/docs/internal/stage-prepare.md index 856fb89..37940b8 100644 --- a/docs/internal/stage-prepare.md +++ b/docs/internal/stage-prepare.md @@ -55,6 +55,8 @@ Does not own: - `session.inputs.audio_s3.prefix` - `pipeline.workspace.root` - `pipeline.spool.root` +- `pipeline.cache.root` +- `pipeline.cache.s3_audio` - `pipeline.storage.s3.bucket` - `pipeline.storage.s3.root_prefix` - `pipeline.scriptorium.artifacts..enabled` @@ -73,6 +75,7 @@ Does not own: ## State and manifest behavior - Ensures workspace layout exists. - Materializes canonical input files and audio files. +- For S3 audio, uses run-scoped spool for active downloads and durable cache for reusable audio files; cache hits copy directly to work audio without downloading the object again. - Records `inputs/session.yml` provenance as local `session_config` or remote `session_config.s3`. - Resolves campaign-provided stable input paths relative to `campaign.yml`. - Resolves session-provided stable input overrides relative to `session.yml`. @@ -86,6 +89,7 @@ Does not own: - records hydrated previous inputs in `manifest.Inputs` with source `previous_session_archive.current`. - If no canonical previous-session requirements exist, prepare does not manage `previous/`. - `manifest.Inputs` is sorted deterministically by `(kind, path)`. +- S3 audio `manifest.Inputs` retain S3 provenance and include `cache_path`; `spool_path` is present only when the current prepare invocation downloaded the file. ## Required and optional previous-session behavior - `previous_session_id` unset: diff --git a/docs/internal/workspace.md b/docs/internal/workspace.md index c356df6..41ea481 100644 --- a/docs/internal/workspace.md +++ b/docs/internal/workspace.md @@ -32,6 +32,8 @@ Does not own: - `pipeline.workspace.cleanup_after_archive` - `pipeline.spool.root` - `pipeline.spool.delete_audio_after_archive` +- `pipeline.cache.root` +- `pipeline.cache.s3_audio` - `session.campaign` - `session.session_id` @@ -43,7 +45,8 @@ None directly in this subsystem. Stages may use object storage adapters and then - Invocation history is persisted per run in run manifests under `runs/{run_id}/manifest.json`. - During each run, stage outputs are often written run-local first (`runs/{run_id}/{stage}/outputs/...`) and promoted to canonical session paths after stage success. - `manifest.Artifacts` entries record `ProducerRunID` for durable outputs. -- For S3 audio sessions, `prepare` records spool/work paths and S3 provenance in `manifest.Inputs`. +- For S3 audio sessions, `prepare` records work/cache paths, S3 provenance, and spool path when the invocation downloaded the object. +- Durable cache state under `pipeline.cache.root` is not workspace state and is not part of session cleanup semantics. ## Skip and Resume Behavior - Skip/resume decisions are made in `internal/app` (`run_control.go`, `resume.go`) using stage status in the session manifest. diff --git a/docs/operations.md b/docs/operations.md index 93f37b3..a01d251 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -189,6 +189,8 @@ Cleanup toggles: - `pipeline.spool.delete_audio_after_archive=true` deletes run-scoped spool audio. - `pipeline.workspace.cleanup_after_archive=true` deletes run-scoped local run directory. +The S3 audio cache under `pipeline.cache.root` is durable input cache state, not workspace or spool state. Cleanup does not delete it. + Cleanup eligibility gates: - archive enabled - archive run upload enabled diff --git a/internal/adapters/storage/fake.go b/internal/adapters/storage/fake.go index 0039a17..882dd2e 100644 --- a/internal/adapters/storage/fake.go +++ b/internal/adapters/storage/fake.go @@ -27,8 +27,9 @@ type FakeBackend struct { Err error Result ArchiveResult - Objects map[string]FakeObject - Uploads []FakeUploadCall + Objects map[string]FakeObject + Uploads []FakeUploadCall + Downloads []FakeDownloadCall ListErr error DownloadErr error @@ -43,6 +44,12 @@ type FakeUploadCall struct { Options UploadOptions } +// FakeDownloadCall captures one download invocation in call order. +type FakeDownloadCall struct { + Key string + LocalPath string +} + // Archive records request and returns configured response. func (f *FakeBackend) Archive(ctx context.Context, req ArchiveRequest) (ArchiveResult, error) { if err := ctx.Err(); err != nil { @@ -130,6 +137,10 @@ func (f *FakeBackend) Download(ctx context.Context, key, localPath string) error if !ok { return fmt.Errorf("download object %q: %w", key, os.ErrNotExist) } + f.Downloads = append(f.Downloads, FakeDownloadCall{ + Key: normalizeObjectKey(key), + LocalPath: localPath, + }) if err := os.MkdirAll(filepath.Dir(localPath), 0o755); err != nil { return fmt.Errorf("download object %q: create parent directory: %w", key, err) diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index 019aa3a..71b2a82 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -410,6 +410,10 @@ func writeValidConfigFiles(t *testing.T, workspaceRoot string, transcribeURL ... pipelineYAML := `workspace: root: ` + workspaceRoot + ` +cache: + root: ` + filepath.Join(workspaceRoot, "cache") + ` +spool: + root: ` + filepath.Join(workspaceRoot, "spool") + ` storage: backend: s3 s3: diff --git a/internal/app/restore_execute.go b/internal/app/restore_execute.go index d0744ff..34d5619 100644 --- a/internal/app/restore_execute.go +++ b/internal/app/restore_execute.go @@ -9,6 +9,7 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/audio" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) @@ -92,6 +93,10 @@ func executeRestoreDownloadAction( return fmt.Errorf("restore plan local path mismatch for %q", action.LocalRelativePath) } + if restoreActionIsAudio(action) { + return executeRestoreAudioAction(ctx, cfg, safeLocalPath, action, store) + } + tmpPath, err := downloadObjectToSiblingTemp(ctx, store, action.RemoteKey, safeLocalPath) if err != nil { return fmt.Errorf("download to temp file: %w", err) @@ -120,6 +125,38 @@ func executeRestoreDownloadAction( return nil } +func executeRestoreAudioAction( + ctx context.Context, + cfg *config.Config, + safeLocalPath string, + action RestoreAction, + store storage.ObjectStore, +) error { + if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Storage.S3 == nil || cfg.Session == nil { + return fmt.Errorf("resolved s3 config and session are required") + } + spoolDir := artifacts.SessionSpoolRestoreAudioDir(cfg.Pipeline.Spool.Root, cfg.Session.Campaign, cfg.Session.SessionID) + spoolPath := filepath.Join(spoolDir, filepath.Base(safeLocalPath)) + cacheEnabled := cfg.Pipeline.Cache.S3Audio == nil || *cfg.Pipeline.Cache.S3Audio + _, err := audio.MaterializeS3Audio(ctx, audio.S3MaterializeRequest{ + Store: store, + Object: storage.ObjectInfo{ + Key: action.RemoteKey, + Size: action.Size, + ETag: action.ETag, + }, + Bucket: strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket), + CacheRoot: strings.TrimSpace(cfg.Pipeline.Cache.Root), + CacheEnabled: cacheEnabled, + SpoolPath: spoolPath, + DestPath: safeLocalPath, + }) + if err != nil { + return fmt.Errorf("materialize audio: %w", err) + } + return nil +} + func downloadObjectToSiblingTemp(ctx context.Context, store storage.ObjectStore, remoteKey, destPath string) (string, error) { if strings.TrimSpace(destPath) == "" { return "", fmt.Errorf("destination path is required") diff --git a/internal/app/restore_execution_test.go b/internal/app/restore_execution_test.go index a51b5fd..72b94da 100644 --- a/internal/app/restore_execution_test.go +++ b/internal/app/restore_execution_test.go @@ -87,6 +87,44 @@ func TestExecuteRestoreIncludeAudioRestoresAudio(t *testing.T) { } } +func TestExecuteRestoreIncludeAudioUsesCacheAfterWorkspaceDeletion(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + + fake := &storage.FakeBackend{} + cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath) + audioKey := sessionPrefix + "audio/alice.flac" + seedRestoreObject(fake, audioKey, []byte("remote-audio")) + + restoreWithStoreAndRealPhases(t, fake) + + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute([]string{"restore", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--include-audio"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("first restore exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if got := fakeDownloadCount(fake, audioKey); got != 1 { + t.Fatalf("audio downloads after first restore = %d, want 1", got) + } + sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID) + mustReadEquals(t, filepath.Join(sessionRoot, "audio", "alice.flac"), "remote-audio") + + if err := os.RemoveAll(sessionRoot); err != nil { + t.Fatalf("remove session root: %v", err) + } + stdout.Reset() + stderr.Reset() + code = Execute([]string{"restore", "--config", pipelinePath, "--campaign", campaignPath, "--session", sessionPath, "--include-audio"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("second restore exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if got := fakeDownloadCount(fake, audioKey); got != 1 { + t.Fatalf("audio downloads after cached restore = %d, want still 1", got) + } + mustReadEquals(t, filepath.Join(sessionRoot, "audio", "alice.flac"), "remote-audio") +} + func TestExecuteRestoreRestoresPreviousCacheWhenPresent(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) @@ -374,6 +412,16 @@ func mustReadEquals(t *testing.T, path, want string) { } } +func fakeDownloadCount(fake *storage.FakeBackend, key string) int { + count := 0 + for _, call := range fake.Downloads { + if call.Key == key { + count++ + } + } + return count +} + type stagedManifestDownloadStore struct { delegate *storage.FakeBackend manifestKey string diff --git a/internal/app/restore_plan.go b/internal/app/restore_plan.go index 7e61237..41bb7d2 100644 --- a/internal/app/restore_plan.go +++ b/internal/app/restore_plan.go @@ -257,6 +257,35 @@ func classifyRestoreAction( return action, nil } + if restoreRelativePathIsAudio(localRelPath) { + if object.Size > 0 { + if info.Size() == object.Size { + action.Kind = RestoreActionSkipSame + action.SameLocal = true + action.Reason = "local audio size matches remote content" + return action, nil + } + if force { + action.Kind = RestoreActionDownload + action.Reason = "local audio differs (size mismatch); overwrite with --force" + return action, nil + } + action.Kind = RestoreActionConflict + action.Conflict = true + action.Reason = "local audio differs (size mismatch)" + return action, nil + } + if force { + action.Kind = RestoreActionDownload + action.Reason = "local audio exists; remote size unavailable; overwrite with --force" + return action, nil + } + action.Kind = RestoreActionConflict + action.Conflict = true + action.Reason = "local audio exists; remote size unavailable" + return action, nil + } + if object.Size > 0 && info.Size() != object.Size { if force { action.Kind = RestoreActionDownload @@ -303,6 +332,15 @@ func classifyRestoreAction( return action, nil } +func restoreActionIsAudio(action RestoreAction) bool { + return restoreRelativePathIsAudio(action.LocalRelativePath) +} + +func restoreRelativePathIsAudio(rel string) bool { + cleanRel := path.Clean(strings.TrimSpace(rel)) + return cleanRel == config.PathAudioDirSegment || strings.HasPrefix(cleanRel, config.PathAudioDirSegment+"/") +} + func writeRestorePlan(out io.Writer, current *RemoteCurrentState, plan *RestorePlan, opts RestorePlanOptions) error { if out == nil { return fmt.Errorf("output writer is required") diff --git a/internal/app/restore_plan_test.go b/internal/app/restore_plan_test.go index ebd666e..029513a 100644 --- a/internal/app/restore_plan_test.go +++ b/internal/app/restore_plan_test.go @@ -58,6 +58,33 @@ func TestRestorePlanIncludeAudio(t *testing.T) { } } +func TestRestorePlanExistingAudioUsesSizeWithoutRemoteChecksumDownload(t *testing.T) { + cfg := restorePlanConfig(t) + current := restorePlanCurrentState(t, cfg) + store := &storage.FakeBackend{} + + seedRestoreObject(store, current.CurrentManifestKey, []byte(`{"session_id":"2026-05-03"}`)) + seedRestoreObject(store, current.SessionPrefix+"audio/alice.flac", []byte("audio")) + sessionRoot := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID) + mustWriteTestFile(t, filepath.Join(sessionRoot, "audio", "alice.flac"), "local") + + plan, err := buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{IncludeAudio: true}) + if err != nil { + t.Fatalf("buildRestorePlan() error = %v", err) + } + if len(store.Downloads) != 0 { + t.Fatalf("downloads = %d, want no remote checksum download for audio", len(store.Downloads)) + } + actionByRel := map[string]RestoreAction{} + for _, action := range plan.Actions { + actionByRel[action.LocalRelativePath] = action + } + audioAction := actionByRel["audio/alice.flac"] + if audioAction.Kind != RestoreActionSkipSame { + t.Fatalf("audio action kind = %q, want %q", audioAction.Kind, RestoreActionSkipSame) + } +} + func TestRestorePlanIncludesPreviousCacheByDefault(t *testing.T) { cfg := restorePlanConfig(t) current := restorePlanCurrentState(t, cfg) diff --git a/internal/artifacts/paths.go b/internal/artifacts/paths.go index 9de9a38..c82cc77 100644 --- a/internal/artifacts/paths.go +++ b/internal/artifacts/paths.go @@ -1,6 +1,8 @@ package artifacts import ( + "fmt" + "path" "path/filepath" "strings" @@ -89,6 +91,44 @@ func SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID string) string { return filepath.Join(spoolRoot, campaign, sessionID, runID, config.PathAudioDirSegment) } +// SessionSpoolRestoreAudioDir returns the local spool audio path for restore downloads. +func SessionSpoolRestoreAudioDir(spoolRoot, campaign, sessionID string) string { + return filepath.Join(spoolRoot, campaign, sessionID, "restore", config.PathAudioDirSegment) +} + +// S3AudioCachePath returns the durable local cache path for one S3 audio object. +func S3AudioCachePath(cacheRoot, bucket, key string) (string, error) { + root := filepath.Clean(strings.TrimSpace(cacheRoot)) + if root == "." || root == "" { + return "", fmt.Errorf("cache root is required") + } + bucket = strings.Trim(strings.TrimSpace(bucket), "/") + if bucket == "" || bucket == "." || bucket == ".." || strings.Contains(bucket, "/") || strings.Contains(bucket, `\`) { + return "", fmt.Errorf("bucket is required and must be a single path segment") + } + rawKey := strings.ReplaceAll(strings.TrimSpace(key), `\`, "/") + if strings.HasPrefix(rawKey, "/") { + return "", fmt.Errorf("s3 key must not be absolute") + } + cleanKey := cleanCacheS3Key(rawKey) + if cleanKey == "" { + return "", fmt.Errorf("s3 key is required") + } + if cleanKey == ".." || strings.HasPrefix(cleanKey, "../") || strings.HasPrefix(cleanKey, "/") { + return "", fmt.Errorf("s3 key must not escape cache root") + } + return filepath.Join(root, "s3", bucket, filepath.FromSlash(cleanKey)), nil +} + +func cleanCacheS3Key(key string) string { + normalized := strings.ReplaceAll(strings.TrimSpace(key), `\`, "/") + normalized = strings.Trim(normalized, "/") + if normalized == "" { + return "" + } + return path.Clean(normalized) +} + // SessionPreviousDir returns the previous-session state directory for already-resolved session paths. func SessionPreviousDir(paths SessionPaths) string { return paths.PreviousDir diff --git a/internal/artifacts/paths_model_test.go b/internal/artifacts/paths_model_test.go index eac78ac..0805162 100644 --- a/internal/artifacts/paths_model_test.go +++ b/internal/artifacts/paths_model_test.go @@ -114,3 +114,46 @@ func TestSessionSpoolAudioDir(t *testing.T) { t.Fatalf("SessionSpoolAudioDir() = %q, want %q", got, want) } } + +func TestSessionSpoolRestoreAudioDir(t *testing.T) { + root := "/var/spool/narratio" + got := SessionSpoolRestoreAudioDir(root, "forsaken", "2026-04-19") + want := filepath.Join(root, "forsaken", "2026-04-19", "restore", "audio") + if got != want { + t.Fatalf("SessionSpoolRestoreAudioDir() = %q, want %q", got, want) + } +} + +func TestS3AudioCachePath(t *testing.T) { + got, err := S3AudioCachePath("/var/cache/narratio", "my-dnd-archive", "dnd/campaigns/forsaken/sessions/2026-04-19/audio/alice.flac") + if err != nil { + t.Fatalf("S3AudioCachePath() error = %v", err) + } + want := filepath.Join("/var/cache/narratio", "s3", "my-dnd-archive", "dnd", "campaigns", "forsaken", "sessions", "2026-04-19", "audio", "alice.flac") + if got != want { + t.Fatalf("S3AudioCachePath() = %q, want %q", got, want) + } +} + +func TestS3AudioCachePathRejectsUnsafeInputs(t *testing.T) { + tests := []struct { + name string + root string + bucket string + key string + }{ + {name: "empty root", root: "", bucket: "bucket", key: "audio/a.flac"}, + {name: "empty bucket", root: "/cache", bucket: "", key: "audio/a.flac"}, + {name: "bucket slash", root: "/cache", bucket: "bad/bucket", key: "audio/a.flac"}, + {name: "empty key", root: "/cache", bucket: "bucket", key: ""}, + {name: "escaping key", root: "/cache", bucket: "bucket", key: "../audio/a.flac"}, + {name: "absolute key", root: "/cache", bucket: "bucket", key: "/audio/a.flac"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got, err := S3AudioCachePath(tt.root, tt.bucket, tt.key); err == nil { + t.Fatalf("S3AudioCachePath() = %q, want error", got) + } + }) + } +} diff --git a/internal/audio/s3_audio.go b/internal/audio/s3_audio.go new file mode 100644 index 0000000..64f71fb --- /dev/null +++ b/internal/audio/s3_audio.go @@ -0,0 +1,223 @@ +package audio + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" +) + +// S3MaterializeRequest describes one S3-backed audio materialization. +type S3MaterializeRequest struct { + Store storage.ObjectStore + Object storage.ObjectInfo + Bucket string + CacheRoot string + CacheEnabled bool + SpoolPath string + DestPath string +} + +// S3MaterializeResult captures local materialization provenance. +type S3MaterializeResult struct { + Checksum string + CachePath string + SpoolPath string + CacheHit bool + Downloaded bool +} + +// MaterializeS3Audio installs one S3 audio object into the destination path, +// reusing and refreshing the durable local cache when enabled. +func MaterializeS3Audio(ctx context.Context, req S3MaterializeRequest) (S3MaterializeResult, error) { + if err := ctx.Err(); err != nil { + return S3MaterializeResult{}, err + } + if req.Store == nil { + return S3MaterializeResult{}, fmt.Errorf("object store is required") + } + key := strings.TrimSpace(req.Object.Key) + if key == "" { + return S3MaterializeResult{}, fmt.Errorf("s3 object key is required") + } + if strings.TrimSpace(req.DestPath) == "" { + return S3MaterializeResult{}, fmt.Errorf("destination path is required") + } + + result := S3MaterializeResult{} + if req.CacheEnabled { + cachePath, err := artifacts.S3AudioCachePath(req.CacheRoot, req.Bucket, key) + if err != nil { + return S3MaterializeResult{}, fmt.Errorf("resolve audio cache path: %w", err) + } + result.CachePath = cachePath + if ok, err := validCachedAudio(cachePath, req.Object.Size); err != nil { + return S3MaterializeResult{}, err + } else if ok { + checksum, err := copyFileAtomicWithChecksum(cachePath, req.DestPath, 0o644) + if err != nil { + return S3MaterializeResult{}, fmt.Errorf("materialize cached audio %q: %w", cachePath, err) + } + result.Checksum = checksum + result.CacheHit = true + return result, nil + } + } + + spoolPath := strings.TrimSpace(req.SpoolPath) + if spoolPath == "" { + return S3MaterializeResult{}, fmt.Errorf("spool path is required for s3 audio download") + } + if err := downloadObjectAtomic(ctx, req.Store, key, spoolPath); err != nil { + return S3MaterializeResult{}, fmt.Errorf("download s3 audio object %q: %w", key, err) + } + if err := validateLocalAudio(spoolPath, req.Object.Size); err != nil { + return S3MaterializeResult{}, fmt.Errorf("validate downloaded audio %q: %w", spoolPath, err) + } + + checksum, err := copyFileAtomicWithChecksum(spoolPath, req.DestPath, 0o644) + if err != nil { + return S3MaterializeResult{}, fmt.Errorf("materialize downloaded audio %q: %w", filepath.Base(req.DestPath), err) + } + result.Checksum = checksum + result.SpoolPath = spoolPath + result.Downloaded = true + + if result.CachePath != "" { + if _, err := copyFileAtomicWithChecksum(spoolPath, result.CachePath, 0o644); err != nil { + return S3MaterializeResult{}, fmt.Errorf("populate audio cache %q: %w", result.CachePath, err) + } + } + + return result, nil +} + +func validCachedAudio(path string, expectedSize int64) (bool, error) { + info, err := os.Stat(path) + if err != nil { + if os.IsNotExist(err) { + return false, nil + } + return false, fmt.Errorf("stat cached audio %q: %w", path, err) + } + if info.IsDir() { + return false, fmt.Errorf("cached audio path is a directory: %q", path) + } + if info.Size() <= 0 { + return false, nil + } + if expectedSize > 0 && info.Size() != expectedSize { + return false, nil + } + return true, nil +} + +func validateLocalAudio(path string, expectedSize int64) error { + info, err := os.Stat(path) + if err != nil { + return fmt.Errorf("stat: %w", err) + } + if info.IsDir() { + return fmt.Errorf("path is a directory") + } + if info.Size() <= 0 { + return fmt.Errorf("file is empty") + } + if expectedSize > 0 && info.Size() != expectedSize { + return fmt.Errorf("size %d does not match remote size %d", info.Size(), expectedSize) + } + return nil +} + +func downloadObjectAtomic(ctx context.Context, store storage.ObjectStore, key, destPath string) error { + if strings.TrimSpace(destPath) == "" { + return fmt.Errorf("destination path is required") + } + dir := filepath.Dir(destPath) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("create destination directory: %w", err) + } + base := filepath.Base(destPath) + tmp, err := os.CreateTemp(dir, "."+base+".download-*.tmp") + if err != nil { + return fmt.Errorf("create temp file: %w", err) + } + tmpPath := tmp.Name() + if err := tmp.Close(); err != nil { + _ = os.Remove(tmpPath) + return fmt.Errorf("close temp file: %w", err) + } + removeTmp := true + defer func() { + if removeTmp { + _ = os.Remove(tmpPath) + } + }() + + if err := store.Download(ctx, key, tmpPath); err != nil { + return err + } + if err := os.Chmod(tmpPath, 0o644); err != nil { + return fmt.Errorf("set temp file permissions: %w", err) + } + if err := os.Rename(tmpPath, destPath); err != nil { + return fmt.Errorf("install downloaded file: %w", err) + } + removeTmp = false + return nil +} + +func copyFileAtomicWithChecksum(src, dst string, perm os.FileMode) (string, error) { + if strings.TrimSpace(src) == "" || strings.TrimSpace(dst) == "" { + return "", fmt.Errorf("source and destination paths are required") + } + in, err := os.Open(src) + if err != nil { + return "", err + } + defer func() { _ = in.Close() }() + + if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil { + return "", fmt.Errorf("create destination directory: %w", err) + } + base := filepath.Base(dst) + tmp, err := os.CreateTemp(filepath.Dir(dst), "."+base+".tmp-*") + if err != nil { + return "", fmt.Errorf("create temp file: %w", err) + } + tmpPath := tmp.Name() + removeTmp := true + defer func() { + if removeTmp { + _ = os.Remove(tmpPath) + } + }() + + digest := sha256.New() + if _, err := io.Copy(io.MultiWriter(tmp, digest), in); err != nil { + _ = tmp.Close() + return "", fmt.Errorf("copy file: %w", err) + } + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + return "", fmt.Errorf("sync temp file: %w", err) + } + if err := tmp.Close(); err != nil { + return "", fmt.Errorf("close temp file: %w", err) + } + if err := os.Chmod(tmpPath, perm); err != nil { + return "", fmt.Errorf("chmod temp file: %w", err) + } + if err := os.Rename(tmpPath, dst); err != nil { + return "", fmt.Errorf("install temp file: %w", err) + } + removeTmp = false + return hex.EncodeToString(digest.Sum(nil)), nil +} diff --git a/internal/audio/s3_audio_test.go b/internal/audio/s3_audio_test.go new file mode 100644 index 0000000..91d7f05 --- /dev/null +++ b/internal/audio/s3_audio_test.go @@ -0,0 +1,145 @@ +package audio + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" +) + +func TestMaterializeS3AudioCacheMissDownloadsAndPopulatesCache(t *testing.T) { + root := t.TempDir() + fake := &storage.FakeBackend{} + key := "dnd/campaigns/forsaken/sessions/2026-04-19/audio/alice.flac" + fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("audio")}) + req := testMaterializeRequest(t, root, fake, key, int64(len("audio"))) + + result, err := MaterializeS3Audio(context.Background(), req) + if err != nil { + t.Fatalf("MaterializeS3Audio() error = %v", err) + } + if !result.Downloaded || result.CacheHit { + t.Fatalf("result = %#v, want downloaded miss", result) + } + if len(fake.Downloads) != 1 { + t.Fatalf("downloads = %d, want 1", len(fake.Downloads)) + } + assertFileEquals(t, req.DestPath, "audio") + assertFileEquals(t, req.SpoolPath, "audio") + assertFileEquals(t, result.CachePath, "audio") + if result.Checksum == "" { + t.Fatalf("checksum is empty") + } +} + +func TestMaterializeS3AudioCacheHitSkipsDownload(t *testing.T) { + root := t.TempDir() + fake := &storage.FakeBackend{} + key := "dnd/campaigns/forsaken/sessions/2026-04-19/audio/alice.flac" + req := testMaterializeRequest(t, root, fake, key, int64(len("audio"))) + cachePath, err := artifacts.S3AudioCachePath(req.CacheRoot, req.Bucket, key) + if err != nil { + t.Fatalf("cache path: %v", err) + } + writeAudioTestFile(t, cachePath, "audio") + + result, err := MaterializeS3Audio(context.Background(), req) + if err != nil { + t.Fatalf("MaterializeS3Audio() error = %v", err) + } + if !result.CacheHit || result.Downloaded { + t.Fatalf("result = %#v, want cache hit", result) + } + if len(fake.Downloads) != 0 { + t.Fatalf("downloads = %d, want 0", len(fake.Downloads)) + } + assertFileEquals(t, req.DestPath, "audio") + if result.SpoolPath != "" { + t.Fatalf("spool path = %q, want empty on cache hit", result.SpoolPath) + } +} + +func TestMaterializeS3AudioInvalidCacheRefreshesFromS3(t *testing.T) { + root := t.TempDir() + fake := &storage.FakeBackend{} + key := "dnd/campaigns/forsaken/sessions/2026-04-19/audio/alice.flac" + fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("fresh-audio")}) + req := testMaterializeRequest(t, root, fake, key, int64(len("fresh-audio"))) + cachePath, err := artifacts.S3AudioCachePath(req.CacheRoot, req.Bucket, key) + if err != nil { + t.Fatalf("cache path: %v", err) + } + writeAudioTestFile(t, cachePath, "stale") + + result, err := MaterializeS3Audio(context.Background(), req) + if err != nil { + t.Fatalf("MaterializeS3Audio() error = %v", err) + } + if !result.Downloaded || result.CacheHit { + t.Fatalf("result = %#v, want refreshed miss", result) + } + if len(fake.Downloads) != 1 { + t.Fatalf("downloads = %d, want 1", len(fake.Downloads)) + } + assertFileEquals(t, req.DestPath, "fresh-audio") + assertFileEquals(t, cachePath, "fresh-audio") +} + +func TestMaterializeS3AudioDownloadFailureLeavesDestinationMissing(t *testing.T) { + root := t.TempDir() + fake := &storage.FakeBackend{DownloadErr: os.ErrPermission} + key := "dnd/campaigns/forsaken/sessions/2026-04-19/audio/alice.flac" + fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("audio")}) + req := testMaterializeRequest(t, root, fake, key, int64(len("audio"))) + + _, err := MaterializeS3Audio(context.Background(), req) + if err == nil || !strings.Contains(err.Error(), "download s3 audio object") { + t.Fatalf("error = %v, want download error", err) + } + assertMissing(t, req.DestPath) +} + +func testMaterializeRequest(t *testing.T, root string, fake *storage.FakeBackend, key string, size int64) S3MaterializeRequest { + t.Helper() + return S3MaterializeRequest{ + Store: fake, + Object: storage.ObjectInfo{Key: key, Size: size, ETag: "etag"}, + Bucket: "my-dnd-archive", + CacheRoot: filepath.Join(root, "cache"), + CacheEnabled: true, + SpoolPath: filepath.Join(root, "spool", "alice.flac"), + DestPath: filepath.Join(root, "work", "audio", "alice.flac"), + } +} + +func writeAudioTestFile(t *testing.T, path, contents string) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatalf("mkdir %q: %v", path, err) + } + if err := os.WriteFile(path, []byte(contents), 0o644); err != nil { + t.Fatalf("write %q: %v", path, err) + } +} + +func assertFileEquals(t *testing.T, path, want string) { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %q: %v", path, err) + } + if string(data) != want { + t.Fatalf("%q = %q, want %q", path, string(data), want) + } +} + +func assertMissing(t *testing.T, path string) { + t.Helper() + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("stat %q = %v, want not exists", path, err) + } +} diff --git a/internal/config/cache_config_test.go b/internal/config/cache_config_test.go new file mode 100644 index 0000000..9006341 --- /dev/null +++ b/internal/config/cache_config_test.go @@ -0,0 +1,67 @@ +package config + +import ( + "strings" + "testing" +) + +func TestCacheDefaults(t *testing.T) { + pipelinePath, sessionPath := writeConfigFiles(t, `workspace: + root: /tmp/narratio +whisperx: + transcribe_url: https://example.com/transcribe +analyzer: + timeout: 20m +notification: + timeout: 10s +`, `session_id: 2026-05-03 +inputs: + audio_dir: ./audio +`) + cfg, err := Load(pipelinePath, sessionPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if cfg.Pipeline.Cache.Root != DefaultCacheRoot { + t.Fatalf("cache.root = %q, want %q", cfg.Pipeline.Cache.Root, DefaultCacheRoot) + } + if cfg.Pipeline.Cache.S3Audio == nil || *cfg.Pipeline.Cache.S3Audio != DefaultCacheS3Audio { + t.Fatalf("cache.s3_audio = %v, want %v", cfg.Pipeline.Cache.S3Audio, DefaultCacheS3Audio) + } +} + +func TestCacheStrictDecodeRejectsUnknownFields(t *testing.T) { + pipelinePath, sessionPath := writeConfigFiles(t, `workspace: + root: /tmp/narratio +cache: + root: /var/cache/narratio + unknown: true +`, `session_id: 2026-05-03 +inputs: + audio_dir: ./audio +`) + _, err := Load(pipelinePath, sessionPath) + if err == nil || !strings.Contains(err.Error(), "strict decode failed") { + t.Fatalf("Load() error = %v, want strict decode failed", err) + } +} + +func TestCacheValidationRequiresRootWhenS3AudioEnabled(t *testing.T) { + pipelinePath, sessionPath := writeConfigFiles(t, `workspace: + root: /tmp/narratio +cache: + root: " " + s3_audio: true +`, `session_id: 2026-05-03 +inputs: + audio_dir: ./audio +`) + cfg, err := Load(pipelinePath, sessionPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + err = Validate(cfg) + if err == nil || !strings.Contains(err.Error(), "pipeline.cache.root is required") { + t.Fatalf("Validate() error = %v, want cache root error", err) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index e7ee159..11cbd1e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -19,6 +19,7 @@ type PipelineConfig struct { Workspace WorkspaceConfig `yaml:"workspace"` Storage StorageConfig `yaml:"storage"` Spool SpoolConfig `yaml:"spool"` + Cache CacheConfig `yaml:"cache"` Archive *ArchiveConfig `yaml:"archive"` Secrets *SecretsConfig `yaml:"secrets"` WhisperX WhisperXConfig `yaml:"whisperx"` @@ -90,6 +91,12 @@ type SpoolConfig struct { DeleteAudioAfterArchive bool `yaml:"delete_audio_after_archive"` } +// CacheConfig configures durable local caches for reusable remote inputs. +type CacheConfig struct { + Root string `yaml:"root"` + S3Audio *bool `yaml:"s3_audio"` +} + // ArchiveConfig configures archive behavior and artifact promotions. type ArchiveConfig struct { Enabled *bool `yaml:"enabled"` diff --git a/internal/config/defaults.go b/internal/config/defaults.go index ab6b48b..c93178f 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -14,6 +14,8 @@ const ( DefaultStorageS3RootPrefix = "dnd" DefaultWorkspaceRoot = "/var/lib/narratio" DefaultSpoolRoot = "/var/spool/narratio" + DefaultCacheRoot = "/var/cache/narratio" + DefaultCacheS3Audio = true DefaultWhisperXLanguage = "en" DefaultWhisperXTimeout = "30m" diff --git a/internal/config/load.go b/internal/config/load.go index 19c8f6e..298d058 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -361,6 +361,7 @@ func applyPipelineDefaults(cfg *PipelineConfig) { applyWorkspaceDefaults(&cfg.Workspace) applyStorageDefaults(&cfg.Storage) applySpoolDefaults(&cfg.Spool) + applyCacheDefaults(&cfg.Cache) applyArchiveDefaults(&cfg.Archive) applyWhisperXDefaults(&cfg.WhisperX) applySeriatimDefaults(&cfg.Seriatim) @@ -409,6 +410,18 @@ func applySpoolDefaults(cfg *SpoolConfig) { } } +func applyCacheDefaults(cfg *CacheConfig) { + if cfg == nil { + return + } + if cfg.Root == "" { + cfg.Root = DefaultCacheRoot + } + if cfg.S3Audio == nil { + cfg.S3Audio = boolPtr(DefaultCacheS3Audio) + } +} + func applyArchiveDefaults(cfg **ArchiveConfig) { if cfg == nil { return diff --git a/internal/config/validate.go b/internal/config/validate.go index 9df5306..5aba6ec 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -63,6 +63,9 @@ func validatePipeline(cfg *PipelineConfig) error { if err := validateSpool(cfg.Spool); err != nil { return err } + if err := validateCache(cfg.Cache); err != nil { + return err + } if err := validateArchive(cfg.Archive, cfg.Scriptorium); err != nil { return err } @@ -120,6 +123,13 @@ func validateSpool(cfg SpoolConfig) error { return nil } +func validateCache(cfg CacheConfig) error { + if cfg.S3Audio != nil && *cfg.S3Audio && strings.TrimSpace(cfg.Root) == "" { + return fmt.Errorf("pipeline.cache.root is required when pipeline.cache.s3_audio is true") + } + return nil +} + func validateArchive(cfg *ArchiveConfig, scriptorium *ScriptoriumConfig) error { if cfg == nil { return nil diff --git a/internal/manifest/manifest.go b/internal/manifest/manifest.go index 15326e8..65a2307 100644 --- a/internal/manifest/manifest.go +++ b/internal/manifest/manifest.go @@ -23,6 +23,7 @@ type InputRecord struct { S3Size int64 `json:"s3_size,omitempty"` S3ETag string `json:"s3_etag,omitempty"` SpoolPath string `json:"spool_path,omitempty"` + CachePath string `json:"cache_path,omitempty"` } // ArtifactRecord captures one produced artifact and optional remote metadata. diff --git a/internal/stage/prepare.go b/internal/stage/prepare.go index 7eb1269..bd395bb 100644 --- a/internal/stage/prepare.go +++ b/internal/stage/prepare.go @@ -13,6 +13,7 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/audio" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" "gopkg.in/yaml.v3" @@ -173,10 +174,13 @@ func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S registerConfigInput(cfgFile.kind, cfgFile.dst, checksum, cfgFile.source) } + var audioCacheStats s3AudioMaterializationStats if useS3Audio { - if err := materializeS3AudioInputs(ctx, env, m, sessionID, &inputs); err != nil { + stats, err := materializeS3AudioInputs(ctx, env, m, sessionID, &inputs) + if err != nil { return nil, fmt.Errorf("prepare: materialize s3 audio: %w", err) } + audioCacheStats = stats } else { if err := materializeLocalAudioInputs(env, paths, resolvedLocalAudio, registerInput); err != nil { return nil, fmt.Errorf("prepare: %w", err) @@ -211,6 +215,11 @@ func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S "inputs_count": len(inputs), "audio_files_resolved": countAudioInputs(inputs), } + if useS3Audio { + metadata["audio_cache_hits"] = audioCacheStats.CacheHits + metadata["audio_cache_misses"] = audioCacheStats.CacheMisses + metadata["audio_s3_downloads"] = audioCacheStats.Downloads + } if len(previousRequirements) > 0 { metadata["previous_requirements_count"] = len(previousRequirements) if previousHydration != nil { @@ -345,28 +354,34 @@ func materializeLocalAudioInputs(env *Env, paths artifacts.SessionPaths, resolve return nil } -func materializeS3AudioInputs(ctx context.Context, env *Env, m *manifest.Manifest, sessionID string, inputs *[]manifest.InputRecord) error { +type s3AudioMaterializationStats struct { + CacheHits int + CacheMisses int + Downloads int +} + +func materializeS3AudioInputs(ctx context.Context, env *Env, m *manifest.Manifest, sessionID string, inputs *[]manifest.InputRecord) (s3AudioMaterializationStats, error) { if env.ObjectStore == nil { - return fmt.Errorf("s3 audio input requires object store backend") + return s3AudioMaterializationStats{}, fmt.Errorf("s3 audio input requires object store backend") } if env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil || env.Config.Pipeline.Storage.S3 == nil || env.Config.Session.Inputs.AudioS3 == nil { - return fmt.Errorf("s3 audio input requires pipeline.storage.s3 and session.inputs.audio_s3 configuration") + return s3AudioMaterializationStats{}, fmt.Errorf("s3 audio input requires pipeline.storage.s3 and session.inputs.audio_s3 configuration") } campaign := strings.TrimSpace(env.Config.Session.Campaign) if campaign == "" { - return fmt.Errorf("session campaign is required for s3 audio input") + return s3AudioMaterializationStats{}, fmt.Errorf("session campaign is required for s3 audio input") } runID := strings.TrimSpace(m.RunID) if runID == "" { - return fmt.Errorf("run id is required for s3 audio input") + return s3AudioMaterializationStats{}, fmt.Errorf("run id is required for s3 audio input") } sessionPrefix := artifacts.S3SessionPrefix(env.Config.Pipeline.Storage.S3.RootPrefix, campaign, sessionID) audioPrefix := artifacts.S3AudioPrefix(sessionPrefix, env.Config.Session.Inputs.AudioS3.Prefix) objects, err := env.ObjectStore.List(ctx, audioPrefix) if err != nil { - return fmt.Errorf("list s3 audio objects under %q: %w", audioPrefix, err) + return s3AudioMaterializationStats{}, fmt.Errorf("list s3 audio objects under %q: %w", audioPrefix, err) } audioObjects := make([]storage.ObjectInfo, 0, len(objects)) @@ -384,7 +399,7 @@ func materializeS3AudioInputs(ctx context.Context, env *Env, m *manifest.Manifes return audioObjects[i].Key < audioObjects[j].Key }) if len(audioObjects) == 0 { - return fmt.Errorf("no .flac files found under s3 audio prefix %q", audioPrefix) + return s3AudioMaterializationStats{}, fmt.Errorf("no .flac files found under s3 audio prefix %q", audioPrefix) } spoolAudioDir := strings.TrimSpace(m.LocalSpoolDir) @@ -394,45 +409,60 @@ func materializeS3AudioInputs(ctx context.Context, env *Env, m *manifest.Manifes workAudioDir := filepath.Join(pathsWorkDirForManifest(env, m, sessionID), "audio") if err := os.MkdirAll(spoolAudioDir, 0o755); err != nil { - return fmt.Errorf("create spool audio directory %q: %w", spoolAudioDir, err) + return s3AudioMaterializationStats{}, fmt.Errorf("create spool audio directory %q: %w", spoolAudioDir, err) } if err := os.MkdirAll(workAudioDir, 0o755); err != nil { - return fmt.Errorf("create work audio directory %q: %w", workAudioDir, err) + return s3AudioMaterializationStats{}, fmt.Errorf("create work audio directory %q: %w", workAudioDir, err) } seenBase := map[string]string{} + stats := s3AudioMaterializationStats{} + cacheEnabled := env.Config.Pipeline.Cache.S3Audio == nil || *env.Config.Pipeline.Cache.S3Audio for _, obj := range audioObjects { base := path.Base(obj.Key) if prev, exists := seenBase[base]; exists && prev != obj.Key { - return fmt.Errorf("duplicate s3 audio basename %q from %q and %q", base, prev, obj.Key) + return s3AudioMaterializationStats{}, fmt.Errorf("duplicate s3 audio basename %q from %q and %q", base, prev, obj.Key) } seenBase[base] = obj.Key spoolPath := filepath.Join(spoolAudioDir, base) - if err := env.ObjectStore.Download(ctx, obj.Key, spoolPath); err != nil { - return fmt.Errorf("download s3 audio object %q: %w", obj.Key, err) - } - workPath := filepath.Join(workAudioDir, base) - checksum, err := copyFileIfChanged(env.ArtifactStore, spoolPath, workPath) + result, err := audio.MaterializeS3Audio(ctx, audio.S3MaterializeRequest{ + Store: env.ObjectStore, + Object: obj, + Bucket: strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket), + CacheRoot: strings.TrimSpace(env.Config.Pipeline.Cache.Root), + CacheEnabled: cacheEnabled, + SpoolPath: spoolPath, + DestPath: workPath, + }) if err != nil { - return fmt.Errorf("materialize downloaded audio %q: %w", base, err) + return s3AudioMaterializationStats{}, err + } + if result.CacheHit { + stats.CacheHits++ + } else { + stats.CacheMisses++ + } + if result.Downloaded { + stats.Downloads++ } *inputs = append(*inputs, manifest.InputRecord{ Kind: "audio", Path: workPath, - Checksum: checksum, + Checksum: result.Checksum, Source: "s3", S3Bucket: strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket), S3Key: obj.Key, S3Size: obj.Size, S3ETag: obj.ETag, - SpoolPath: spoolPath, + SpoolPath: result.SpoolPath, + CachePath: result.CachePath, }) } - return nil + return stats, nil } func countAudioInputs(inputs []manifest.InputRecord) int { diff --git a/internal/stage/prepare_test.go b/internal/stage/prepare_test.go index 7b6ff81..20d5891 100644 --- a/internal/stage/prepare_test.go +++ b/internal/stage/prepare_test.go @@ -245,6 +245,15 @@ func TestPrepareStageS3AudioDownloadAndMaterialization(t *testing.T) { t.Fatalf("expected file %q: %v", p, err) } } + if got := result.Metadata["audio_cache_hits"]; got != 0 { + t.Fatalf("audio_cache_hits = %#v, want 0", got) + } + if got := result.Metadata["audio_cache_misses"]; got != 2 { + t.Fatalf("audio_cache_misses = %#v, want 2", got) + } + if got := result.Metadata["audio_s3_downloads"]; got != 2 { + t.Fatalf("audio_s3_downloads = %#v, want 2", got) + } audioInputs := 0 for _, in := range m.Inputs { @@ -258,15 +267,75 @@ func TestPrepareStageS3AudioDownloadAndMaterialization(t *testing.T) { if in.S3Bucket != "my-dnd-archive" { t.Fatalf("audio input bucket = %q", in.S3Bucket) } - if in.S3Key == "" || in.SpoolPath == "" || in.Checksum == "" { + if in.S3Key == "" || in.SpoolPath == "" || in.CachePath == "" || in.Checksum == "" { t.Fatalf("audio input missing provenance: %#v", in) } + if _, err := os.Stat(in.CachePath); err != nil { + t.Fatalf("expected cache path %q: %v", in.CachePath, err) + } } if audioInputs != 2 { t.Fatalf("audio input count = %d, want 2", audioInputs) } } +func TestPrepareStageS3AudioUsesCacheOnRerun(t *testing.T) { + env, m := setupPrepareEnv(t) + env.Config.Session.Campaign = "forsaken" + env.Config.Session.Inputs.AudioDir = "" + env.Config.Session.Inputs.AudioFiles = nil + env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "audio/"} + env.Config.Pipeline.Spool = config.SpoolConfig{Root: filepath.Join(t.TempDir(), "spool")} + env.Config.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "my-dnd-archive", RootPrefix: "dnd"} + m.RunID = "20260515T031522Z-a1b2c3d4" + m.LocalWorkDir = artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, "forsaken", m.SessionID, m.RunID) + m.LocalSpoolDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, "forsaken", m.SessionID, m.RunID) + + fake := &storage.FakeBackend{} + fake.SeedObject(storage.FakeObject{Key: "dnd/campaigns/forsaken/sessions/2026-05-03/audio/alice.flac", Data: []byte("alice")}) + env.ObjectStore = fake + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("first prepare.Run() error = %v", err) + } + if len(fake.Downloads) != 1 { + t.Fatalf("downloads = %d, want 1", len(fake.Downloads)) + } + + if err := os.RemoveAll(filepath.Join(m.LocalWorkDir, "audio")); err != nil { + t.Fatalf("remove work audio: %v", err) + } + if err := os.RemoveAll(m.LocalSpoolDir); err != nil { + t.Fatalf("remove spool audio: %v", err) + } + fake.DownloadErr = os.ErrPermission + + m2 := manifest.New("2026-05-03", time.Now().UTC()) + m2.RunID = "20260515T041522Z-a1b2c3d4" + m2.LocalWorkDir = artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, "forsaken", m2.SessionID, m2.RunID) + m2.LocalSpoolDir = artifacts.SessionSpoolAudioDir(env.Config.Pipeline.Spool.Root, "forsaken", m2.SessionID, m2.RunID) + result, err := (prepareStage{}).Run(context.Background(), env, m2) + if err != nil { + t.Fatalf("cached prepare.Run() error = %v", err) + } + if len(fake.Downloads) != 1 { + t.Fatalf("downloads = %d, want cached rerun to avoid new download", len(fake.Downloads)) + } + if got := result.Metadata["audio_cache_hits"]; got != 1 { + t.Fatalf("audio_cache_hits = %#v, want 1", got) + } + if got := result.Metadata["audio_s3_downloads"]; got != 0 { + t.Fatalf("audio_s3_downloads = %#v, want 0", got) + } + audioInput := findManifestInput(t, m2.Inputs, "audio") + if audioInput.CachePath == "" { + t.Fatalf("audio input missing cache path: %#v", audioInput) + } + if audioInput.SpoolPath != "" { + t.Fatalf("audio input spool path = %q, want empty on cache hit", audioInput.SpoolPath) + } + mustReadFileEquals(t, filepath.Join(m2.LocalWorkDir, "audio", "alice.flac"), "alice") +} + func TestPrepareStageS3AudioFailures(t *testing.T) { tests := []struct { name string @@ -551,7 +620,10 @@ inputs: writeFile(t, filepath.Join(cfgDir, "glossary.yml"), "[]\n") cfg := &config.Config{ - Pipeline: &config.PipelineConfig{Workspace: config.WorkspaceConfig{Root: workspace}}, + Pipeline: &config.PipelineConfig{ + Workspace: config.WorkspaceConfig{Root: workspace}, + Cache: config.CacheConfig{Root: filepath.Join(t.TempDir(), "cache"), S3Audio: boolPtr(true)}, + }, Campaign: &config.CampaignConfig{Campaign: "sample-campaign"}, SessionPath: sessionPath, CampaignPath: campaignPath, @@ -616,3 +688,14 @@ func findManifestInput(t *testing.T, inputs []manifest.InputRecord, kind string) t.Fatalf("manifest input kind %q not found in %#v", kind, inputs) return manifest.InputRecord{} } + +func mustReadFileEquals(t *testing.T, path, want string) { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %q: %v", path, err) + } + if string(data) != want { + t.Fatalf("%q = %q, want %q", path, string(data), want) + } +}