Archive durable previous-session cache files with session state
This commit is contained in:
@@ -118,6 +118,10 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
return nil, fmt.Errorf("archive: collect run files: %w", err)
|
return nil, fmt.Errorf("archive: collect run files: %w", err)
|
||||||
}
|
}
|
||||||
sessionPaths := archiveSessionPaths(env, m)
|
sessionPaths := archiveSessionPaths(env, m)
|
||||||
|
previousFiles, err := collectArchivePreviousFiles(sessionPaths.PreviousDir)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("archive: collect previous files: %w", err)
|
||||||
|
}
|
||||||
runtimeCatalog, err := buildArchiveRuntimeArtifactCatalog(sessionPaths, env.Config.Pipeline.Scriptorium)
|
runtimeCatalog, err := buildArchiveRuntimeArtifactCatalog(sessionPaths, env.Config.Pipeline.Scriptorium)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err)
|
return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err)
|
||||||
@@ -144,6 +148,15 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
promotedUploaded = append(promotedUploaded, promotion.Dest)
|
promotedUploaded = append(promotedUploaded, promotion.Dest)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
previousUploaded := make([]string, 0, len(previousFiles))
|
||||||
|
for _, file := range previousFiles {
|
||||||
|
key := artifacts.S3PromotedArtifactKey(sessionPrefix, file.RelativePath)
|
||||||
|
if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil {
|
||||||
|
return nil, fmt.Errorf("archive: upload previous file %q to %q: %w", file.RelativePath, key, err)
|
||||||
|
}
|
||||||
|
previousUploaded = append(previousUploaded, file.RelativePath)
|
||||||
|
}
|
||||||
|
|
||||||
currentManifestKey, currentRunPointerKey := artifacts.ResolveArchiveCurrentStateKeys(sessionPrefix)
|
currentManifestKey, currentRunPointerKey := artifacts.ResolveArchiveCurrentStateKeys(sessionPrefix)
|
||||||
manifestTempPath, err := writeCurrentManifestSnapshot(m, archiveMetadataPreview(
|
manifestTempPath, err := writeCurrentManifestSnapshot(m, archiveMetadataPreview(
|
||||||
bucket,
|
bucket,
|
||||||
@@ -151,6 +164,7 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
sessionPrefix,
|
sessionPrefix,
|
||||||
runUploaded,
|
runUploaded,
|
||||||
promotedUploaded,
|
promotedUploaded,
|
||||||
|
previousUploaded,
|
||||||
skippedOptional,
|
skippedOptional,
|
||||||
currentManifestKey,
|
currentManifestKey,
|
||||||
))
|
))
|
||||||
@@ -187,6 +201,8 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
"run_uploaded_paths": runUploaded,
|
"run_uploaded_paths": runUploaded,
|
||||||
"promoted_files_uploaded": len(promotedUploaded),
|
"promoted_files_uploaded": len(promotedUploaded),
|
||||||
"promoted_paths": promotedUploaded,
|
"promoted_paths": promotedUploaded,
|
||||||
|
"previous_files_uploaded": len(previousUploaded),
|
||||||
|
"previous_uploaded_paths": previousUploaded,
|
||||||
"skipped_optional_promotions": skippedOptional,
|
"skipped_optional_promotions": skippedOptional,
|
||||||
"current_manifest_key": currentManifestKey,
|
"current_manifest_key": currentManifestKey,
|
||||||
"current_run_id_key": currentRunPointerKey,
|
"current_run_id_key": currentRunPointerKey,
|
||||||
@@ -497,6 +513,51 @@ func collectArchiveRunFiles(runRoot, manifestPath string) ([]archiveUploadFile,
|
|||||||
return files, nil
|
return files, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func collectArchivePreviousFiles(previousDir string) ([]archiveUploadFile, error) {
|
||||||
|
previousDir = filepath.Clean(strings.TrimSpace(previousDir))
|
||||||
|
if previousDir == "" {
|
||||||
|
return nil, fmt.Errorf("previous directory is required")
|
||||||
|
}
|
||||||
|
info, err := os.Stat(previousDir)
|
||||||
|
if err != nil {
|
||||||
|
if os.IsNotExist(err) {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("stat %q: %w", previousDir, err)
|
||||||
|
}
|
||||||
|
if !info.IsDir() {
|
||||||
|
return nil, fmt.Errorf("previous path %q is not a directory", previousDir)
|
||||||
|
}
|
||||||
|
|
||||||
|
files := make([]archiveUploadFile, 0, 16)
|
||||||
|
err = filepath.WalkDir(previousDir, func(path string, d fs.DirEntry, walkErr error) error {
|
||||||
|
if walkErr != nil {
|
||||||
|
return walkErr
|
||||||
|
}
|
||||||
|
if d.IsDir() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
rel, err := filepath.Rel(previousDir, path)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("relative path from %q to %q: %w", previousDir, path, err)
|
||||||
|
}
|
||||||
|
rel = filepath.ToSlash(rel)
|
||||||
|
files = append(files, archiveUploadFile{
|
||||||
|
RelativePath: filepath.ToSlash(filepath.Join(config.PathPreviousDirSegment, rel)),
|
||||||
|
LocalPath: path,
|
||||||
|
})
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("walk %q: %w", previousDir, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
sort.Slice(files, func(i, j int) bool {
|
||||||
|
return files[i].RelativePath < files[j].RelativePath
|
||||||
|
})
|
||||||
|
return files, nil
|
||||||
|
}
|
||||||
|
|
||||||
func resolveArchiveRunManifestSource(runRoot string) (string, error) {
|
func resolveArchiveRunManifestSource(runRoot string) (string, error) {
|
||||||
path := filepath.Join(filepath.Clean(runRoot), "manifest.json")
|
path := filepath.Join(filepath.Clean(runRoot), "manifest.json")
|
||||||
info, err := os.Stat(path)
|
info, err := os.Stat(path)
|
||||||
@@ -600,6 +661,7 @@ func archiveMetadataPreview(
|
|||||||
bucket, runPrefix, sessionPrefix string,
|
bucket, runPrefix, sessionPrefix string,
|
||||||
runUploaded []string,
|
runUploaded []string,
|
||||||
promotedUploaded []string,
|
promotedUploaded []string,
|
||||||
|
previousUploaded []string,
|
||||||
skippedOptional []string,
|
skippedOptional []string,
|
||||||
currentManifestKey string,
|
currentManifestKey string,
|
||||||
) map[string]any {
|
) map[string]any {
|
||||||
@@ -612,6 +674,8 @@ func archiveMetadataPreview(
|
|||||||
"run_uploaded_paths": append([]string(nil), runUploaded...),
|
"run_uploaded_paths": append([]string(nil), runUploaded...),
|
||||||
"promoted_files_uploaded": len(promotedUploaded),
|
"promoted_files_uploaded": len(promotedUploaded),
|
||||||
"promoted_paths": append([]string(nil), promotedUploaded...),
|
"promoted_paths": append([]string(nil), promotedUploaded...),
|
||||||
|
"previous_files_uploaded": len(previousUploaded),
|
||||||
|
"previous_uploaded_paths": append([]string(nil), previousUploaded...),
|
||||||
"skipped_optional_promotions": append([]string(nil), skippedOptional...),
|
"skipped_optional_promotions": append([]string(nil), skippedOptional...),
|
||||||
"current_manifest_key": currentManifestKey,
|
"current_manifest_key": currentManifestKey,
|
||||||
"current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix),
|
"current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix),
|
||||||
|
|||||||
@@ -130,6 +130,50 @@ func TestArchiveUploadsRunRecordPromotionsAndCurrentPointer(t *testing.T) {
|
|||||||
if result.Metadata["promoted_files_uploaded"] != 2 {
|
if result.Metadata["promoted_files_uploaded"] != 2 {
|
||||||
t.Fatalf("metadata promoted_files_uploaded = %#v, want 2", result.Metadata["promoted_files_uploaded"])
|
t.Fatalf("metadata promoted_files_uploaded = %#v, want 2", result.Metadata["promoted_files_uploaded"])
|
||||||
}
|
}
|
||||||
|
if result.Metadata["previous_files_uploaded"] != 0 {
|
||||||
|
t.Fatalf("metadata previous_files_uploaded = %#v, want 0", result.Metadata["previous_files_uploaded"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveUploadsPreviousCacheWhenPresent(t *testing.T) {
|
||||||
|
env, m, _ := archiveFixture(t)
|
||||||
|
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||||
|
sessionRoot := artifacts.SessionWorkDirForCampaign(
|
||||||
|
env.Config.Pipeline.Workspace.Root,
|
||||||
|
env.Config.Session.Campaign,
|
||||||
|
env.Config.Session.SessionID,
|
||||||
|
)
|
||||||
|
writeStageTestFile(t, filepath.Join(sessionRoot, "previous", "manifest.json"), "{\"session_id\":\"2026-04-12\"}\n")
|
||||||
|
writeStageTestFile(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n")
|
||||||
|
|
||||||
|
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
previousManifestKey := m.S3SessionPrefix + "previous/manifest.json"
|
||||||
|
previousRecapKey := m.S3SessionPrefix + "previous/artifacts/session_recap.md"
|
||||||
|
if _, ok := fake.Objects[previousManifestKey]; !ok {
|
||||||
|
t.Fatalf("missing archived previous manifest key %q", previousManifestKey)
|
||||||
|
}
|
||||||
|
if _, ok := fake.Objects[previousRecapKey]; !ok {
|
||||||
|
t.Fatalf("missing archived previous artifact key %q", previousRecapKey)
|
||||||
|
}
|
||||||
|
if result.Metadata["previous_files_uploaded"] != 2 {
|
||||||
|
t.Fatalf("metadata previous_files_uploaded = %#v, want 2", result.Metadata["previous_files_uploaded"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveToleratesMissingPreviousCache(t *testing.T) {
|
||||||
|
env, m, _ := archiveFixture(t)
|
||||||
|
|
||||||
|
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if result.Metadata["previous_files_uploaded"] != 0 {
|
||||||
|
t.Fatalf("metadata previous_files_uploaded = %#v, want 0", result.Metadata["previous_files_uploaded"])
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestArchiveUsesCustomPromotionRules(t *testing.T) {
|
func TestArchiveUsesCustomPromotionRules(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user