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

Integrate artifact locks into publish

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-05 13:04:08 +00:00
parent b74e31e1cc
commit 4bc33773fa
13 changed files with 503 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
internal/stage/publish.go
View File

@@ -114,26 +114,45 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if err != nil {
return nil, fmt.Errorf("publish: build runtime artifact catalog: %w", err)
}
publishOutputs, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, err := resolvePublishOutputs(
sessionPaths,
m,
publishRules, err := resolvePublishRules(
runtimeCatalog,
env.Config.Pipeline.Publish.Outputs,
env.Config.Pipeline.Publish.Locks,
env.SelectedArtifactKeys,
runPrefix,
)
if err != nil {
return nil, fmt.Errorf("publish: resolve publish output rules: %w", err)
}
initialLocks, err := readPublishOutputLocks(ctx, env.ObjectStore, sessionPrefix, publishRules)
if err != nil {
return nil, fmt.Errorf("publish: read publish output locks: %w", err)
}
var priorCommit *artifacts.RemoteCommitManifest
if publishRulesNeedPriorCommit(publishRules, initialLocks) {
priorCommit, err = loadPriorPublishCommit(ctx, env.ObjectStore, sessionPrefix, m.Campaign, m.SessionID)
if err != nil {
return nil, fmt.Errorf("publish: load prior current commit: %w", err)
}
}
discoveredOutputs, err := discoverPublishOutputs(sessionPaths, m, runtimeCatalog, publishRules, initialLocks)
if err != nil {
return nil, fmt.Errorf("publish: discover publish outputs: %w", err)
}
outputPlan, err := buildPublishOutputPlan(sessionPrefix, runPrefix, publishRules, discoveredOutputs, initialLocks, priorCommit)
if err != nil {
return nil, fmt.Errorf("publish: plan publish outputs: %w", err)
}
runUploaded := publishUploadPaths(runFiles)
publishedUploaded := publishOutputPaths(publishOutputs)
publishedUploaded := publishOutputPaths(outputPlan)
previousUploaded := publishUploadPaths(previousFiles)
commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID)
pointerKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix)
transfers, err := buildPublishTransferPlan(runPrefix, runFiles, runSources, previousFiles, previousSources, publishOutputs)
carriedDeclarations, err := verifyCarriedPublishOutputs(ctx, env.ObjectStore, outputPlan)
if err != nil {
return nil, fmt.Errorf("publish: %w", err)
}
transfers, err := buildPublishTransferPlan(runPrefix, runFiles, runSources, previousFiles, previousSources, outputPlan)
if err != nil {
return nil, fmt.Errorf("publish: build immutable upload mapping: %w", err)
}
@@ -146,9 +165,7 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
runUploaded,
publishedUploaded,
previousUploaded,
skippedOptionalOutputs,
skippedUnselectedOutputs,
lockedOutputs,
outputPlan,
commitKey,
pointerKey,
))
@@ -165,10 +182,11 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
// Durability boundary: transfer every declared immutable object, write the
// immutable commit, perform pre-pointer checks, then write the sole mutable
// current pointer exactly once.
declarations, err := uploadPublishTransfers(ctx, env.ObjectStore, uploader, transfers)
uploadedDeclarations, err := uploadPublishTransfers(ctx, env.ObjectStore, uploader, transfers)
if err != nil {
return nil, fmt.Errorf("publish: %w", err)
}
declarations := combinePublishDeclarations(uploadedDeclarations, carriedDeclarations)
committed, err := writeImmutablePublishCommit(
ctx,
env.ObjectStore,
@@ -182,51 +200,24 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if err != nil {
return nil, fmt.Errorf("publish: %w", err)
}
if err := revalidatePublishCommitLocks(ctx, env, publishOutputs); err != nil {
if err := revalidatePublishOutputLocks(ctx, env.ObjectStore, sessionPrefix, outputPlan); err != nil {
return nil, fmt.Errorf("publish: revalidate locks before current commit selection: %w", err)
}
if err := writePublishCurrentPointer(ctx, uploader, committed); err != nil {
return nil, fmt.Errorf("publish: %w", err)
}
return &StageResult{
Metadata: map[string]any{
"stage": "publish",
"uploaded": true,
"s3_bucket": bucket,
"s3_run_prefix": runPrefix,
"run_files_uploaded": len(runUploaded),
"run_uploaded_paths": runUploaded,
"published_files_uploaded": len(publishedUploaded),
"published_paths": publishedUploaded,
"previous_files_uploaded": len(previousUploaded),
"previous_uploaded_paths": previousUploaded,
"skipped_optional_outputs": skippedOptionalOutputs,
"skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs),
"locked_output_count": len(lockedOutputs),
"locked_outputs": lockedOutputMetadata(lockedOutputs),
"remote_commit_key": commitKey,
"current_commit_pointer_key": pointerKey,
"audio_upload_skipped": true,
},
}, nil
}
func revalidatePublishCommitLocks(ctx context.Context, env *Env, outputs []publishOutput) error {
if env == nil || env.RevalidatePublishLocks == nil || len(outputs) == 0 {
return nil
}
locks, err := env.RevalidatePublishLocks(ctx)
if err != nil {
return err
}
lockSet := publishLockSet(locks)
for _, output := range outputs {
if lock, locked := lockSet[output.Source]; locked {
return fmt.Errorf("source %q is locked: %s", output.Source, strings.TrimSpace(lock.Reason))
}
}
return nil
return &StageResult{Metadata: publishMetadata(
bucket,
runPrefix,
runID,
runUploaded,
publishedUploaded,
previousUploaded,
outputPlan,
commitKey,
pointerKey,
)}, nil
}
func publishDisabled(env *Env) bool {
@@ -310,9 +301,7 @@ func publishMetadata(
runUploaded []string,
publishedUploaded []string,
previousUploaded []string,
skippedOptionalOutputs []string,
skippedUnselectedOutputs []publishSkippedUnselectedOutput,
lockedOutputs []publishLockedOutput,
outputPlan *publishOutputPlan,
commitKey string,
pointerKey string,
) map[string]any {
@@ -328,40 +317,9 @@ func publishMetadata(
"published_paths": append([]string(nil), publishedUploaded...),
"previous_files_uploaded": len(previousUploaded),
"previous_uploaded_paths": append([]string(nil), previousUploaded...),
"skipped_optional_outputs": append([]string(nil), skippedOptionalOutputs...),
"skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs),
"locked_output_count": len(lockedOutputs),
"locked_outputs": lockedOutputMetadata(lockedOutputs),
"publish_output_outcomes": publishOutputOutcomeMetadata(outputPlan),
"remote_commit_key": commitKey,
"current_commit_pointer_key": pointerKey,
"audio_upload_skipped": true,
}
}
func skippedUnselectedOutputMetadata(skipped []publishSkippedUnselectedOutput) []map[string]any {
out := make([]map[string]any, 0, len(skipped))
for _, item := range skipped {
out = append(out, map[string]any{
"source": item.Source,
"dest": item.Dest,
"required": item.Required,
})
}
return out
}
func lockedOutputMetadata(locked []publishLockedOutput) []map[string]any {
out := make([]map[string]any, 0, len(locked))
for _, item := range locked {
out = append(out, map[string]any{
"source": item.Source,
"dest": item.Dest,
"remote_key": item.RemoteKey,
"reason": item.Reason,
"required": item.Required,
"local_path": item.LocalPath,
"provenance": item.Provenance,
})
}
return out
}