Separate publish transfer and commit mechanics
This commit is contained in:
@@ -30,7 +30,10 @@ Exact remote placement and the operator workflow belong in
|
||||
manifest-derived archive metadata, confined opening, and checksum validation
|
||||
live in `publish_sources.go`; artifact-catalog matching, selection, required
|
||||
and optional rules, destination resolution, and collision inputs live in
|
||||
`publish_outputs.go`.
|
||||
`publish_outputs.go`. Copying, hashing, immutable upload, reuse checks, and
|
||||
remote verification live in `publish_transfer.go`; committed-manifest,
|
||||
immutable-commit, and current-pointer encoding and writes live in
|
||||
`publish_commit.go`.
|
||||
|
||||
- when publishing or run upload is disabled, completes successfully with no
|
||||
outputs and records explanatory metadata. This is not an explicit self-skip:
|
||||
@@ -122,6 +125,7 @@ The commit boundary and cleanup gate are normative architecture invariants; see
|
||||
helpers.
|
||||
- Implementation and tests: `internal/stage/publish.go`,
|
||||
`internal/stage/publish_sources.go`, `internal/stage/publish_outputs.go`,
|
||||
`internal/stage/publish_transfer.go`, `internal/stage/publish_commit.go`,
|
||||
`internal/stage/publish_test.go`,
|
||||
`internal/stage/semantic_contracts_delivery.go`,
|
||||
`internal/app/operator_helpers_test.go`, and
|
||||
|
||||
@@ -195,7 +195,7 @@ file-level ownership.
|
||||
|
||||
## Stage 5 — Separate Transfer And Commit Mechanics
|
||||
|
||||
**Status:** Pending
|
||||
**Status:** Completed
|
||||
|
||||
### Goal
|
||||
|
||||
|
||||
@@ -1,16 +1,10 @@
|
||||
package stage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||
@@ -168,46 +162,31 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
||||
return nil, fmt.Errorf("publish: validate immutable upload mapping: %w", err)
|
||||
}
|
||||
|
||||
commit := artifacts.RemoteCommitManifest{
|
||||
FormatVersion: artifacts.RemoteCommitFormatVersion,
|
||||
Campaign: strings.TrimSpace(m.Campaign),
|
||||
SessionID: strings.TrimSpace(m.SessionID),
|
||||
RunID: runID,
|
||||
Artifacts: make([]artifacts.RemoteArtifact, 0, len(transfers)),
|
||||
}
|
||||
for index := range transfers {
|
||||
artifact, err := uploadVerifiedPublishTransfer(ctx, env.ObjectStore, uploader, &transfers[index])
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("publish: upload immutable object %q: %w", transfers[index].Artifact.DestinationKey, err)
|
||||
}
|
||||
commit.Artifacts = append(commit.Artifacts, artifact)
|
||||
}
|
||||
commitData, err := artifacts.EncodeRemoteCommitManifest(commit)
|
||||
// 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)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("publish: encode immutable commit: %w", err)
|
||||
return nil, fmt.Errorf("publish: %w", err)
|
||||
}
|
||||
commitInfo, err := uploadVerifiedPublishBytes(ctx, env.ObjectStore, uploader, commitKey, commitData, "application/json")
|
||||
committed, err := writeImmutablePublishCommit(
|
||||
ctx,
|
||||
env.ObjectStore,
|
||||
uploader,
|
||||
sessionPrefix,
|
||||
m.Campaign,
|
||||
m.SessionID,
|
||||
runID,
|
||||
declarations,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("publish: upload immutable commit %q: %w", commitKey, err)
|
||||
return nil, fmt.Errorf("publish: %w", err)
|
||||
}
|
||||
if err := revalidatePublishCommitLocks(ctx, env, publishOutputs); err != nil {
|
||||
return nil, fmt.Errorf("publish: revalidate locks before current commit selection: %w", err)
|
||||
}
|
||||
pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{
|
||||
FormatVersion: artifacts.RemoteCommitFormatVersion,
|
||||
Campaign: commit.Campaign,
|
||||
SessionID: commit.SessionID,
|
||||
RunID: commit.RunID,
|
||||
CommitKey: commitKey,
|
||||
CommitSHA256: publishSHA256(commitData),
|
||||
CommitSize: int64(len(commitData)),
|
||||
CommitGeneration: commitInfo.ETag,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("publish: encode current commit pointer: %w", err)
|
||||
}
|
||||
if _, err := uploader.UploadReader(ctx, bytes.NewReader(pointerData), pointerKey, storage.UploadOptions{ContentType: "application/json"}); err != nil {
|
||||
return nil, fmt.Errorf("publish: update current commit pointer %q: %w", pointerKey, err)
|
||||
if err := writePublishCurrentPointer(ctx, uploader, committed); err != nil {
|
||||
return nil, fmt.Errorf("publish: %w", err)
|
||||
}
|
||||
|
||||
return &StageResult{
|
||||
@@ -250,264 +229,6 @@ func revalidatePublishCommitLocks(ctx context.Context, env *Env, outputs []publi
|
||||
return nil
|
||||
}
|
||||
|
||||
type publishTransfer struct {
|
||||
Artifact artifacts.RemoteArtifact
|
||||
Source *os.File
|
||||
TemporaryPath string
|
||||
}
|
||||
|
||||
func buildPublishTransferPlan(
|
||||
runPrefix string,
|
||||
runFiles []publishUploadFile,
|
||||
runSources []*os.File,
|
||||
previousFiles []publishUploadFile,
|
||||
previousSources []*os.File,
|
||||
publishOutputs []publishOutput,
|
||||
) ([]publishTransfer, error) {
|
||||
if len(runFiles) != len(runSources) || len(previousFiles) != len(previousSources) {
|
||||
return nil, fmt.Errorf("verified publish sources do not match their declared files")
|
||||
}
|
||||
transfers := make([]publishTransfer, 0, len(runFiles)+len(previousFiles)+len(publishOutputs))
|
||||
add := func(artifact artifacts.RemoteArtifact, source *os.File) error {
|
||||
if source == nil {
|
||||
return fmt.Errorf("source for %q is required", artifact.DestinationKey)
|
||||
}
|
||||
transfers = append(transfers, publishTransfer{Artifact: artifact, Source: source})
|
||||
return nil
|
||||
}
|
||||
for index, file := range runFiles {
|
||||
if err := add(artifacts.RemoteArtifact{
|
||||
Type: artifacts.RemoteArtifactTypeRunFile,
|
||||
Source: "run:" + file.RelativePath,
|
||||
DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
|
||||
}, runSources[index]); err != nil {
|
||||
closePublishTransfers(transfers)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
for index, file := range previousFiles {
|
||||
if err := add(artifacts.RemoteArtifact{
|
||||
Type: artifacts.RemoteArtifactTypePreviousArtifact,
|
||||
Source: "previous:" + file.SourceRelativePath,
|
||||
DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
|
||||
}, previousSources[index]); err != nil {
|
||||
closePublishTransfers(transfers)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
for _, output := range publishOutputs {
|
||||
source, err := openPublishOutputSource(output.LocalPath)
|
||||
if err != nil {
|
||||
closePublishTransfers(transfers)
|
||||
return nil, fmt.Errorf("open output source %q: %w", output.Source, err)
|
||||
}
|
||||
if err := add(artifacts.RemoteArtifact{
|
||||
Type: artifacts.RemoteArtifactTypePublishedOutput,
|
||||
Source: output.Source,
|
||||
DestinationKey: publishOutputRemoteKey(runPrefix, output.Dest),
|
||||
}, source); err != nil {
|
||||
_ = source.Close()
|
||||
closePublishTransfers(transfers)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
sort.Slice(transfers, func(i, j int) bool {
|
||||
return transfers[i].Artifact.DestinationKey < transfers[j].Artifact.DestinationKey
|
||||
})
|
||||
return transfers, nil
|
||||
}
|
||||
|
||||
func appendCommittedSessionManifest(transfers *[]publishTransfer, localPath, destinationKey string) error {
|
||||
if transfers == nil {
|
||||
return fmt.Errorf("transfer plan is required")
|
||||
}
|
||||
source, err := os.Open(localPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
*transfers = append(*transfers, publishTransfer{
|
||||
Artifact: artifacts.RemoteArtifact{
|
||||
Type: artifacts.RemoteArtifactTypeSessionManifest,
|
||||
Source: "session.manifest",
|
||||
DestinationKey: destinationKey,
|
||||
},
|
||||
Source: source,
|
||||
TemporaryPath: localPath,
|
||||
})
|
||||
sort.Slice(*transfers, func(i, j int) bool {
|
||||
return (*transfers)[i].Artifact.DestinationKey < (*transfers)[j].Artifact.DestinationKey
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func validatePublishTransferPlan(transfers []publishTransfer, commitKey string) error {
|
||||
artifactsToUpload := make([]artifacts.RemoteArtifact, 0, len(transfers))
|
||||
for _, transfer := range transfers {
|
||||
if transfer.Artifact.DestinationKey == commitKey {
|
||||
return fmt.Errorf("artifact destination %q is reserved for the immutable commit", commitKey)
|
||||
}
|
||||
artifactsToUpload = append(artifactsToUpload, transfer.Artifact)
|
||||
}
|
||||
return artifacts.ValidateRemoteArtifactMapping(artifactsToUpload)
|
||||
}
|
||||
|
||||
func uploadVerifiedPublishTransfer(
|
||||
ctx context.Context,
|
||||
store storage.ObjectStore,
|
||||
uploader storage.ReaderUploader,
|
||||
transfer *publishTransfer,
|
||||
) (artifacts.RemoteArtifact, error) {
|
||||
if transfer == nil || transfer.Source == nil {
|
||||
return artifacts.RemoteArtifact{}, fmt.Errorf("transfer source is required")
|
||||
}
|
||||
if err := preparePublishTransfer(transfer); err != nil {
|
||||
return artifacts.RemoteArtifact{}, err
|
||||
}
|
||||
if existing, found, err := verifyExistingImmutableObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.SHA256, transfer.Artifact.Size); err != nil {
|
||||
return artifacts.RemoteArtifact{}, err
|
||||
} else if found {
|
||||
transfer.Artifact.Generation = existing.ETag
|
||||
return transfer.Artifact, nil
|
||||
}
|
||||
info, err := uploader.UploadReader(ctx, transfer.Source, transfer.Artifact.DestinationKey, storage.UploadOptions{})
|
||||
if err != nil {
|
||||
return artifacts.RemoteArtifact{}, err
|
||||
}
|
||||
verified, err := verifyPublishedObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.Size, info.ETag)
|
||||
if err != nil {
|
||||
return artifacts.RemoteArtifact{}, err
|
||||
}
|
||||
transfer.Artifact.Generation = verified.ETag
|
||||
return transfer.Artifact, nil
|
||||
}
|
||||
|
||||
func uploadVerifiedPublishBytes(
|
||||
ctx context.Context,
|
||||
store storage.ObjectStore,
|
||||
uploader storage.ReaderUploader,
|
||||
key string,
|
||||
data []byte,
|
||||
contentType string,
|
||||
) (storage.ObjectInfo, error) {
|
||||
checksum := publishSHA256(data)
|
||||
if existing, found, err := verifyExistingImmutableObject(ctx, store, key, checksum, int64(len(data))); err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
} else if found {
|
||||
return existing, nil
|
||||
}
|
||||
info, err := uploader.UploadReader(ctx, bytes.NewReader(data), key, storage.UploadOptions{ContentType: contentType})
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
return verifyPublishedObject(ctx, store, key, int64(len(data)), info.ETag)
|
||||
}
|
||||
|
||||
func verifyExistingImmutableObject(ctx context.Context, store storage.ObjectStore, key, wantSHA256 string, wantSize int64) (storage.ObjectInfo, bool, error) {
|
||||
exists, err := store.Exists(ctx, key)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("check immutable object %q: %w", key, err)
|
||||
}
|
||||
if !exists {
|
||||
return storage.ObjectInfo{}, false, nil
|
||||
}
|
||||
path, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-verify-immutable-*.bin")
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("download existing immutable object %q: %w", key, err)
|
||||
}
|
||||
defer func() { _ = os.Remove(path) }()
|
||||
file, err := os.Open(path)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("open existing immutable object %q: %w", key, err)
|
||||
}
|
||||
hash := sha256.New()
|
||||
size, copyErr := io.Copy(hash, file)
|
||||
closeErr := file.Close()
|
||||
if copyErr != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("checksum existing immutable object %q: %w", key, copyErr)
|
||||
}
|
||||
if closeErr != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("close existing immutable object %q: %w", key, closeErr)
|
||||
}
|
||||
if size != wantSize || hex.EncodeToString(hash.Sum(nil)) != wantSHA256 {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("immutable object %q already exists with different content", key)
|
||||
}
|
||||
info, err := verifyPublishedObject(ctx, store, key, wantSize, "")
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, false, err
|
||||
}
|
||||
return info, true, nil
|
||||
}
|
||||
|
||||
func preparePublishTransfer(transfer *publishTransfer) error {
|
||||
if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil {
|
||||
return fmt.Errorf("rewind source %q: %w", transfer.Artifact.Source, err)
|
||||
}
|
||||
info, err := transfer.Source.Stat()
|
||||
if err != nil {
|
||||
return fmt.Errorf("stat source %q: %w", transfer.Artifact.Source, err)
|
||||
}
|
||||
if !info.Mode().IsRegular() {
|
||||
return fmt.Errorf("source %q is not a regular file", transfer.Artifact.Source)
|
||||
}
|
||||
hash := sha256.New()
|
||||
if _, err := io.Copy(hash, transfer.Source); err != nil {
|
||||
return fmt.Errorf("checksum source %q: %w", transfer.Artifact.Source, err)
|
||||
}
|
||||
if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil {
|
||||
return fmt.Errorf("rewind source %q after checksum: %w", transfer.Artifact.Source, err)
|
||||
}
|
||||
transfer.Artifact.Size = info.Size()
|
||||
transfer.Artifact.SHA256 = hex.EncodeToString(hash.Sum(nil))
|
||||
return nil
|
||||
}
|
||||
|
||||
func verifyPublishedObject(ctx context.Context, store storage.ObjectStore, key string, wantSize int64, uploadedGeneration string) (storage.ObjectInfo, error) {
|
||||
objects, err := store.List(ctx, key)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("list uploaded object %q: %w", key, err)
|
||||
}
|
||||
var found *storage.ObjectInfo
|
||||
for _, object := range objects {
|
||||
if object.Key != key {
|
||||
continue
|
||||
}
|
||||
if found != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is ambiguous", key)
|
||||
}
|
||||
copy := object
|
||||
found = ©
|
||||
}
|
||||
if found == nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is missing", key)
|
||||
}
|
||||
if found.Size != wantSize {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q size is %d, want %d", key, found.Size, wantSize)
|
||||
}
|
||||
if strings.TrimSpace(found.ETag) == "" {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q has no generation", key)
|
||||
}
|
||||
if strings.TrimSpace(uploadedGeneration) != "" && found.ETag != uploadedGeneration {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q generation changed during verification", key)
|
||||
}
|
||||
return *found, nil
|
||||
}
|
||||
|
||||
func closePublishTransfers(transfers []publishTransfer) {
|
||||
for _, transfer := range transfers {
|
||||
if transfer.Source != nil {
|
||||
_ = transfer.Source.Close()
|
||||
}
|
||||
if transfer.TemporaryPath != "" {
|
||||
_ = os.Remove(transfer.TemporaryPath)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func publishSHA256(data []byte) string {
|
||||
sum := sha256.Sum256(data)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func publishDisabled(env *Env) bool {
|
||||
cfg := env.Config.Pipeline.Publish
|
||||
if cfg == nil {
|
||||
@@ -584,62 +305,6 @@ func publishSessionPaths(env *Env, m *manifest.Manifest) artifacts.SessionPaths
|
||||
return store.SessionPathsFor(campaign, sessionID)
|
||||
}
|
||||
|
||||
func writeCommittedSessionManifest(m *manifest.Manifest, publishMetadata map[string]any) (string, error) {
|
||||
if m == nil {
|
||||
return "", fmt.Errorf("manifest is required")
|
||||
}
|
||||
clone := *m
|
||||
clone.Stages = make(map[string]*manifest.StageRecord, len(m.Stages))
|
||||
for name, sr := range m.Stages {
|
||||
if sr == nil {
|
||||
continue
|
||||
}
|
||||
stageCopy := *sr
|
||||
if sr.Outputs != nil {
|
||||
stageCopy.Outputs = append([]manifest.ArtifactRecord(nil), sr.Outputs...)
|
||||
}
|
||||
if sr.Logs != nil {
|
||||
stageCopy.Logs = append([]string(nil), sr.Logs...)
|
||||
}
|
||||
if sr.GeneratedConfigs != nil {
|
||||
stageCopy.GeneratedConfigs = append([]string(nil), sr.GeneratedConfigs...)
|
||||
}
|
||||
if sr.Metadata != nil {
|
||||
metaCopy := make(map[string]any, len(sr.Metadata))
|
||||
for k, v := range sr.Metadata {
|
||||
metaCopy[k] = v
|
||||
}
|
||||
stageCopy.Metadata = metaCopy
|
||||
}
|
||||
clone.Stages[name] = &stageCopy
|
||||
}
|
||||
|
||||
clone.MarkStageSucceeded("publish", clone.UpdatedAt, nil)
|
||||
if sr := clone.Stages["publish"]; sr != nil {
|
||||
sr.Metadata = publishMetadata
|
||||
}
|
||||
|
||||
data, err := json.MarshalIndent(&clone, "", " ")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("marshal manifest: %w", err)
|
||||
}
|
||||
data = append(data, '\n')
|
||||
|
||||
tmp, err := os.CreateTemp("", "narratio-committed-session-manifest-*.json")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("create temp file: %w", err)
|
||||
}
|
||||
path := tmp.Name()
|
||||
if _, err := tmp.Write(data); err != nil {
|
||||
_ = tmp.Close()
|
||||
return "", fmt.Errorf("write temp manifest: %w", err)
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
return "", fmt.Errorf("close temp manifest: %w", err)
|
||||
}
|
||||
return path, nil
|
||||
}
|
||||
|
||||
func publishMetadata(
|
||||
bucket, runPrefix, runID string,
|
||||
runUploaded []string,
|
||||
|
||||
136
internal/stage/publish_commit.go
Normal file
136
internal/stage/publish_commit.go
Normal file
@@ -0,0 +1,136 @@
|
||||
package stage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||
)
|
||||
|
||||
type publishedCommit struct {
|
||||
Manifest artifacts.RemoteCommitManifest
|
||||
Key string
|
||||
Data []byte
|
||||
Generation string
|
||||
PointerKey string
|
||||
}
|
||||
|
||||
func writeImmutablePublishCommit(
|
||||
ctx context.Context,
|
||||
store storage.ObjectStore,
|
||||
uploader storage.ReaderUploader,
|
||||
sessionPrefix string,
|
||||
campaign string,
|
||||
sessionID string,
|
||||
runID string,
|
||||
declarations []artifacts.RemoteArtifact,
|
||||
) (*publishedCommit, error) {
|
||||
commit := artifacts.RemoteCommitManifest{
|
||||
FormatVersion: artifacts.RemoteCommitFormatVersion,
|
||||
Campaign: strings.TrimSpace(campaign),
|
||||
SessionID: strings.TrimSpace(sessionID),
|
||||
RunID: runID,
|
||||
Artifacts: declarations,
|
||||
}
|
||||
commitData, err := artifacts.EncodeRemoteCommitManifest(commit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("encode immutable commit: %w", err)
|
||||
}
|
||||
commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID)
|
||||
commitInfo, err := uploadVerifiedPublishBytes(ctx, store, uploader, commitKey, commitData, "application/json")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("upload immutable commit %q: %w", commitKey, err)
|
||||
}
|
||||
return &publishedCommit{
|
||||
Manifest: commit,
|
||||
Key: commitKey,
|
||||
Data: commitData,
|
||||
Generation: commitInfo.ETag,
|
||||
PointerKey: artifacts.S3CurrentCommitPointerKey(sessionPrefix),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func writePublishCurrentPointer(ctx context.Context, uploader storage.ReaderUploader, commit *publishedCommit) error {
|
||||
if commit == nil {
|
||||
return fmt.Errorf("published commit is required")
|
||||
}
|
||||
pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{
|
||||
FormatVersion: artifacts.RemoteCommitFormatVersion,
|
||||
Campaign: commit.Manifest.Campaign,
|
||||
SessionID: commit.Manifest.SessionID,
|
||||
RunID: commit.Manifest.RunID,
|
||||
CommitKey: commit.Key,
|
||||
CommitSHA256: publishSHA256(commit.Data),
|
||||
CommitSize: int64(len(commit.Data)),
|
||||
CommitGeneration: commit.Generation,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode current commit pointer: %w", err)
|
||||
}
|
||||
if _, err := uploader.UploadReader(ctx, bytes.NewReader(pointerData), commit.PointerKey, storage.UploadOptions{ContentType: "application/json"}); err != nil {
|
||||
return fmt.Errorf("update current commit pointer %q: %w", commit.PointerKey, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func writeCommittedSessionManifest(m *manifest.Manifest, publishMetadata map[string]any) (string, error) {
|
||||
if m == nil {
|
||||
return "", fmt.Errorf("manifest is required")
|
||||
}
|
||||
clone := *m
|
||||
clone.Stages = make(map[string]*manifest.StageRecord, len(m.Stages))
|
||||
for name, sr := range m.Stages {
|
||||
if sr == nil {
|
||||
continue
|
||||
}
|
||||
stageCopy := *sr
|
||||
if sr.Outputs != nil {
|
||||
stageCopy.Outputs = append([]manifest.ArtifactRecord(nil), sr.Outputs...)
|
||||
}
|
||||
if sr.Logs != nil {
|
||||
stageCopy.Logs = append([]string(nil), sr.Logs...)
|
||||
}
|
||||
if sr.GeneratedConfigs != nil {
|
||||
stageCopy.GeneratedConfigs = append([]string(nil), sr.GeneratedConfigs...)
|
||||
}
|
||||
if sr.Metadata != nil {
|
||||
metaCopy := make(map[string]any, len(sr.Metadata))
|
||||
for k, v := range sr.Metadata {
|
||||
metaCopy[k] = v
|
||||
}
|
||||
stageCopy.Metadata = metaCopy
|
||||
}
|
||||
clone.Stages[name] = &stageCopy
|
||||
}
|
||||
|
||||
clone.MarkStageSucceeded("publish", clone.UpdatedAt, nil)
|
||||
if sr := clone.Stages["publish"]; sr != nil {
|
||||
sr.Metadata = publishMetadata
|
||||
}
|
||||
|
||||
data, err := json.MarshalIndent(&clone, "", " ")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("marshal manifest: %w", err)
|
||||
}
|
||||
data = append(data, '\n')
|
||||
|
||||
tmp, err := os.CreateTemp("", "narratio-committed-session-manifest-*.json")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("create temp file: %w", err)
|
||||
}
|
||||
path := tmp.Name()
|
||||
if _, err := tmp.Write(data); err != nil {
|
||||
_ = tmp.Close()
|
||||
return "", fmt.Errorf("write temp manifest: %w", err)
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
return "", fmt.Errorf("close temp manifest: %w", err)
|
||||
}
|
||||
return path, nil
|
||||
}
|
||||
291
internal/stage/publish_transfer.go
Normal file
291
internal/stage/publish_transfer.go
Normal file
@@ -0,0 +1,291 @@
|
||||
package stage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||
)
|
||||
|
||||
type publishTransfer struct {
|
||||
Artifact artifacts.RemoteArtifact
|
||||
Source *os.File
|
||||
TemporaryPath string
|
||||
}
|
||||
|
||||
func buildPublishTransferPlan(
|
||||
runPrefix string,
|
||||
runFiles []publishUploadFile,
|
||||
runSources []*os.File,
|
||||
previousFiles []publishUploadFile,
|
||||
previousSources []*os.File,
|
||||
publishOutputs []publishOutput,
|
||||
) ([]publishTransfer, error) {
|
||||
if len(runFiles) != len(runSources) || len(previousFiles) != len(previousSources) {
|
||||
return nil, fmt.Errorf("verified publish sources do not match their declared files")
|
||||
}
|
||||
transfers := make([]publishTransfer, 0, len(runFiles)+len(previousFiles)+len(publishOutputs))
|
||||
add := func(artifact artifacts.RemoteArtifact, source *os.File) error {
|
||||
if source == nil {
|
||||
return fmt.Errorf("source for %q is required", artifact.DestinationKey)
|
||||
}
|
||||
transfers = append(transfers, publishTransfer{Artifact: artifact, Source: source})
|
||||
return nil
|
||||
}
|
||||
for index, file := range runFiles {
|
||||
if err := add(artifacts.RemoteArtifact{
|
||||
Type: artifacts.RemoteArtifactTypeRunFile,
|
||||
Source: "run:" + file.RelativePath,
|
||||
DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
|
||||
}, runSources[index]); err != nil {
|
||||
closePublishTransfers(transfers)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
for index, file := range previousFiles {
|
||||
if err := add(artifacts.RemoteArtifact{
|
||||
Type: artifacts.RemoteArtifactTypePreviousArtifact,
|
||||
Source: "previous:" + file.SourceRelativePath,
|
||||
DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
|
||||
}, previousSources[index]); err != nil {
|
||||
closePublishTransfers(transfers)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
for _, output := range publishOutputs {
|
||||
source, err := openPublishOutputSource(output.LocalPath)
|
||||
if err != nil {
|
||||
closePublishTransfers(transfers)
|
||||
return nil, fmt.Errorf("open output source %q: %w", output.Source, err)
|
||||
}
|
||||
if err := add(artifacts.RemoteArtifact{
|
||||
Type: artifacts.RemoteArtifactTypePublishedOutput,
|
||||
Source: output.Source,
|
||||
DestinationKey: publishOutputRemoteKey(runPrefix, output.Dest),
|
||||
}, source); err != nil {
|
||||
_ = source.Close()
|
||||
closePublishTransfers(transfers)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
sort.Slice(transfers, func(i, j int) bool {
|
||||
return transfers[i].Artifact.DestinationKey < transfers[j].Artifact.DestinationKey
|
||||
})
|
||||
return transfers, nil
|
||||
}
|
||||
|
||||
func appendCommittedSessionManifest(transfers *[]publishTransfer, localPath, destinationKey string) error {
|
||||
if transfers == nil {
|
||||
return fmt.Errorf("transfer plan is required")
|
||||
}
|
||||
source, err := os.Open(localPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
*transfers = append(*transfers, publishTransfer{
|
||||
Artifact: artifacts.RemoteArtifact{
|
||||
Type: artifacts.RemoteArtifactTypeSessionManifest,
|
||||
Source: "session.manifest",
|
||||
DestinationKey: destinationKey,
|
||||
},
|
||||
Source: source,
|
||||
TemporaryPath: localPath,
|
||||
})
|
||||
sort.Slice(*transfers, func(i, j int) bool {
|
||||
return (*transfers)[i].Artifact.DestinationKey < (*transfers)[j].Artifact.DestinationKey
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func validatePublishTransferPlan(transfers []publishTransfer, commitKey string) error {
|
||||
artifactsToUpload := make([]artifacts.RemoteArtifact, 0, len(transfers))
|
||||
for _, transfer := range transfers {
|
||||
if transfer.Artifact.DestinationKey == commitKey {
|
||||
return fmt.Errorf("artifact destination %q is reserved for the immutable commit", commitKey)
|
||||
}
|
||||
artifactsToUpload = append(artifactsToUpload, transfer.Artifact)
|
||||
}
|
||||
return artifacts.ValidateRemoteArtifactMapping(artifactsToUpload)
|
||||
}
|
||||
|
||||
func uploadPublishTransfers(
|
||||
ctx context.Context,
|
||||
store storage.ObjectStore,
|
||||
uploader storage.ReaderUploader,
|
||||
transfers []publishTransfer,
|
||||
) ([]artifacts.RemoteArtifact, error) {
|
||||
declarations := make([]artifacts.RemoteArtifact, 0, len(transfers))
|
||||
for index := range transfers {
|
||||
artifact, err := uploadVerifiedPublishTransfer(ctx, store, uploader, &transfers[index])
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("upload immutable object %q: %w", transfers[index].Artifact.DestinationKey, err)
|
||||
}
|
||||
declarations = append(declarations, artifact)
|
||||
}
|
||||
return declarations, nil
|
||||
}
|
||||
|
||||
func uploadVerifiedPublishTransfer(
|
||||
ctx context.Context,
|
||||
store storage.ObjectStore,
|
||||
uploader storage.ReaderUploader,
|
||||
transfer *publishTransfer,
|
||||
) (artifacts.RemoteArtifact, error) {
|
||||
if transfer == nil || transfer.Source == nil {
|
||||
return artifacts.RemoteArtifact{}, fmt.Errorf("transfer source is required")
|
||||
}
|
||||
if err := preparePublishTransfer(transfer); err != nil {
|
||||
return artifacts.RemoteArtifact{}, err
|
||||
}
|
||||
if existing, found, err := verifyExistingImmutableObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.SHA256, transfer.Artifact.Size); err != nil {
|
||||
return artifacts.RemoteArtifact{}, err
|
||||
} else if found {
|
||||
transfer.Artifact.Generation = existing.ETag
|
||||
return transfer.Artifact, nil
|
||||
}
|
||||
info, err := uploader.UploadReader(ctx, transfer.Source, transfer.Artifact.DestinationKey, storage.UploadOptions{})
|
||||
if err != nil {
|
||||
return artifacts.RemoteArtifact{}, err
|
||||
}
|
||||
verified, err := verifyPublishedObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.Size, info.ETag)
|
||||
if err != nil {
|
||||
return artifacts.RemoteArtifact{}, err
|
||||
}
|
||||
transfer.Artifact.Generation = verified.ETag
|
||||
return transfer.Artifact, nil
|
||||
}
|
||||
|
||||
func uploadVerifiedPublishBytes(
|
||||
ctx context.Context,
|
||||
store storage.ObjectStore,
|
||||
uploader storage.ReaderUploader,
|
||||
key string,
|
||||
data []byte,
|
||||
contentType string,
|
||||
) (storage.ObjectInfo, error) {
|
||||
checksum := publishSHA256(data)
|
||||
if existing, found, err := verifyExistingImmutableObject(ctx, store, key, checksum, int64(len(data))); err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
} else if found {
|
||||
return existing, nil
|
||||
}
|
||||
info, err := uploader.UploadReader(ctx, bytes.NewReader(data), key, storage.UploadOptions{ContentType: contentType})
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
return verifyPublishedObject(ctx, store, key, int64(len(data)), info.ETag)
|
||||
}
|
||||
|
||||
func verifyExistingImmutableObject(ctx context.Context, store storage.ObjectStore, key, wantSHA256 string, wantSize int64) (storage.ObjectInfo, bool, error) {
|
||||
exists, err := store.Exists(ctx, key)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("check immutable object %q: %w", key, err)
|
||||
}
|
||||
if !exists {
|
||||
return storage.ObjectInfo{}, false, nil
|
||||
}
|
||||
path, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-verify-immutable-*.bin")
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("download existing immutable object %q: %w", key, err)
|
||||
}
|
||||
defer func() { _ = os.Remove(path) }()
|
||||
file, err := os.Open(path)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("open existing immutable object %q: %w", key, err)
|
||||
}
|
||||
hash := sha256.New()
|
||||
size, copyErr := io.Copy(hash, file)
|
||||
closeErr := file.Close()
|
||||
if copyErr != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("checksum existing immutable object %q: %w", key, copyErr)
|
||||
}
|
||||
if closeErr != nil {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("close existing immutable object %q: %w", key, closeErr)
|
||||
}
|
||||
if size != wantSize || hex.EncodeToString(hash.Sum(nil)) != wantSHA256 {
|
||||
return storage.ObjectInfo{}, false, fmt.Errorf("immutable object %q already exists with different content", key)
|
||||
}
|
||||
info, err := verifyPublishedObject(ctx, store, key, wantSize, "")
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, false, err
|
||||
}
|
||||
return info, true, nil
|
||||
}
|
||||
|
||||
func preparePublishTransfer(transfer *publishTransfer) error {
|
||||
if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil {
|
||||
return fmt.Errorf("rewind source %q: %w", transfer.Artifact.Source, err)
|
||||
}
|
||||
info, err := transfer.Source.Stat()
|
||||
if err != nil {
|
||||
return fmt.Errorf("stat source %q: %w", transfer.Artifact.Source, err)
|
||||
}
|
||||
if !info.Mode().IsRegular() {
|
||||
return fmt.Errorf("source %q is not a regular file", transfer.Artifact.Source)
|
||||
}
|
||||
hash := sha256.New()
|
||||
if _, err := io.Copy(hash, transfer.Source); err != nil {
|
||||
return fmt.Errorf("checksum source %q: %w", transfer.Artifact.Source, err)
|
||||
}
|
||||
if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil {
|
||||
return fmt.Errorf("rewind source %q after checksum: %w", transfer.Artifact.Source, err)
|
||||
}
|
||||
transfer.Artifact.Size = info.Size()
|
||||
transfer.Artifact.SHA256 = hex.EncodeToString(hash.Sum(nil))
|
||||
return nil
|
||||
}
|
||||
|
||||
func verifyPublishedObject(ctx context.Context, store storage.ObjectStore, key string, wantSize int64, uploadedGeneration string) (storage.ObjectInfo, error) {
|
||||
objects, err := store.List(ctx, key)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("list uploaded object %q: %w", key, err)
|
||||
}
|
||||
var found *storage.ObjectInfo
|
||||
for _, object := range objects {
|
||||
if object.Key != key {
|
||||
continue
|
||||
}
|
||||
if found != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is ambiguous", key)
|
||||
}
|
||||
copy := object
|
||||
found = ©
|
||||
}
|
||||
if found == nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is missing", key)
|
||||
}
|
||||
if found.Size != wantSize {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q size is %d, want %d", key, found.Size, wantSize)
|
||||
}
|
||||
if strings.TrimSpace(found.ETag) == "" {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q has no generation", key)
|
||||
}
|
||||
if strings.TrimSpace(uploadedGeneration) != "" && found.ETag != uploadedGeneration {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q generation changed during verification", key)
|
||||
}
|
||||
return *found, nil
|
||||
}
|
||||
|
||||
func closePublishTransfers(transfers []publishTransfer) {
|
||||
for _, transfer := range transfers {
|
||||
if transfer.Source != nil {
|
||||
_ = transfer.Source.Close()
|
||||
}
|
||||
if transfer.TemporaryPath != "" {
|
||||
_ = os.Remove(transfer.TemporaryPath)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func publishSHA256(data []byte) string {
|
||||
sum := sha256.Sum256(data)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
Reference in New Issue
Block a user