From 9afb3550c428a425679f63e0cb09b18f796a38fe Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 8 Jun 2026 19:02:12 +0000 Subject: [PATCH] Implement shared-root publish execution --- docs/config.md | 2 +- docs/integrations/destination-state.md | 6 +- docs/internal/publish.md | 8 +- docs/policy/architecture.md | 2 +- internal/adapters/local/backend.go | 22 ++- internal/adapters/s3/backend.go | 16 +- internal/adapters/ssh/backend.go | 22 ++- internal/app/run_test.go | 82 +++++++++- internal/publish/execute.go | 201 ++++++++++++++++++++++++- internal/publish/plan.go | 4 + internal/publish/shared_root_test.go | 144 +++++++++++++++++- internal/storage/backend.go | 1 + internal/storage/errors.go | 25 +-- internal/storage/fake/backend.go | 18 ++- internal/storage/path.go | 20 ++- internal/storage/path_test.go | 20 +++ 16 files changed, 541 insertions(+), 52 deletions(-) diff --git a/docs/config.md b/docs/config.md index f84e890..2e94be0 100644 --- a/docs/config.md +++ b/docs/config.md @@ -373,7 +373,7 @@ state: `single_owner` state records one pipeline/destination owner for each destination bundle path and is the state mode written by `run`. -`shared_root` is accepted by configuration validation and by destination state parsing for shared-root `.distributor.json` files. Current publish execution writes single-owner destination state. +`shared_root` records multiple pipeline/destination owners in one destination root. Publish execution preserves unrelated owners, rejects path ownership conflicts, and writes shared-root destination state. ## Reconciliation Policy diff --git a/docs/integrations/destination-state.md b/docs/integrations/destination-state.md index bc8db19..58f478f 100644 --- a/docs/integrations/destination-state.md +++ b/docs/integrations/destination-state.md @@ -6,7 +6,7 @@ Each managed destination bundle path contains `.distributor.json`. This file is ## Single-Owner State Schema -Current state written by `run` uses schema version `2`. +State written by `run` for `state.mode: single_owner` uses schema version `2`. ```json { @@ -119,11 +119,11 @@ Normal replacement deletes only managed output paths recorded in `outputs` plus `distributor` can read schema version `1` destination state for compatibility. When v1 state is read, it is treated as single-owner state with `reconciliation.mode: replace`. Missing top-level `created_at` and `updated_at` are inferred from `published_at`, and missing per-output timestamps are also inferred from `published_at`. -Newly written destination state from publish execution uses schema version `2`. +Newly written single-owner destination state from publish execution uses schema version `2`. ## Shared-Root State Schema -`distributor` can parse and validate shared-root destination state with schema version `3`. Publish execution currently writes single-owner state. +State written by `run` for `state.mode: shared_root` uses schema version `3`. ```json { diff --git a/docs/internal/publish.md b/docs/internal/publish.md index 83b3e88..a29ac2e 100644 --- a/docs/internal/publish.md +++ b/docs/internal/publish.md @@ -36,9 +36,9 @@ Execution writes destination state after selected outputs are written. Destinati ## Skip And Resume Behavior -`skip_same` and `skip_destination_newer` execute as no-ops. Replacement-mode updates remove only managed output paths from existing state plus `.distributor.json`, verify the destination is empty, and write state whose outputs are exactly the new plan. Merge-mode updates retain omitted managed outputs, overwrite only paths already recorded as managed, reject unmanaged destination path collisions, and write cumulative output state. Failed writes trigger cleanup where practical; merge cleanup removes only newly created outputs from the failed attempt. +`skip_same` and `skip_destination_newer` execute as no-ops. Replacement-mode single-owner updates remove managed output paths from existing state plus `.distributor.json`, verify the destination is empty, and write state whose outputs are exactly the new plan. Replacement-mode shared-root updates remove only current-owner omitted outputs and preserve unrelated owners. Merge-mode updates retain omitted managed outputs, overwrite only paths already recorded as managed, reject unmanaged destination path collisions, and write cumulative output state. Failed writes trigger cleanup where practical; merge cleanup removes only newly created outputs from the failed attempt. -Shared-root write actions are planned but not executed. Execution rejects shared-root `publish_new`, `replace_older`, and `force_replace` actions. +Shared-root execution writes schema version `3` state. It preserves unrelated owner records and outputs, updates only the publishing owner metadata, preserves root `created_at`, and updates root `updated_at` after successful state writes. Forced replacement is explicit per request and deletes the bounded destination bundle path before writing new outputs and state. @@ -60,11 +60,11 @@ Execution fails on delete, read, transform output, unmanaged merge path collisio - Planning is deterministic for the same request and destination state. - Destination bundle paths are caller-supplied and backend-root-relative. - URL generation uses URL path semantics and never infers public URLs from backend config. -- Replacement reconciliation deletes only managed paths recorded in existing state plus `.distributor.json`. +- Replacement reconciliation deletes only managed paths recorded in existing state plus `.distributor.json` for single-owner state, and only current-owner omitted outputs for shared-root state. - Merge reconciliation never adopts unmanaged content. - Merge state output records are cumulative for the single owner. - Shared-root planning is owner-scoped and preserves unrelated owner outputs. -- Shared-root execution is disabled until the shared-root execution path exists. +- Shared-root execution writes owner-scoped changes without deleting unrelated owners. - Forced replacement deletes only within the supplied destination bundle path. - Destination state is written after selected outputs are written. - Transform resolution stays behind a caller-supplied interface. diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 76f7d60..8e9d9ac 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -96,7 +96,7 @@ The source manifest should remain minimal. Routing, destination selection, publi Each destination bundle path is managed by `.distributor.json`. This file is both the destination sentinel and the destination state record. -Publish execution currently writes single-owner destination state. `internal/state` also parses and validates shared-root destination state, where one `.distributor.json` records multiple pipeline/destination owners and every managed output carries its owner identity. +Publish execution writes single-owner or shared-root destination state according to destination `state.mode`. In shared-root state, one `.distributor.json` records multiple pipeline/destination owners and every managed output carries its owner identity. Single-owner state records: diff --git a/internal/adapters/local/backend.go b/internal/adapters/local/backend.go index d4fe466..df8b8ee 100644 --- a/internal/adapters/local/backend.go +++ b/internal/adapters/local/backend.go @@ -199,10 +199,22 @@ func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) { } func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + return b.deleteManagedTargets(ctx, storage.OpDeleteManagedBundle, func() ([]string, error) { + return storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + }, opts) +} + +func (b *Backend) DeleteManagedOutputs(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + return b.deleteManagedTargets(ctx, storage.OpDeleteManagedOutputs, func() ([]string, error) { + return storage.ManagedOutputTargets(bundlePath, managedOutputPaths) + }, opts) +} + +func (b *Backend) deleteManagedTargets(ctx context.Context, op string, targetsFunc func() ([]string, error), opts storage.DeleteOptions) error { if err := ctx.Err(); err != nil { return err } - targets, err := storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + targets, err := targetsFunc() if err != nil { return err } @@ -213,20 +225,20 @@ func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, ma return err } if nativePath == b.root { - return storage.NewError(storage.OpDeleteManagedBundle, backendName, logicalPath, storage.ErrInvalidPath, nil) + return storage.NewError(op, backendName, logicalPath, storage.ErrInvalidPath, nil) } info, err := os.Lstat(nativePath) if err != nil { if opts.IgnoreMissing && errors.Is(err, fs.ErrNotExist) { continue } - return b.translateError(storage.OpDeleteManagedBundle, logicalPath, err) + return b.translateError(op, logicalPath, err) } if info.IsDir() { - return storage.NewError(storage.OpDeleteManagedBundle, backendName, logicalPath, storage.ErrUnsupported, nil) + return storage.NewError(op, backendName, logicalPath, storage.ErrUnsupported, nil) } if err := os.Remove(nativePath); err != nil { - return b.translateError(storage.OpDeleteManagedBundle, logicalPath, err) + return b.translateError(op, logicalPath, err) } if opts.PruneEmptyDirs { b.pruneEmptyParents(filepath.Dir(nativePath)) diff --git a/internal/adapters/s3/backend.go b/internal/adapters/s3/backend.go index 576c15d..1479f82 100644 --- a/internal/adapters/s3/backend.go +++ b/internal/adapters/s3/backend.go @@ -205,15 +205,27 @@ func (b *Backend) HasAny(ctx context.Context, logicalPrefix string) (bool, error } func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + return b.deleteManagedTargets(ctx, storage.OpDeleteManagedBundle, func() ([]string, error) { + return storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + }, opts) +} + +func (b *Backend) DeleteManagedOutputs(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + return b.deleteManagedTargets(ctx, storage.OpDeleteManagedOutputs, func() ([]string, error) { + return storage.ManagedOutputTargets(bundlePath, managedOutputPaths) + }, opts) +} + +func (b *Backend) deleteManagedTargets(ctx context.Context, op string, targetsFunc func() ([]string, error), opts storage.DeleteOptions) error { if err := ctx.Err(); err != nil { return err } - targets, err := storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + targets, err := targetsFunc() if err != nil { return err } for _, target := range targets { - if err := b.deleteObject(ctx, storage.OpDeleteManagedBundle, target, opts); err != nil { + if err := b.deleteObject(ctx, op, target, opts); err != nil { return err } } diff --git a/internal/adapters/ssh/backend.go b/internal/adapters/ssh/backend.go index dae5236..713fd7e 100644 --- a/internal/adapters/ssh/backend.go +++ b/internal/adapters/ssh/backend.go @@ -223,10 +223,22 @@ func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) { } func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + return b.deleteManagedTargets(ctx, storage.OpDeleteManagedBundle, func() ([]string, error) { + return storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + }, opts) +} + +func (b *Backend) DeleteManagedOutputs(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + return b.deleteManagedTargets(ctx, storage.OpDeleteManagedOutputs, func() ([]string, error) { + return storage.ManagedOutputTargets(bundlePath, managedOutputPaths) + }, opts) +} + +func (b *Backend) deleteManagedTargets(ctx context.Context, op string, targetsFunc func() ([]string, error), opts storage.DeleteOptions) error { if err := ctx.Err(); err != nil { return err } - targets, err := storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + targets, err := targetsFunc() if err != nil { return err } @@ -236,20 +248,20 @@ func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, ma return err } if nativePath == b.root { - return storage.NewError(storage.OpDeleteManagedBundle, BackendName, target, storage.ErrInvalidPath, nil) + return storage.NewError(op, BackendName, target, storage.ErrInvalidPath, nil) } info, err := b.client.Lstat(nativePath) if err != nil { if opts.IgnoreMissing && isNotExist(err) { continue } - return b.translateError(storage.OpDeleteManagedBundle, target, err) + return b.translateError(op, target, err) } if info.IsDir() { - return storage.NewError(storage.OpDeleteManagedBundle, BackendName, target, storage.ErrUnsupported, nil) + return storage.NewError(op, BackendName, target, storage.ErrUnsupported, nil) } if err := b.client.Remove(nativePath); err != nil { - return b.translateError(storage.OpDeleteManagedBundle, target, err) + return b.translateError(op, target, err) } if opts.PruneEmptyDirs { b.pruneEmptyParents(parentOf(target)) diff --git a/internal/app/run_test.go b/internal/app/run_test.go index b131f63..fd3c459 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -222,7 +222,7 @@ func TestRunPublishesNewLocalBundle(t *testing.T) { } } -func TestRunThreadsSharedRootStateModeIntoPublish(t *testing.T) { +func TestRunSharedRootDryRunWritesNoOutputsOrState(t *testing.T) { sourceRoot := t.TempDir() destinationRoot := t.TempDir() writeSourceBundle(t, sourceRoot, "", testBundleOptions{ @@ -241,16 +241,48 @@ func TestRunThreadsSharedRootStateModeIntoPublish(t *testing.T) { if got, want := report.Actions[0].Action, string(publish.ActionPublishNew); got != want { t.Fatalf("dry-run action = %q, want %q", got, want) } - - err = Run(context.Background(), RunOptions{ConfigPath: configPath}) - if err == nil || !strings.Contains(err.Error(), "shared-root publish execution is not implemented") { - t.Fatalf("Run() error = %v, want shared-root execution guard", err) + if _, statErr := os.Stat(filepath.Join(destinationRoot, "report.md")); !os.IsNotExist(statErr) { + t.Fatalf("output stat error = %v, want absent", statErr) } if _, statErr := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); !os.IsNotExist(statErr) { t.Fatalf("state file stat error = %v, want absent", statErr) } } +func TestRunPublishesTwoPipelinesIntoSharedRoot(t *testing.T) { + firstSourceRoot := t.TempDir() + secondSourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, firstSourceRoot, "", testBundleOptions{ + ID: "reports.first", + Files: []testFile{{Path: "first.md", Data: "# First\n"}}, + }) + writeSourceBundle(t, secondSourceRoot, "", testBundleOptions{ + ID: "reports.second", + Files: []testFile{{Path: "second.md", Data: "# Second\n"}}, + }) + + err := Run(context.Background(), RunOptions{ConfigPath: writeTwoPipelineSharedRootConfig(t, firstSourceRoot, secondSourceRoot, destinationRoot)}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + testutil.AssertFile(t, filepath.Join(destinationRoot, "first.md"), "# First\n") + testutil.AssertFile(t, filepath.Join(destinationRoot, "second.md"), "# Second\n") + destinationState := readSharedRootStateFile(t, filepath.Join(destinationRoot, storage.StateFileName)) + if got, want := len(destinationState.Owners), 2; got != want { + t.Fatalf("owner count = %d, want %d", got, want) + } + if _, ok := destinationState.Owner(state.CurrentOwnerScope("reports-first", "archive")); !ok { + t.Fatal("reports-first/archive owner missing") + } + if _, ok := destinationState.Owner(state.CurrentOwnerScope("reports-second", "archive")); !ok { + t.Fatal("reports-second/archive owner missing") + } + if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "first.md,second.md"; got != want { + t.Fatalf("managed paths = %q, want %q", got, want) + } +} + func TestRunPipelineWithLocalSourcePublishesConfiguredDestination(t *testing.T) { sourceRoot := t.TempDir() destinationRoot := t.TempDir() @@ -1764,6 +1796,33 @@ pipelines: `) } +func writeTwoPipelineSharedRootConfig(t *testing.T, firstSourceRoot, secondSourceRoot, destinationRoot string) string { + t.Helper() + return writeConfigFile(t, ` +pipelines: + - id: reports-first + source: + backend: local + path: `+firstSourceRoot+` + destinations: + - id: archive + backend: local + path: `+destinationRoot+` + state: + mode: shared_root + - id: reports-second + source: + backend: local + path: `+secondSourceRoot+` + destinations: + - id: archive + backend: local + path: `+destinationRoot+` + state: + mode: shared_root +`) +} + func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string { t.Helper() return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination) @@ -1844,6 +1903,19 @@ func readStateFile(t *testing.T, path string) state.DistributorState { return testutil.ReadDestinationState(t, path) } +func readSharedRootStateFile(t *testing.T, path string) state.SharedRootState { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read shared-root state: %v", err) + } + destinationState, err := state.ParseSharedRoot(data) + if err != nil { + t.Fatalf("parse shared-root state: %v", err) + } + return destinationState +} + func outputsByPath(outputs []state.OutputFile) map[string]state.OutputFile { byPath := make(map[string]state.OutputFile, len(outputs)) for _, output := range outputs { diff --git a/internal/publish/execute.go b/internal/publish/execute.go index d4cdc30..948bed9 100644 --- a/internal/publish/execute.go +++ b/internal/publish/execute.go @@ -17,8 +17,8 @@ func Execute(ctx context.Context, req Request, plan Plan) error { case ActionSkipSame, ActionSkipDestinationNewer: return nil case ActionPublishNew, ActionReplaceOlder, ActionForceReplace: - if normalizeState(req.State).Mode == config.StateModeSharedRoot || plan.StateMode == config.StateModeSharedRoot { - return fmt.Errorf("shared-root publish execution is not implemented") + if usesSharedRootState(req, plan) { + return executeSharedRoot(ctx, req, plan) } default: return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason) @@ -138,6 +138,89 @@ func Execute(ctx context.Context, req Request, plan Plan) error { return nil } +func executeSharedRoot(ctx context.Context, req Request, plan Plan) error { + plan.Reconciliation = normalizeReconciliation(plan.Reconciliation) + if plan.Action == ActionForceReplace { + if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil { + return err + } + if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil { + return err + } + } + if plan.Action == ActionReplaceOlder && plan.Reconciliation.Mode == config.ReconciliationModeReplace { + if err := req.DestinationBackend.DeleteManagedOutputs(ctx, req.DestinationBundlePath, sharedRootOutputPaths(plan.OwnerOutputsToDelete), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil { + return err + } + } + + writtenOutputs := make([]Output, 0, len(plan.Outputs)) + newOutputs := make([]Output, 0, len(plan.Outputs)) + cleanup := func() { + outputs := writtenOutputs + if plan.Reconciliation.Mode == config.ReconciliationModeMerge { + outputs = newOutputs + } + _ = req.DestinationBackend.DeleteManagedOutputs(ctx, req.DestinationBundlePath, ManagedOutputPaths(outputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}) + } + for _, output := range plan.Outputs { + destinationPath, err := storage.Join(req.DestinationBundlePath, output.DestinationPath) + if err != nil { + cleanup() + return err + } + data := output.Data + if output.Kind == state.OutputKindSource { + sourcePath, err := storage.Join(req.SourceBundle.RootRelativePath, output.SourcePath) + if err != nil { + cleanup() + return err + } + data, err = req.SourceBackend.ReadFile(ctx, sourcePath) + if err != nil { + cleanup() + return err + } + } + managed := outputManagedBySharedRootPlan(output, plan) + if _, err := req.DestinationBackend.WriteFile(ctx, destinationPath, data, storage.WriteOptions{Overwrite: managed, PreferAtomic: true}); err != nil { + cleanup() + return err + } + writtenOutputs = append(writtenOutputs, output) + if !managed { + newOutputs = append(newOutputs, output) + } + } + + now := time.Now().UTC() + sharedRootState, err := sharedRootStateForPlan(req, plan, now) + if err != nil { + cleanup() + return err + } + if err := state.ValidateSharedRoot(sharedRootState); err != nil { + cleanup() + return err + } + data, err := json.MarshalIndent(sharedRootState, "", " ") + if err != nil { + cleanup() + return err + } + data = append(data, '\n') + statePath, err := storage.StatePath(req.DestinationBundlePath) + if err != nil { + cleanup() + return err + } + if _, err := req.DestinationBackend.WriteFile(ctx, statePath, data, storage.WriteOptions{Overwrite: sharedRootStateWriteOverwrites(plan), PreferAtomic: true}); err != nil { + cleanup() + return err + } + return nil +} + func ensureMergeOutputPaths(ctx context.Context, backend storage.Backend, bundlePath string, plan Plan) error { for _, output := range plan.Outputs { if outputManagedByExistingState(output, plan.ExistingState) { @@ -156,6 +239,10 @@ func ensureMergeOutputPaths(ctx context.Context, backend storage.Backend, bundle return nil } +func usesSharedRootState(req Request, plan Plan) bool { + return normalizeState(req.State).Mode == config.StateModeSharedRoot || plan.StateMode == config.StateModeSharedRoot +} + func outputManagedByExistingState(output Output, existing *state.DistributorState) bool { if existing == nil { return false @@ -164,6 +251,15 @@ func outputManagedByExistingState(output Output, existing *state.DistributorStat return ok } +func outputManagedBySharedRootPlan(output Output, plan Plan) bool { + for _, existing := range currentOwnerSharedRootOutputs(plan) { + if existing.Path == output.DestinationPath { + return true + } + } + return false +} + func stateOutputsForPlan(plan Plan, now time.Time) ([]state.OutputFile, error) { existingOutputs := []state.OutputFile(nil) if plan.ExistingState != nil { @@ -179,3 +275,104 @@ func stateOutputsForPlan(plan Plan, now time.Time) ([]state.OutputFile, error) { func usesMergeRetention(plan Plan) bool { return plan.Reconciliation.Mode == config.ReconciliationModeMerge && plan.Action == ActionReplaceOlder } + +func sharedRootStateForPlan(req Request, plan Plan, now time.Time) (state.SharedRootState, error) { + now = now.UTC() + scope := plan.OwnerScope + if scope.PipelineID == "" && scope.DestinationID == "" { + scope = state.CurrentOwnerScope(req.PipelineID, req.DestinationID) + } + base := sharedRootBaseState(req, plan, now) + owner := state.OwnerRecord{ + Scope: scope, + Reconciliation: state.ReconciliationPolicy{Mode: plan.Reconciliation.Mode}, + Source: state.SourceState{Manifest: req.SourceBundle.Manifest}, + } + if plan.PrimaryURL != "" { + owner.Links = &state.LinkState{PrimaryURL: plan.PrimaryURL} + } + planned := state.ProjectSharedRootOutputs(StateOutputProjections(plan.Outputs), currentOwnerSharedRootOutputs(plan), scope, req.SourceBundle.Manifest, now) + if plan.Action == ActionReplaceOlder && plan.Reconciliation.Mode == config.ReconciliationModeMerge { + return state.MergeOwnerOutputs(base, scope, owner, planned) + } + return state.ReplaceOwnerOutputs(base, scope, owner, planned) +} + +func sharedRootBaseState(req Request, plan Plan, now time.Time) state.SharedRootState { + if plan.Action == ActionForceReplace { + return newSharedRootState(req, now) + } + if plan.ExistingSharedRoot != nil { + base := *plan.ExistingSharedRoot + base.Owners = append([]state.OwnerRecord(nil), plan.ExistingSharedRoot.Owners...) + base.Outputs = append([]state.SharedRootOutputFile(nil), plan.ExistingSharedRoot.Outputs...) + base.DistributorVersion = req.DistributorVersion + base.UpdatedAt = now + return base + } + if plan.ExistingState != nil { + base := newSharedRootState(req, now) + base.CreatedAt = plan.ExistingState.CreatedAt + base.UpdatedAt = now + return base + } + return newSharedRootState(req, now) +} + +func newSharedRootState(req Request, now time.Time) state.SharedRootState { + return state.SharedRootState{ + SchemaVersion: state.SharedRootSchemaVersion, + DistributorVersion: req.DistributorVersion, + CreatedAt: now, + UpdatedAt: now, + State: state.StatePolicy{Mode: state.StateModeSharedRoot}, + Owners: []state.OwnerRecord{}, + Outputs: []state.SharedRootOutputFile{}, + } +} + +func currentOwnerSharedRootOutputs(plan Plan) []state.SharedRootOutputFile { + if plan.ExistingSharedRoot != nil { + outputs := make([]state.SharedRootOutputFile, 0, len(plan.ExistingSharedRoot.Outputs)) + for _, output := range plan.ExistingSharedRoot.Outputs { + if output.Owner == plan.OwnerScope { + outputs = append(outputs, output) + } + } + return outputs + } + if plan.ExistingState != nil { + outputs := make([]state.SharedRootOutputFile, 0, len(plan.ExistingState.Outputs)) + for _, output := range plan.ExistingState.Outputs { + outputs = append(outputs, state.SharedRootOutputFile{ + Path: output.Path, + Kind: output.Kind, + SourcePath: output.SourcePath, + Transform: output.Transform, + URL: output.URL, + SHA256: output.SHA256, + Size: output.Size, + Owner: plan.OwnerScope, + SourceID: plan.ExistingState.Source.Manifest.ID, + SourceDigest: plan.ExistingState.Source.Manifest.Digest, + SourceCreated: plan.ExistingState.Source.Manifest.Created, + CreatedAt: output.CreatedAt, + UpdatedAt: output.UpdatedAt, + }) + } + return outputs + } + return nil +} + +func sharedRootOutputPaths(outputs []state.SharedRootOutputFile) []string { + paths := make([]string, 0, len(outputs)) + for _, output := range outputs { + paths = append(paths, output.Path) + } + return paths +} + +func sharedRootStateWriteOverwrites(plan Plan) bool { + return plan.ExistingSharedRoot != nil || plan.ExistingState != nil || plan.Action == ActionForceReplace +} diff --git a/internal/publish/plan.go b/internal/publish/plan.go index 34459f1..5314df7 100644 --- a/internal/publish/plan.go +++ b/internal/publish/plan.go @@ -240,6 +240,10 @@ func planSharedRootOwner(ctx context.Context, req Request, status state.Destinat } plannedPaths := outputPaths(outputs) + if action == ActionForceReplace { + details.OwnerOutputsToWrite = append([]Output(nil), outputs...) + return details, nil + } if conflict, ok := sharedRootPathOwnershipConflict(status, scope, plannedPaths); ok { reason := fmt.Sprintf("destination output path %s is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID) details.Action = ActionFailConflict diff --git a/internal/publish/shared_root_test.go b/internal/publish/shared_root_test.go index b548e4f..25b9966 100644 --- a/internal/publish/shared_root_test.go +++ b/internal/publish/shared_root_test.go @@ -132,19 +132,131 @@ func TestBuildSharedRootRejectsUnmanagedPathCollision(t *testing.T) { } } -func TestExecuteSharedRootWriteActionsAreDisabled(t *testing.T) { +func TestExecuteSharedRootPublishesOwnerAndPreservesOtherOwners(t *testing.T) { sourceBackend := fake.New() sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}}, }) destinationBackend := fake.New() + writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRootStateWithOwners(t, sourceBundle.Manifest, false)) req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) plan, err := Build(context.Background(), req) if err != nil { t.Fatalf("Build() error = %v", err) } - if err := Execute(context.Background(), req, plan); err == nil || !strings.Contains(err.Error(), "shared-root publish execution is not implemented") { - t.Fatalf("Execute() error = %v, want shared-root disabled error", err) + if err := Execute(context.Background(), req, plan); err != nil { + t.Fatalf("Execute() error = %v", err) + } + testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\n") + testutil.AssertFakeFile(t, destinationBackend, "bundle/other/report.md", "old") + destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") + if got, want := len(destinationState.Owners), 2; got != want { + t.Fatalf("owner count = %d, want %d", got, want) + } + if got, want := destinationState.State.Mode, state.StateModeSharedRoot; got != want { + t.Fatalf("state mode = %q, want %q", got, want) + } + if _, ok := destinationState.Owner(state.CurrentOwnerScope("other", "archive")); !ok { + t.Fatal("other owner missing from shared-root state") + } + if _, ok := destinationState.Owner(state.CurrentOwnerScope("reports", "archive")); !ok { + t.Fatal("current owner missing from shared-root state") + } + if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "other/report.md,report.md"; got != want { + t.Fatalf("managed paths = %q, want %q", got, want) + } +} + +func TestExecuteSharedRootReplaceDeletesOnlyCurrentOwnerOmittedOutputs(t *testing.T) { + sourceBackend := fake.New() + sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ + Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, + }) + destinationBackend := fake.New() + existing := sharedRootStateWithOwners(t, sourceBundle.Manifest, true) + writeFakeSharedRootState(t, destinationBackend, "bundle", existing) + + req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if err := Execute(context.Background(), req, plan); err != nil { + t.Fatalf("Execute() error = %v", err) + } + testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nNew.\n") + testutil.AssertFakeMissing(t, destinationBackend, "bundle/old.md") + testutil.AssertFakeFile(t, destinationBackend, "bundle/other/report.md", "old") + destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") + if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "other/report.md,report.md"; got != want { + t.Fatalf("managed paths = %q, want %q", got, want) + } + if !destinationState.CreatedAt.Equal(existing.CreatedAt) { + t.Fatalf("created_at = %s, want %s", destinationState.CreatedAt, existing.CreatedAt) + } + output, ok := findSharedRootOutputForTest(destinationState.Outputs, "report.md") + if !ok { + t.Fatal("report.md missing from shared-root outputs") + } + if !output.CreatedAt.Equal(existing.Outputs[1].CreatedAt) || !output.UpdatedAt.After(existing.Outputs[1].UpdatedAt) { + t.Fatalf("report.md timestamps = created:%s updated:%s, want preserved created and newer updated", output.CreatedAt, output.UpdatedAt) + } +} + +func TestExecuteSharedRootMergeRetainsCurrentOwnerOmittedOutputs(t *testing.T) { + sourceBackend := fake.New() + sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ + Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, + }) + destinationBackend := fake.New() + writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRootStateWithOwners(t, sourceBundle.Manifest, true)) + + req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge) + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if err := Execute(context.Background(), req, plan); err != nil { + t.Fatalf("Execute() error = %v", err) + } + testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nNew.\n") + testutil.AssertFakeFile(t, destinationBackend, "bundle/old.md", "old") + testutil.AssertFakeFile(t, destinationBackend, "bundle/other/report.md", "old") + destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") + if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "other/report.md,report.md,old.md"; got != want { + t.Fatalf("managed paths = %q, want %q", got, want) + } +} + +func TestExecuteSharedRootForceReplaceDeletesOnlyBundlePath(t *testing.T) { + sourceBackend := fake.New() + sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ + Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}}, + }) + destinationBackend := fake.New() + testutil.WriteFakeFile(t, destinationBackend, "bundle/unmanaged.txt", "unmanaged") + testutil.WriteFakeFile(t, destinationBackend, "bundle/nested/old.txt", "old") + testutil.WriteFakeFile(t, destinationBackend, "bundle-sibling/keep.txt", "keep") + + req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) + req.Force = true + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionForceReplace { + t.Fatalf("plan action = %s, want force_replace", plan.Action) + } + if err := Execute(context.Background(), req, plan); err != nil { + t.Fatalf("Execute() error = %v", err) + } + testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\n") + testutil.AssertFakeMissing(t, destinationBackend, "bundle/unmanaged.txt") + testutil.AssertFakeMissing(t, destinationBackend, "bundle/nested/old.txt") + testutil.AssertFakeFile(t, destinationBackend, "bundle-sibling/keep.txt", "keep") + destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") + if got, want := len(destinationState.Owners), 1; got != want { + t.Fatalf("owner count = %d, want %d", got, want) } } @@ -184,6 +296,32 @@ func writeFakeSharedRootState(t *testing.T, backend *fake.Backend, relative stri } } +func readFakeSharedRootState(t *testing.T, backend *fake.Backend, relative string) state.SharedRootState { + t.Helper() + statePath, err := storage.StatePath(relative) + if err != nil { + t.Fatalf("state path: %v", err) + } + data, err := backend.ReadFile(context.Background(), statePath) + if err != nil { + t.Fatalf("read shared-root state: %v", err) + } + destinationState, err := state.ParseSharedRoot(data) + if err != nil { + t.Fatalf("parse shared-root state: %v", err) + } + return destinationState +} + +func findSharedRootOutputForTest(outputs []state.SharedRootOutputFile, path string) (state.SharedRootOutputFile, bool) { + for _, output := range outputs { + if output.Path == path { + return output, true + } + } + return state.SharedRootOutputFile{}, false +} + func sharedRootStateWithOwners(t *testing.T, current bundle.Manifest, includeCurrent bool) state.SharedRootState { t.Helper() createdAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC) diff --git a/internal/storage/backend.go b/internal/storage/backend.go index 0cb1a21..6da7874 100644 --- a/internal/storage/backend.go +++ b/internal/storage/backend.go @@ -29,6 +29,7 @@ type Backend interface { Stat(ctx context.Context, path string) (Entry, error) Walk(ctx context.Context, prefix string, opts WalkOptions, fn WalkFunc) error HasAny(ctx context.Context, prefix string) (bool, error) + DeleteManagedOutputs(ctx context.Context, bundlePath string, managedOutputPaths []string, opts DeleteOptions) error DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts DeleteOptions) error DeletePrefix(ctx context.Context, prefix string, opts DeleteOptions) error } diff --git a/internal/storage/errors.go b/internal/storage/errors.go index 3cd123c..ac575d1 100644 --- a/internal/storage/errors.go +++ b/internal/storage/errors.go @@ -20,18 +20,19 @@ const ( ) const ( - OpValidatePath = "validate path" - OpReadFile = "read file" - OpOpenReader = "open reader" - OpWriteFile = "write file" - OpWriteFrom = "write stream" - OpStat = "stat" - OpWalk = "walk" - OpHasAny = "has any" - OpDeleteManagedBundle = "delete managed bundle" - OpDeletePrefix = "delete prefix" - OpRegisterBackend = "register backend" - OpOpenBackend = "open backend" + OpValidatePath = "validate path" + OpReadFile = "read file" + OpOpenReader = "open reader" + OpWriteFile = "write file" + OpWriteFrom = "write stream" + OpStat = "stat" + OpWalk = "walk" + OpHasAny = "has any" + OpDeleteManagedOutputs = "delete managed outputs" + OpDeleteManagedBundle = "delete managed bundle" + OpDeletePrefix = "delete prefix" + OpRegisterBackend = "register backend" + OpOpenBackend = "open backend" ) type Error struct { diff --git a/internal/storage/fake/backend.go b/internal/storage/fake/backend.go index 9ad7033..112ea84 100644 --- a/internal/storage/fake/backend.go +++ b/internal/storage/fake/backend.go @@ -159,23 +159,35 @@ func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) { } func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + return b.deleteManagedTargets(ctx, storage.OpDeleteManagedBundle, func() ([]string, error) { + return storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + }, opts) +} + +func (b *Backend) DeleteManagedOutputs(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + return b.deleteManagedTargets(ctx, storage.OpDeleteManagedOutputs, func() ([]string, error) { + return storage.ManagedOutputTargets(bundlePath, managedOutputPaths) + }, opts) +} + +func (b *Backend) deleteManagedTargets(ctx context.Context, op string, targetsFunc func() ([]string, error), opts storage.DeleteOptions) error { if err := ctx.Err(); err != nil { return err } - targets, err := storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + targets, err := targetsFunc() if err != nil { return err } for _, target := range targets { if _, ok := b.dirs[target]; ok { - return storage.NewError(storage.OpDeleteManagedBundle, backendName, target, storage.ErrUnsupported, nil) + return storage.NewError(op, backendName, target, storage.ErrUnsupported, nil) } if !b.exists(target) { if opts.IgnoreMissing { continue } - return storage.NewError(storage.OpDeleteManagedBundle, backendName, target, storage.ErrNotFound, nil) + return storage.NewError(op, backendName, target, storage.ErrNotFound, nil) } delete(b.files, target) delete(b.symlinks, target) diff --git a/internal/storage/path.go b/internal/storage/path.go index 496bb88..32b735d 100644 --- a/internal/storage/path.go +++ b/internal/storage/path.go @@ -50,10 +50,23 @@ func DisplayPath(path string) string { } func ManagedBundleTargets(bundlePath string, managedOutputPaths []string) ([]string, error) { + targets, err := ManagedOutputTargets(bundlePath, managedOutputPaths) + if err != nil { + return nil, err + } + statePath, err := StatePath(bundlePath) + if err != nil { + return nil, err + } + targets = append(targets, statePath) + return targets, nil +} + +func ManagedOutputTargets(bundlePath string, managedOutputPaths []string) ([]string, error) { if err := ValidatePrefix(bundlePath); err != nil { return nil, err } - targets := make([]string, 0, len(managedOutputPaths)+1) + targets := make([]string, 0, len(managedOutputPaths)) for _, outputPath := range managedOutputPaths { target, err := Join(bundlePath, outputPath) if err != nil { @@ -61,11 +74,6 @@ func ManagedBundleTargets(bundlePath string, managedOutputPaths []string) ([]str } targets = append(targets, target) } - statePath, err := StatePath(bundlePath) - if err != nil { - return nil, err - } - targets = append(targets, statePath) return targets, nil } diff --git a/internal/storage/path_test.go b/internal/storage/path_test.go index 4767b57..7394a37 100644 --- a/internal/storage/path_test.go +++ b/internal/storage/path_test.go @@ -132,6 +132,22 @@ func TestManagedBundleTargetsRejectsInvalidOutputPath(t *testing.T) { } } +func TestManagedOutputTargetsOmitsStateFile(t *testing.T) { + targets, err := ManagedOutputTargets("bundle", []string{"report.md", "nested/report.html"}) + if err != nil { + t.Fatalf("ManagedOutputTargets() error = %v", err) + } + want := []string{"bundle/report.md", "bundle/nested/report.html"} + if len(targets) != len(want) { + t.Fatalf("targets = %v, want %v", targets, want) + } + for index := range want { + if targets[index] != want[index] { + t.Fatalf("targets = %v, want %v", targets, want) + } + } +} + func TestListSortsEntries(t *testing.T) { backend := walkBackend{ entries: []Entry{ @@ -215,6 +231,10 @@ func (b walkBackend) DeleteManagedBundle(context.Context, string, []string, Dele return nil } +func (b walkBackend) DeleteManagedOutputs(context.Context, string, []string, DeleteOptions) error { + return nil +} + func (b walkBackend) DeletePrefix(context.Context, string, DeleteOptions) error { return nil }