Applied a more general bugfix to path-resolution issues in the archive stage
This commit is contained in:
@@ -125,7 +125,8 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: collect run files: %w", err)
|
return nil, fmt.Errorf("archive: collect run files: %w", err)
|
||||||
}
|
}
|
||||||
promotions, err := resolveArchivePromotions(workDir, env.Config.Pipeline.Archive.PromoteArtifacts)
|
workDirCandidates := archiveSourceWorkDirs(env, m, workDir)
|
||||||
|
promotions, err := resolveArchivePromotions(workDirCandidates, env.Config.Pipeline.Archive.PromoteArtifacts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
||||||
}
|
}
|
||||||
@@ -330,22 +331,38 @@ func archiveBucket(env *Env, m *manifest.Manifest) string {
|
|||||||
return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket)
|
return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket)
|
||||||
}
|
}
|
||||||
|
|
||||||
func resolveArchivePromotions(workDir string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
|
func resolveArchivePromotions(workDirs []string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
|
||||||
|
if len(workDirs) == 0 {
|
||||||
|
return nil, fmt.Errorf("at least one workdir candidate is required")
|
||||||
|
}
|
||||||
out := make([]archivePromotion, 0, len(rules))
|
out := make([]archivePromotion, 0, len(rules))
|
||||||
for _, rule := range rules {
|
for _, rule := range rules {
|
||||||
from := strings.TrimSpace(rule.From)
|
from := strings.TrimSpace(rule.From)
|
||||||
to := strings.TrimSpace(rule.To)
|
to := strings.TrimSpace(rule.To)
|
||||||
required := rule.Required == nil || *rule.Required
|
required := rule.Required == nil || *rule.Required
|
||||||
|
|
||||||
localPath, err := resolveWorkDirRelativePath(workDir, from)
|
var (
|
||||||
|
localPath string
|
||||||
|
exists bool
|
||||||
|
)
|
||||||
|
for _, candidateRoot := range workDirs {
|
||||||
|
resolvedPath, err := resolveWorkDirRelativePath(candidateRoot, from)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("promotion from %q: %w", from, err)
|
return nil, fmt.Errorf("promotion from %q: %w", from, err)
|
||||||
}
|
}
|
||||||
info, err := os.Stat(localPath)
|
info, err := os.Stat(resolvedPath)
|
||||||
exists := err == nil && !info.IsDir()
|
if err == nil && !info.IsDir() {
|
||||||
|
localPath = resolvedPath
|
||||||
|
exists = true
|
||||||
|
break
|
||||||
|
}
|
||||||
if err != nil && !os.IsNotExist(err) {
|
if err != nil && !os.IsNotExist(err) {
|
||||||
return nil, fmt.Errorf("promotion source %q: %w", from, err)
|
return nil, fmt.Errorf("promotion source %q: %w", from, err)
|
||||||
}
|
}
|
||||||
|
if localPath == "" {
|
||||||
|
localPath = resolvedPath
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
out = append(out, archivePromotion{
|
out = append(out, archivePromotion{
|
||||||
From: from,
|
From: from,
|
||||||
@@ -358,6 +375,33 @@ func resolveArchivePromotions(workDir string, rules []config.ArchivePromotionRul
|
|||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func archiveSourceWorkDirs(env *Env, m *manifest.Manifest, runWorkDir string) []string {
|
||||||
|
candidates := make([]string, 0, 2)
|
||||||
|
if strings.TrimSpace(runWorkDir) != "" {
|
||||||
|
candidates = append(candidates, filepath.Clean(runWorkDir))
|
||||||
|
}
|
||||||
|
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
||||||
|
if sessionID == "" && m != nil {
|
||||||
|
sessionID = strings.TrimSpace(m.SessionID)
|
||||||
|
}
|
||||||
|
if sessionID != "" {
|
||||||
|
sessionWorkDir := filepath.Clean(artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID))
|
||||||
|
if sessionWorkDir != "" {
|
||||||
|
candidates = append(candidates, sessionWorkDir)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
seen := make(map[string]struct{}, len(candidates))
|
||||||
|
out := make([]string, 0, len(candidates))
|
||||||
|
for _, c := range candidates {
|
||||||
|
if _, ok := seen[c]; ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
seen[c] = struct{}{}
|
||||||
|
out = append(out, c)
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
|
func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
|
||||||
rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel)))
|
rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel)))
|
||||||
if rel == "." || rel == "" {
|
if rel == "." || rel == "" {
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package stage
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"reflect"
|
"reflect"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -182,6 +183,29 @@ func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestArchivePromotionFallsBackToSessionWorkDir(t *testing.T) {
|
||||||
|
env, m, runWorkDir := archiveFixture(t)
|
||||||
|
sessionWorkDir := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, m.SessionID)
|
||||||
|
sessionTrimmed := filepath.Join(sessionWorkDir, "transcripts", "trimmed.json")
|
||||||
|
|
||||||
|
if err := os.Remove(filepath.Join(runWorkDir, "transcripts", "trimmed.json")); err != nil {
|
||||||
|
t.Fatalf("remove run scoped trimmed transcript: %v", err)
|
||||||
|
}
|
||||||
|
writeStageTestFile(t, sessionTrimmed, "{}\n")
|
||||||
|
|
||||||
|
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if result.Metadata["promoted_files_uploaded"] != 2 {
|
||||||
|
t.Fatalf("metadata promoted_files_uploaded = %#v, want 2", result.Metadata["promoted_files_uploaded"])
|
||||||
|
}
|
||||||
|
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||||
|
if _, ok := fake.Objects[m.S3SessionPrefix+"transcripts/trimmed.json"]; !ok {
|
||||||
|
t.Fatalf("missing promoted trimmed key from session fallback")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestArchiveDoesNotWriteCurrentPointerWhenPromotionUploadFails(t *testing.T) {
|
func TestArchiveDoesNotWriteCurrentPointerWhenPromotionUploadFails(t *testing.T) {
|
||||||
env, m, _ := archiveFixture(t)
|
env, m, _ := archiveFixture(t)
|
||||||
fake := env.ObjectStore.(*storage.FakeBackend)
|
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||||
|
|||||||
Reference in New Issue
Block a user