eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Implement shared-root publish execution

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-08 19:02:12 +00:00
parent 89169f810f
commit 9afb3550c4
16 changed files with 541 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/config.md
View File

@@ -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

6
docs/integrations/destination-state.md
View File

@@ -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
{

8
docs/internal/publish.md
View File

@@ -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.

2
docs/policy/architecture.md
View File

@@ -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:

22
internal/adapters/local/backend.go
View File

@@ -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))

16
internal/adapters/s3/backend.go
View File

@@ -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
}
}

22
internal/adapters/ssh/backend.go
View File

@@ -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))

82
internal/app/run_test.go
View File

@@ -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 {

201
internal/publish/execute.go
View File

@@ -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
}

4
internal/publish/plan.go
View File

@@ -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

144
internal/publish/shared_root_test.go
View File

@@ -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)

1
internal/storage/backend.go
View File

@@ -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
}

1
internal/storage/errors.go
View File

@@ -28,6 +28,7 @@ const (
OpStat = "stat"
OpWalk = "walk"
OpHasAny = "has any"
OpDeleteManagedOutputs = "delete managed outputs"
OpDeleteManagedBundle = "delete managed bundle"
OpDeletePrefix = "delete prefix"
OpRegisterBackend = "register backend"

18
internal/storage/fake/backend.go
View File

@@ -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)

20
internal/storage/path.go
View File

@@ -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
}

20
internal/storage/path_test.go
View File

@@ -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
}