eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 27 Wiki Activity

Decouple publish from run manifests

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-03 04:00:25 +00:00
parent 49324d3f35
commit 6d748d65d3
5 changed files with 118 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
internal/stage/publish.go
View File

@@ -75,17 +75,9 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
return nil, fmt.Errorf("publish: remote object store backend is required when publish run upload is enabled")
}
runManifestPath, err := resolvePublishRunManifestPath(env, m)
runRoot, err := resolvePublishRunRoot(env, m)
if err != nil {
return nil, fmt.Errorf("publish: resolve run manifest path: %w", err)
}
runRoot := filepath.Dir(runManifestPath)
runRootInfo, err := os.Lstat(runRoot)
if err != nil {
return nil, fmt.Errorf("publish: run root %q: %w", runRoot, err)
}
if runRootInfo.Mode()&os.ModeSymlink != 0 || !runRootInfo.IsDir() {
return nil, fmt.Errorf("publish: run root %q is not a directory", runRoot)
return nil, fmt.Errorf("publish: resolve run root: %w", err)
}
runPrefix, err := artifacts.ResolvePublishRunPrefix(env.Config, m)
@@ -105,24 +97,11 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
return nil, fmt.Errorf("publish: run id is required")
}
runManifestFile, err := openPublishOutputSource(filepath.Join(runRoot, filepath.Base(runManifestPath)))
if err != nil {
return nil, fmt.Errorf("publish: open run manifest: %w", err)
}
defer runManifestFile.Close()
runManifest, err := (&manifest.LocalStore{}).LoadRunReader(ctx, runManifestFile)
if err != nil {
return nil, fmt.Errorf("publish: load run manifest: %w", err)
}
if _, err := runManifestFile.Seek(0, io.SeekStart); err != nil {
return nil, fmt.Errorf("publish: rewind run manifest: %w", err)
}
runFiles, err := collectPublishRunFiles(runRoot, runManifest)
runFiles, err := collectPublishRunFiles(runRoot, m)
if err != nil {
return nil, fmt.Errorf("publish: collect run files: %w", err)
}
runSources, err := openPublishRunFiles(runRoot, runFiles, runManifestFile)
runSources, err := openPublishRunFiles(runRoot, runFiles)
if err != nil {
return nil, fmt.Errorf("publish: verify run files: %w", err)
}
@@ -139,7 +118,7 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if previousManifestFile != nil {
defer previousManifestFile.Close()
}
previousSources, err := openPublishRunFiles(sessionPaths.PreviousDir, previousFiles, previousManifestFile)
previousSources, err := openPublishRunFiles(sessionPaths.PreviousDir, previousFiles)
if err != nil {
return nil, fmt.Errorf("publish: verify previous files: %w", err)
}
@@ -333,15 +312,9 @@ func buildPublishTransferPlan(
return nil
}
for index, file := range runFiles {
artifactType := artifacts.RemoteArtifactTypeRunFile
source := "run:" + file.RelativePath
if file.RelativePath == config.S3ManifestFile {
artifactType = artifacts.RemoteArtifactTypeRunManifest
source = "run.manifest"
}
if err := add(artifacts.RemoteArtifact{
Type: artifactType,
Source: source,
Type: artifacts.RemoteArtifactTypeRunFile,
Source: "run:" + file.RelativePath,
DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
}, runSources[index]); err != nil {
closePublishTransfers(transfers)
@@ -638,7 +611,7 @@ func validatePublishPrerequisites(m *manifest.Manifest) error {
return nil
}
func resolvePublishRunManifestPath(env *Env, m *manifest.Manifest) (string, error) {
func resolvePublishRunRoot(env *Env, m *manifest.Manifest) (string, error) {
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
if sessionID == "" && m != nil {
sessionID = strings.TrimSpace(m.SessionID)
@@ -658,13 +631,13 @@ func resolvePublishRunManifestPath(env *Env, m *manifest.Manifest) (string, erro
return "", fmt.Errorf("run id is required")
}
canonical := filepath.Clean(artifacts.SessionRunManifestPathForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
canonicalExists, err := directoryExists(filepath.Dir(canonical))
canonical := filepath.Clean(artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
info, err := os.Lstat(canonical)
if err != nil {
return "", fmt.Errorf("check canonical run manifest directory %q: %w", filepath.Dir(canonical), err)
return "", fmt.Errorf("inspect canonical run root %q: %w", canonical, err)
}
if !canonicalExists {
return "", fmt.Errorf("run manifest directory not found for campaign %q session %q run %q at canonical path %q", campaign, sessionID, runID, filepath.Dir(canonical))
if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
return "", fmt.Errorf("canonical run root %q is not a directory", canonical)
}
return canonical, nil
}
@@ -850,13 +823,11 @@ func buildPublishRuntimeArtifactCatalog(
return catalog, nil
}
func collectPublishRunFiles(runRoot string, runManifest *manifest.RunManifest) ([]publishUploadFile, error) {
if runManifest == nil {
return nil, fmt.Errorf("run manifest is required")
}
files := map[string]publishUploadFile{
"manifest.json": {RelativePath: "manifest.json", SourceRelativePath: "manifest.json"},
func collectPublishRunFiles(runRoot string, sessionManifest *manifest.Manifest) ([]publishUploadFile, error) {
if sessionManifest == nil {
return nil, fmt.Errorf("session manifest is required")
}
files := map[string]publishUploadFile{}
add := func(path, checksum string) error {
if strings.TrimSpace(path) == "" {
return nil
@@ -875,7 +846,7 @@ func collectPublishRunFiles(runRoot string, runManifest *manifest.RunManifest) (
files[rel] = candidate
return nil
}
for _, record := range runManifest.Stages {
for _, record := range sessionManifest.Stages {
if record == nil {
continue
}
@@ -1001,26 +972,16 @@ func collectPublishPreviousFiles(ctx context.Context, previousDir string) ([]pub
return out, manifestFile, nil
}
func openPublishRunFiles(runRoot string, files []publishUploadFile, manifestFile *os.File) ([]*os.File, error) {
func openPublishRunFiles(runRoot string, files []publishUploadFile) ([]*os.File, error) {
sources := make([]*os.File, 0, len(files))
for _, candidate := range files {
if manifestFile == nil {
file, err := openPublishOutputSource(filepath.Join(runRoot, filepath.FromSlash(candidate.SourceRelativePath)))
if err != nil {
closePublishSources(sources)
return nil, fmt.Errorf("archive manifest is required")
}
file := manifestFile
if candidate.SourceRelativePath != "manifest.json" {
var err error
file, err = openPublishOutputSource(filepath.Join(runRoot, filepath.FromSlash(candidate.SourceRelativePath)))
if err != nil {
closePublishSources(sources)
return nil, err
}
return nil, err
}
if err := verifyPublishSourceChecksum(file, candidate.RelativePath, candidate.Checksum); err != nil {
if file != manifestFile {
_ = file.Close()
}
_ = file.Close()
closePublishSources(sources)
return nil, err
}
@@ -1052,17 +1013,6 @@ func closePublishSources(sources []*os.File) {
}
}
func directoryExists(path string) (bool, error) {
info, err := os.Stat(path)
if err == nil {
return info.IsDir(), nil
}
if os.IsNotExist(err) {
return false, nil
}
return false, err
}
func writeCommittedSessionManifest(m *manifest.Manifest, publishMetadata map[string]any) (string, error) {
if m == nil {
return "", fmt.Errorf("manifest is required")