diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 8818871..1655b73 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -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 diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index ad064ea..3b66f19 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -195,7 +195,7 @@ file-level ownership. ## Stage 5 — Separate Transfer And Commit Mechanics -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/stage/publish.go b/internal/stage/publish.go index d72c75b..83c541c 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -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, diff --git a/internal/stage/publish_commit.go b/internal/stage/publish_commit.go new file mode 100644 index 0000000..6a35c1c --- /dev/null +++ b/internal/stage/publish_commit.go @@ -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 +} diff --git a/internal/stage/publish_transfer.go b/internal/stage/publish_transfer.go new file mode 100644 index 0000000..a350da1 --- /dev/null +++ b/internal/stage/publish_transfer.go @@ -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[:]) +}