Add shared-root publish planning

This commit is contained in:
2026-06-08 18:53:22 +00:00
parent eb86cf9ab6
commit 89169f810f
10 changed files with 629 additions and 23 deletions

View File

@@ -8,9 +8,9 @@ Audience: developers and LLM coding agents changing `internal/publish`.
## Inputs And Outputs ## Inputs And Outputs
Inputs are a source bundle, source backend, destination backend, pipeline id, destination id, destination bundle path, path mapping mode, publish policy, transform policy, optional link policy, reconciliation policy, transformer resolver, transfer policy, distributor version, and force flag. Inputs are a source bundle, source backend, destination backend, pipeline id, destination id, destination bundle path, path mapping mode, publish policy, transform policy, optional link policy, state policy, reconciliation policy, transformer resolver, transfer policy, distributor version, and force flag.
Output from planning is a `Plan` with action, reason, destination identity, selected outputs, reconciliation mode, optional existing state, optional primary URL, and force metadata. Execution writes selected source outputs, generated outputs, and `.distributor.json` for executable publish or replacement actions. Output from planning is a `Plan` with action, reason, destination identity, selected outputs, state mode, owner scope, reconciliation mode, optional existing single-owner or shared-root state, optional primary URL, and force metadata. Shared-root plans also expose other-owner outputs to preserve, current-owner outputs retained by merge, current-owner outputs deleted by replace, and current-owner outputs to write. Execution writes selected source outputs, generated outputs, and `.distributor.json` for executable single-owner publish or replacement actions.
## Boundaries ## Boundaries
@@ -20,7 +20,7 @@ External destination state semantics are documented in `docs/integrations/destin
## Config Fields Used ## Config Fields Used
The package consumes already-defaulted config values for destination `publish`, `transform`, `links`, `reconciliation`, `transfer`, and path mapping mode. It uses `config.ValidatePublishTransformPolicy` for publish/transform consistency. The package consumes already-defaulted config values for destination `publish`, `transform`, `links`, `state`, `reconciliation`, `transfer`, and path mapping mode. It uses `config.ValidatePublishTransformPolicy` for publish/transform consistency.
## Adapters Used ## Adapters Used
@@ -30,17 +30,21 @@ The package depends on `internal/storage.Backend` for source and destination IO,
Planning inspects destination state through `internal/state`, compares it with the source manifest, and maps comparison outcomes plus transfer policy into actions: `publish_new`, `replace_older`, `force_replace`, `skip_same`, `skip_destination_newer`, `fail_conflict`, or `fail_unmanaged`. Planning inspects destination state through `internal/state`, compares it with the source manifest, and maps comparison outcomes plus transfer policy into actions: `publish_new`, `replace_older`, `force_replace`, `skip_same`, `skip_destination_newer`, `fail_conflict`, or `fail_unmanaged`.
Single-owner destinations compare the whole destination state against the configured pipeline and destination ids. Shared-root destinations compare only the current owner scope, keyed by pipeline id and destination id. An absent shared-root owner is publishable for that owner unless a planned output collides with unmanaged storage content. Planned writes to a path owned by another shared-root owner fail as conflicts.
Execution writes destination state after selected outputs are written. Destination state includes copied source output metadata, generated output metadata, output timestamps, embedded source manifest, reconciliation metadata, link metadata when configured, pipeline id, destination id, and publication timestamps. Execution writes destination state after selected outputs are written. Destination state includes copied source output metadata, generated output metadata, output timestamps, embedded source manifest, reconciliation metadata, link metadata when configured, pipeline id, destination id, and publication timestamps.
## Skip And Resume Behavior ## 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 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.
Shared-root write actions are planned but not executed. Execution rejects shared-root `publish_new`, `replace_older`, and `force_replace` actions.
Forced replacement is explicit per request and deletes the bounded destination bundle path before writing new outputs and state. Forced replacement is explicit per request and deletes the bounded destination bundle path before writing new outputs and state.
## Failure Behavior ## Failure Behavior
Planning fails for incomplete requests, invalid publish/transform policy, invalid reconciliation mode, output path collisions, invalid destination state, unmanaged destination content without force, conflict outcomes not allowed by transfer policy, unresolved transforms, invalid Markdown output selection, and invalid link URL planning. Planning fails for incomplete requests, invalid publish/transform policy, invalid state mode, invalid reconciliation mode, output path collisions, invalid destination state, unmanaged destination content without force, shared-root owner path conflicts, conflict outcomes not allowed by transfer policy, unresolved transforms, invalid Markdown output selection, and invalid link URL planning.
Execution fails on delete, read, transform output, unmanaged merge path collision, write, state validation, state serialization, or context errors. Execution refuses actions that are not executable publish or replacement actions. Execution fails on delete, read, transform output, unmanaged merge path collision, write, state validation, state serialization, or context errors. Execution refuses actions that are not executable publish or replacement actions.
@@ -59,6 +63,8 @@ Execution fails on delete, read, transform output, unmanaged merge path collisio
- 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`.
- Merge reconciliation never adopts unmanaged content. - Merge reconciliation never adopts unmanaged content.
- Merge state output records are cumulative for the single owner. - 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.
- Forced replacement deletes only within the supplied destination bundle path. - Forced replacement deletes only within the supplied destination bundle path.
- Destination state is written after selected outputs are written. - Destination state is written after selected outputs are written.
- Transform resolution stays behind a caller-supplied interface. - Transform resolution stays behind a caller-supplied interface.

View File

@@ -42,6 +42,8 @@ For shared-root state, helpers parse either state shape, identify the current ow
Comparison is pure. It returns outcomes for absent state, unmanaged content, invalid state, pipeline/destination mismatch, same source manifest, older destination, newer destination, same-created digest conflict, and different source id conflict. It does not decide whether to skip, replace, force, or fail; publish planning maps outcomes to actions. Comparison is pure. It returns outcomes for absent state, unmanaged content, invalid state, pipeline/destination mismatch, same source manifest, older destination, newer destination, same-created digest conflict, and different source id conflict. It does not decide whether to skip, replace, force, or fail; publish planning maps outcomes to actions.
Shared-root comparison is owner-scoped. It compares only the owner keyed by the current pipeline id and destination id, treats an absent owner as absent destination state for that owner, and can compare compatible single-owner state for the current owner without converting unrelated single-owner state.
## Failure Behavior ## Failure Behavior
Parsing rejects invalid JSON, trailing data, missing required fields, invalid timestamps, invalid state mode, invalid reconciliation mode, invalid embedded manifests, duplicate owners, duplicate outputs, invalid output paths, unsupported output kinds, missing generated transforms, invalid URLs, invalid digests, negative sizes, and shared-root outputs whose owner is not registered. Parsing rejects invalid JSON, trailing data, missing required fields, invalid timestamps, invalid state mode, invalid reconciliation mode, invalid embedded manifests, duplicate owners, duplicate outputs, invalid output paths, unsupported output kinds, missing generated transforms, invalid URLs, invalid digests, negative sizes, and shared-root outputs whose owner is not registered.

View File

@@ -69,6 +69,7 @@ func processDestinationSelection(ctx context.Context, request runDestinationRequ
Publish: *request.destination.Publish, Publish: *request.destination.Publish,
Transform: request.destination.Transform, Transform: request.destination.Transform,
Links: request.destination.Links, Links: request.destination.Links,
State: request.destination.State,
Reconciliation: request.destination.Reconciliation, Reconciliation: request.destination.Reconciliation,
Transformers: request.transforms, Transformers: request.transforms,
Transfer: request.destination.Transfer, Transfer: request.destination.Transfer,

View File

@@ -15,6 +15,7 @@ import (
"gitea.maximumdirect.net/eric/distributor/internal/bundle" "gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/notify" "gitea.maximumdirect.net/eric/distributor/internal/notify"
"gitea.maximumdirect.net/eric/distributor/internal/publish"
"gitea.maximumdirect.net/eric/distributor/internal/state" "gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage" "gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake" "gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
@@ -221,6 +222,35 @@ func TestRunPublishesNewLocalBundle(t *testing.T) {
} }
} }
func TestRunThreadsSharedRootStateModeIntoPublish(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, sourceRoot, "", testBundleOptions{
Files: []testFile{{Path: "report.md", Data: "# Report\n"}},
})
configPath := writeSharedRootLocalConfig(t, sourceRoot, destinationRoot)
cfg, err := config.LoadFile(configPath)
if err != nil {
t.Fatalf("load config: %v", err)
}
report, err := buildRunReportWithBackendFactory(context.Background(), cfg, RunOptions{DryRun: true}, newBackendFactoryWithEnvironment)
if err != nil {
t.Fatalf("dry-run buildRunReportWithBackendFactory() error = %v", err)
}
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, storage.StateFileName)); !os.IsNotExist(statErr) {
t.Fatalf("state file stat error = %v, want absent", statErr)
}
}
func TestRunPipelineWithLocalSourcePublishesConfiguredDestination(t *testing.T) { func TestRunPipelineWithLocalSourcePublishesConfiguredDestination(t *testing.T) {
sourceRoot := t.TempDir() sourceRoot := t.TempDir()
destinationRoot := t.TempDir() destinationRoot := t.TempDir()
@@ -1717,6 +1747,23 @@ func writeLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string {
return testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot) return testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
} }
func writeSharedRootLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string {
t.Helper()
return writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
state:
mode: shared_root
`)
}
func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string { func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string {
t.Helper() t.Helper()
return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination) return testutil.WriteFanoutLocalConfig(t, sourceRoot, firstDestination, secondDestination)

View File

@@ -17,6 +17,9 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
case ActionSkipSame, ActionSkipDestinationNewer: case ActionSkipSame, ActionSkipDestinationNewer:
return nil return nil
case ActionPublishNew, ActionReplaceOlder, ActionForceReplace: 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")
}
default: default:
return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason) return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason)
} }

View File

@@ -34,6 +34,7 @@ type Request struct {
Publish config.PublishPolicy Publish config.PublishPolicy
Transform config.Transform Transform config.Transform
Links *config.Links Links *config.Links
State config.StatePolicy
Reconciliation config.ReconciliationPolicy Reconciliation config.ReconciliationPolicy
Transformers TransformerResolver Transformers TransformerResolver
Transfer config.TransferPolicy Transfer config.TransferPolicy
@@ -56,9 +57,16 @@ type Plan struct {
Reason string Reason string
Force bool Force bool
PrimaryURL string PrimaryURL string
StateMode string
OwnerScope state.OwnerScope
Reconciliation config.ReconciliationPolicy Reconciliation config.ReconciliationPolicy
Outputs []Output Outputs []Output
ExistingState *state.DistributorState ExistingState *state.DistributorState
ExistingSharedRoot *state.SharedRootState
OtherOwnerOutputs []state.SharedRootOutputFile
RetainedOwnerOutputs []state.SharedRootOutputFile
OwnerOutputsToDelete []state.SharedRootOutputFile
OwnerOutputsToWrite []Output
} }
type Output struct { type Output struct {
@@ -91,6 +99,7 @@ func Build(ctx context.Context, req Request) (Plan, error) {
comparison := compareDestination(req, status) comparison := compareDestination(req, status)
action, reason := actionForComparison(comparison, req.Transfer, req.Force) action, reason := actionForComparison(comparison, req.Transfer, req.Force)
reconciliation := normalizeReconciliation(req.Reconciliation) reconciliation := normalizeReconciliation(req.Reconciliation)
stateMode := normalizeState(req.State).Mode
plan := Plan{ plan := Plan{
PipelineID: req.PipelineID, PipelineID: req.PipelineID,
DestinationID: req.DestinationID, DestinationID: req.DestinationID,
@@ -102,9 +111,24 @@ func Build(ctx context.Context, req Request) (Plan, error) {
Reason: reason, Reason: reason,
Force: action == ActionForceReplace, Force: action == ActionForceReplace,
PrimaryURL: primaryURL, PrimaryURL: primaryURL,
StateMode: stateMode,
OwnerScope: state.CurrentOwnerScope(req.PipelineID, req.DestinationID),
Reconciliation: reconciliation, Reconciliation: reconciliation,
Outputs: outputs, Outputs: outputs,
ExistingState: status.State, ExistingState: status.State,
ExistingSharedRoot: status.SharedRoot,
}
if stateMode == config.StateModeSharedRoot {
sharedDetails, err := planSharedRootOwner(ctx, req, status, action, reconciliation, outputs)
plan.OtherOwnerOutputs = sharedDetails.OtherOwnerOutputs
plan.RetainedOwnerOutputs = sharedDetails.RetainedOwnerOutputs
plan.OwnerOutputsToDelete = sharedDetails.OwnerOutputsToDelete
plan.OwnerOutputsToWrite = sharedDetails.OwnerOutputsToWrite
if err != nil {
plan.Action = sharedDetails.Action
plan.Reason = sharedDetails.Reason
return plan, err
}
} }
if action == ActionFailConflict || action == ActionFailUnmanaged { if action == ActionFailConflict || action == ActionFailUnmanaged {
return plan, fmt.Errorf("%s: %s", action, reason) return plan, fmt.Errorf("%s: %s", action, reason)
@@ -133,6 +157,11 @@ func validateRequest(req Request) error {
default: default:
return fmt.Errorf("reconciliation.mode must be %s or %s", config.ReconciliationModeReplace, config.ReconciliationModeMerge) return fmt.Errorf("reconciliation.mode must be %s or %s", config.ReconciliationModeReplace, config.ReconciliationModeMerge)
} }
switch normalizeState(req.State).Mode {
case config.StateModeSingleOwner, config.StateModeSharedRoot:
default:
return fmt.Errorf("state.mode must be %s or %s", config.StateModeSingleOwner, config.StateModeSharedRoot)
}
return nil return nil
} }
@@ -143,7 +172,32 @@ func normalizeReconciliation(policy config.ReconciliationPolicy) config.Reconcil
return policy return policy
} }
func normalizeState(policy config.StatePolicy) config.StatePolicy {
if policy.Mode == "" {
policy.Mode = config.StateModeSingleOwner
}
return policy
}
func compareDestination(req Request, status state.DestinationStatus) state.Comparison { func compareDestination(req Request, status state.DestinationStatus) state.Comparison {
if normalizeState(req.State).Mode == config.StateModeSharedRoot {
scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID)
comparison := state.CompareSharedRootOwner(req.SourceBundle.Manifest, scope, status)
if req.PathMapping != config.PathMappingFixed || comparison.Outcome != state.OutcomeDifferentSourceConflict {
return comparison
}
destinationManifest, ok := sharedRootComparisonManifest(status, scope)
if !ok {
return comparison
}
if destinationManifest.Created.Before(req.SourceBundle.Manifest.Created) {
return state.Comparison{Outcome: state.OutcomeDestinationOlder, Reason: "fixed destination source is older than selected source"}
}
if destinationManifest.Created.After(req.SourceBundle.Manifest.Created) {
return state.Comparison{Outcome: state.OutcomeDestinationNewer, Reason: "fixed destination source is newer than selected source"}
}
return comparison
}
comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status) comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status)
if req.PathMapping != config.PathMappingFixed || comparison.Outcome != state.OutcomeDifferentSourceConflict || status.State == nil { if req.PathMapping != config.PathMappingFixed || comparison.Outcome != state.OutcomeDifferentSourceConflict || status.State == nil {
return comparison return comparison
@@ -158,6 +212,169 @@ func compareDestination(req Request, status state.DestinationStatus) state.Compa
return comparison return comparison
} }
func sharedRootComparisonManifest(status state.DestinationStatus, scope state.OwnerScope) (bundle.Manifest, bool) {
if status.SharedRoot != nil {
return status.SharedRoot.SourceManifest(scope)
}
if status.State != nil && status.State.PipelineID == scope.PipelineID && status.State.DestinationID == scope.DestinationID {
return status.State.Source.Manifest, true
}
return bundle.Manifest{}, false
}
type sharedRootPlanDetails struct {
Action Action
Reason string
OtherOwnerOutputs []state.SharedRootOutputFile
RetainedOwnerOutputs []state.SharedRootOutputFile
OwnerOutputsToDelete []state.SharedRootOutputFile
OwnerOutputsToWrite []Output
}
func planSharedRootOwner(ctx context.Context, req Request, status state.DestinationStatus, action Action, reconciliation config.ReconciliationPolicy, outputs []Output) (sharedRootPlanDetails, error) {
scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID)
details := sharedRootPlanDetails{Action: action}
if !isWriteAction(action) {
details.OtherOwnerOutputs = otherOwnerOutputs(status, scope)
return details, nil
}
plannedPaths := outputPaths(outputs)
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
details.Reason = reason
return details, fmt.Errorf("%s: %s", ActionFailConflict, reason)
}
if err := rejectSharedRootUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, status, scope, plannedPaths); err != nil {
details.Action = ActionFailUnmanaged
details.Reason = err.Error()
return details, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
}
details.OtherOwnerOutputs = otherOwnerOutputs(status, scope)
ownerOutputs := currentOwnerOutputs(status, scope)
planned := make(map[string]struct{}, len(plannedPaths))
for _, path := range plannedPaths {
planned[path] = struct{}{}
}
for _, output := range ownerOutputs {
if _, exists := planned[output.Path]; exists {
continue
}
if action == ActionReplaceOlder && reconciliation.Mode == config.ReconciliationModeReplace {
details.OwnerOutputsToDelete = append(details.OwnerOutputsToDelete, output)
continue
}
if action == ActionReplaceOlder && reconciliation.Mode == config.ReconciliationModeMerge {
details.RetainedOwnerOutputs = append(details.RetainedOwnerOutputs, output)
}
}
details.OwnerOutputsToWrite = append([]Output(nil), outputs...)
return details, nil
}
func isWriteAction(action Action) bool {
switch action {
case ActionPublishNew, ActionReplaceOlder, ActionForceReplace:
return true
default:
return false
}
}
func outputPaths(outputs []Output) []string {
paths := make([]string, 0, len(outputs))
for _, output := range outputs {
paths = append(paths, output.DestinationPath)
}
return paths
}
func sharedRootPathOwnershipConflict(status state.DestinationStatus, scope state.OwnerScope, paths []string) (state.PathOwnershipConflict, bool) {
if status.SharedRoot != nil {
return status.SharedRoot.PathOwnershipConflict(scope, paths)
}
return state.PathOwnershipConflict{}, false
}
func rejectSharedRootUnmanagedCollisions(ctx context.Context, backend storage.Backend, bundlePath string, status state.DestinationStatus, scope state.OwnerScope, paths []string) error {
for _, path := range paths {
if pathManagedBySharedRootStatus(status, scope, path) {
continue
}
destinationPath, err := storage.Join(bundlePath, path)
if err != nil {
return err
}
if _, err := backend.Stat(ctx, destinationPath); err == nil {
return fmt.Errorf("destination output path %s exists but is not managed by destination state", storage.DisplayPath(path))
} else if !storage.IsNotFound(err) {
return err
}
}
return nil
}
func pathManagedBySharedRootStatus(status state.DestinationStatus, scope state.OwnerScope, path string) bool {
if status.SharedRoot != nil {
_, exists := status.SharedRoot.OutputOwner(path)
return exists
}
if status.State != nil && status.State.PipelineID == scope.PipelineID && status.State.DestinationID == scope.DestinationID {
_, exists := state.FindOutputByPath(status.State.Outputs, path)
return exists
}
return false
}
func otherOwnerOutputs(status state.DestinationStatus, scope state.OwnerScope) []state.SharedRootOutputFile {
if status.SharedRoot == nil {
return nil
}
outputs := make([]state.SharedRootOutputFile, 0, len(status.SharedRoot.Outputs))
for _, output := range status.SharedRoot.Outputs {
if output.Owner != scope {
outputs = append(outputs, output)
}
}
return outputs
}
func currentOwnerOutputs(status state.DestinationStatus, scope state.OwnerScope) []state.SharedRootOutputFile {
if status.SharedRoot != nil {
outputs := make([]state.SharedRootOutputFile, 0, len(status.SharedRoot.Outputs))
for _, output := range status.SharedRoot.Outputs {
if output.Owner == scope {
outputs = append(outputs, output)
}
}
return outputs
}
if status.State != nil && status.State.PipelineID == scope.PipelineID && status.State.DestinationID == scope.DestinationID {
outputs := make([]state.SharedRootOutputFile, 0, len(status.State.Outputs))
for _, output := range status.State.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: scope,
SourceID: status.State.Source.Manifest.ID,
SourceDigest: status.State.Source.Manifest.Digest,
SourceCreated: status.State.Source.Manifest.Created,
CreatedAt: output.CreatedAt,
UpdatedAt: output.UpdatedAt,
})
}
return outputs
}
return nil
}
func actionForComparison(comparison state.Comparison, transfer config.TransferPolicy, force bool) (Action, string) { func actionForComparison(comparison state.Comparison, transfer config.TransferPolicy, force bool) (Action, string) {
switch comparison.Outcome { switch comparison.Outcome {
case state.OutcomeDestinationAbsent: case state.OutcomeDestinationAbsent:

View File

@@ -14,11 +14,11 @@ func inspectDestination(ctx context.Context, backend storage.Backend, bundlePath
} }
data, err := backend.ReadFile(ctx, statePath) data, err := backend.ReadFile(ctx, statePath)
if err == nil { if err == nil {
destinationState, parseErr := state.Parse(data) document, parseErr := state.ParseDocument(data)
if parseErr != nil { if parseErr != nil {
return state.DestinationStatus{StateErr: parseErr}, nil return state.DestinationStatus{StateErr: parseErr}, nil
} }
return state.DestinationStatus{State: &destinationState, HasContents: true}, nil return state.DestinationStatus{State: document.SingleOwner, SharedRoot: document.SharedRoot, HasContents: true}, nil
} }
if !storage.IsNotFound(err) { if !storage.IsNotFound(err) {
return state.DestinationStatus{}, err return state.DestinationStatus{}, err

View File

@@ -0,0 +1,261 @@
package publish
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)
func TestBuildSharedRootTreatsAbsentOwnerAsPublishable(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 plan.Action != ActionPublishNew {
t.Fatalf("plan action = %s, want publish_new", plan.Action)
}
if plan.OwnerScope != state.CurrentOwnerScope("reports", "archive") {
t.Fatalf("owner scope = %#v, want reports/archive", plan.OwnerScope)
}
if got, want := len(plan.OtherOwnerOutputs), 1; got != want {
t.Fatalf("other owner output count = %d, want %d", got, want)
}
if got, want := len(plan.OwnerOutputsToWrite), 1; got != want {
t.Fatalf("owner output write count = %d, want %d", got, want)
}
if len(plan.OwnerOutputsToDelete) != 0 || len(plan.RetainedOwnerOutputs) != 0 {
t.Fatalf("delete=%#v retained=%#v, want none", plan.OwnerOutputsToDelete, plan.RetainedOwnerOutputs)
}
}
func TestBuildSharedRootReplaceDeletesOnlyCurrentOwnerOmittedOutputs(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.ReconciliationModeReplace)
plan, err := Build(context.Background(), req)
if err != nil {
t.Fatalf("Build() error = %v", err)
}
if plan.Action != ActionReplaceOlder {
t.Fatalf("plan action = %s, want replace_older", plan.Action)
}
if got, want := sharedRootOutputPathList(plan.OwnerOutputsToDelete), "old.md"; got != want {
t.Fatalf("owner outputs to delete = %q, want %q", got, want)
}
if got, want := sharedRootOutputPathList(plan.OtherOwnerOutputs), "other/report.md"; got != want {
t.Fatalf("other owner outputs = %q, want %q", got, want)
}
if len(plan.RetainedOwnerOutputs) != 0 {
t.Fatalf("retained owner outputs = %#v, want none", plan.RetainedOwnerOutputs)
}
}
func TestBuildSharedRootMergeRetainsCurrentOwnerOmittedOutputs(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 got, want := sharedRootOutputPathList(plan.RetainedOwnerOutputs), "old.md"; got != want {
t.Fatalf("retained owner outputs = %q, want %q", got, want)
}
if len(plan.OwnerOutputsToDelete) != 0 {
t.Fatalf("owner outputs to delete = %#v, want none", plan.OwnerOutputsToDelete)
}
}
func TestBuildSharedRootRejectsOtherOwnerPathConflict(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()
sharedRoot := sharedRootStateWithOwners(t, sourceBundle.Manifest, false)
sharedRoot.Outputs[0].Path = "report.md"
writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRoot)
req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace)
plan, err := Build(context.Background(), req)
if err == nil || !strings.Contains(err.Error(), "fail_conflict") {
t.Fatalf("Build() error = %v, want fail_conflict", err)
}
if plan.Action != ActionFailConflict {
t.Fatalf("plan action = %s, want fail_conflict", plan.Action)
}
}
func TestBuildSharedRootRejectsUnmanagedPathCollision(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))
testutil.WriteFakeFile(t, destinationBackend, "bundle/report.md", "unmanaged")
req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace)
plan, err := Build(context.Background(), req)
if err == nil || !strings.Contains(err.Error(), "fail_unmanaged") {
t.Fatalf("Build() error = %v, want fail_unmanaged", err)
}
if plan.Action != ActionFailUnmanaged {
t.Fatalf("plan action = %s, want fail_unmanaged", plan.Action)
}
}
func TestExecuteSharedRootWriteActionsAreDisabled(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()
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)
}
}
func sharedRootRequest(sourceBackend, destinationBackend *fake.Backend, sourceBundle bundle.Bundle, reconciliationMode string) Request {
return Request{
PipelineID: "reports",
DestinationID: "archive",
SourceBundle: sourceBundle,
SourceBackend: sourceBackend,
DestinationBackend: destinationBackend,
DestinationBundlePath: sourceBundle.RootRelativePath,
Publish: config.PublishPolicy{Source: true},
State: config.StatePolicy{Mode: config.StateModeSharedRoot},
Reconciliation: config.ReconciliationPolicy{Mode: reconciliationMode},
Transfer: defaultTransfer(),
DistributorVersion: "test",
}
}
func writeFakeSharedRootState(t *testing.T, backend *fake.Backend, relative string, sharedRoot state.SharedRootState) {
t.Helper()
data, err := json.MarshalIndent(sharedRoot, "", " ")
if err != nil {
t.Fatalf("marshal shared-root state: %v", err)
}
statePath, err := storage.StatePath(relative)
if err != nil {
t.Fatalf("state path: %v", err)
}
testutil.WriteFakeFile(t, backend, statePath, string(append(data, '\n')))
for _, output := range sharedRoot.Outputs {
path, err := storage.Join(relative, output.Path)
if err != nil {
t.Fatalf("join output path: %v", err)
}
testutil.WriteFakeFile(t, backend, path, "old")
}
}
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)
otherManifest := testutil.ValidManifest(testutil.BundleOptions{
ID: "other.source",
Files: []testutil.SourceFile{{Path: "other/report.md", Data: "# Other\n"}},
})
sharedRoot := state.SharedRootState{
SchemaVersion: state.SharedRootSchemaVersion,
DistributorVersion: "test",
CreatedAt: createdAt,
UpdatedAt: createdAt,
State: state.StatePolicy{Mode: state.StateModeSharedRoot},
Owners: []state.OwnerRecord{{
Scope: state.CurrentOwnerScope("other", "archive"),
Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace},
Source: state.SourceState{Manifest: otherManifest},
}},
Outputs: []state.SharedRootOutputFile{{
Path: "other/report.md",
Kind: state.OutputKindSource,
SourcePath: "other/report.md",
SHA256: otherManifest.Files[0].SHA256,
Size: otherManifest.Files[0].Size,
Owner: state.CurrentOwnerScope("other", "archive"),
SourceID: otherManifest.ID,
SourceDigest: otherManifest.Digest,
SourceCreated: otherManifest.Created,
CreatedAt: createdAt,
UpdatedAt: createdAt,
}},
}
if !includeCurrent {
return sharedRoot
}
older := current
older.Created = older.Created.Add(-time.Hour)
older.Files = append([]bundle.ManifestFile(nil), current.Files...)
older.Files = append(older.Files, bundle.ManifestFile{
Path: "old.md",
SHA256: bundle.FileDigest([]byte("old\n")),
Size: int64(len("old\n")),
})
older.Digest = bundle.BundleDigest(older.Files)
scope := state.CurrentOwnerScope("reports", "archive")
sharedRoot.Owners = append(sharedRoot.Owners, state.OwnerRecord{
Scope: scope,
Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace},
Source: state.SourceState{Manifest: older},
})
for _, file := range older.Files {
sharedRoot.Outputs = append(sharedRoot.Outputs, state.SharedRootOutputFile{
Path: file.Path,
Kind: state.OutputKindSource,
SourcePath: file.Path,
SHA256: file.SHA256,
Size: file.Size,
Owner: scope,
SourceID: older.ID,
SourceDigest: older.Digest,
SourceCreated: older.Created,
CreatedAt: createdAt,
UpdatedAt: createdAt,
})
}
return sharedRoot
}
func sharedRootOutputPathList(outputs []state.SharedRootOutputFile) string {
paths := make([]string, 0, len(outputs))
for _, output := range outputs {
paths = append(paths, output.Path)
}
return strings.Join(paths, ",")
}

View File

@@ -22,6 +22,7 @@ const (
type DestinationStatus struct { type DestinationStatus struct {
State *DistributorState State *DistributorState
SharedRoot *SharedRootState
StateErr error StateErr error
HasContents bool HasContents bool
} }
@@ -31,6 +32,58 @@ type Comparison struct {
Reason string Reason string
} }
func CompareSharedRootOwner(source bundle.Manifest, scope OwnerScope, status DestinationStatus) Comparison {
if status.StateErr != nil {
return Comparison{Outcome: OutcomeInvalidState, Reason: status.StateErr.Error()}
}
if status.SharedRoot != nil {
if err := ValidateSharedRoot(*status.SharedRoot); err != nil {
return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error()}
}
owner, ok := status.SharedRoot.Owner(scope)
if !ok {
return Comparison{Outcome: OutcomeDestinationAbsent, Reason: fmt.Sprintf("destination owner %s/%s is absent", scope.PipelineID, scope.DestinationID)}
}
return compareManifests(source, owner.Source.Manifest)
}
if status.State != nil {
destinationState := *status.State
if err := Validate(destinationState); err != nil {
return Comparison{Outcome: OutcomeInvalidState, Reason: err.Error()}
}
if destinationState.PipelineID != scope.PipelineID {
return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("pipeline id %q does not match %q", destinationState.PipelineID, scope.PipelineID)}
}
if destinationState.DestinationID != scope.DestinationID {
return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("destination id %q does not match %q", destinationState.DestinationID, scope.DestinationID)}
}
return compareManifests(source, destinationState.Source.Manifest)
}
if status.HasContents {
return Comparison{Outcome: OutcomeDestinationUnmanaged, Reason: "destination has content but no distributor state"}
}
return Comparison{Outcome: OutcomeDestinationAbsent, Reason: "destination state is absent"}
}
func compareManifests(source, destination bundle.Manifest) Comparison {
if manifestsEqual(source, destination) {
return Comparison{Outcome: OutcomeSameSource, Reason: "destination source manifest matches source"}
}
if destination.ID != source.ID {
return Comparison{Outcome: OutcomeDifferentSourceConflict, Reason: "destination source id differs from source"}
}
if destination.Created.Before(source.Created) {
return Comparison{Outcome: OutcomeDestinationOlder, Reason: "destination source is older than source"}
}
if destination.Created.After(source.Created) {
return Comparison{Outcome: OutcomeDestinationNewer, Reason: "destination source is newer than source"}
}
if destination.Digest != source.Digest {
return Comparison{Outcome: OutcomeSameCreatedConflict, Reason: "destination source has same id and created time but different digest"}
}
return Comparison{Outcome: OutcomeInvalidState, Reason: "destination source differs from source without a supported comparison outcome"}
}
func Compare(source bundle.Manifest, pipelineID, destinationID string, status DestinationStatus) Comparison { func Compare(source bundle.Manifest, pipelineID, destinationID string, status DestinationStatus) Comparison {
if status.StateErr != nil { if status.StateErr != nil {
return Comparison{Outcome: OutcomeInvalidState, Reason: status.StateErr.Error()} return Comparison{Outcome: OutcomeInvalidState, Reason: status.StateErr.Error()}
@@ -53,23 +106,7 @@ func Compare(source bundle.Manifest, pipelineID, destinationID string, status De
return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("destination id %q does not match %q", destinationState.DestinationID, destinationID)} return Comparison{Outcome: OutcomeIdentityMismatch, Reason: fmt.Sprintf("destination id %q does not match %q", destinationState.DestinationID, destinationID)}
} }
destinationManifest := destinationState.Source.Manifest return compareManifests(source, destinationState.Source.Manifest)
if manifestsEqual(source, destinationManifest) {
return Comparison{Outcome: OutcomeSameSource, Reason: "destination source manifest matches source"}
}
if destinationManifest.ID != source.ID {
return Comparison{Outcome: OutcomeDifferentSourceConflict, Reason: "destination source id differs from source"}
}
if destinationManifest.Created.Before(source.Created) {
return Comparison{Outcome: OutcomeDestinationOlder, Reason: "destination source is older than source"}
}
if destinationManifest.Created.After(source.Created) {
return Comparison{Outcome: OutcomeDestinationNewer, Reason: "destination source is newer than source"}
}
if destinationManifest.Digest != source.Digest {
return Comparison{Outcome: OutcomeSameCreatedConflict, Reason: "destination source has same id and created time but different digest"}
}
return Comparison{Outcome: OutcomeInvalidState, Reason: "destination source differs from source without a supported comparison outcome"}
} }
func manifestsEqual(a, b bundle.Manifest) bool { func manifestsEqual(a, b bundle.Manifest) bool {

View File

@@ -200,6 +200,38 @@ func TestSharedRootOwnerOutputHelpersRejectConflicts(t *testing.T) {
} }
} }
func TestCompareSharedRootOwnerScopesCurrentOwner(t *testing.T) {
state := validSharedRootState(t)
source := validManifest(t)
source.Created = source.Created.Add(time.Hour)
scope := CurrentOwnerScope("reports", "archive")
comparison := CompareSharedRootOwner(source, scope, DestinationStatus{SharedRoot: &state, HasContents: true})
if comparison.Outcome != OutcomeDestinationOlder {
t.Fatalf("comparison = %#v, want destination older for current owner", comparison)
}
missing := CompareSharedRootOwner(source, CurrentOwnerScope("missing", "archive"), DestinationStatus{SharedRoot: &state, HasContents: true})
if missing.Outcome != OutcomeDestinationAbsent {
t.Fatalf("missing owner comparison = %#v, want destination absent", missing)
}
}
func TestCompareSharedRootOwnerAcceptsMatchingSingleOwnerState(t *testing.T) {
singleOwner, err := Parse([]byte(validStateJSON(t)))
if err != nil {
t.Fatalf("Parse() error = %v", err)
}
source := singleOwner.Source.Manifest
source.Created = source.Created.Add(time.Hour)
scope := CurrentOwnerScope(singleOwner.PipelineID, singleOwner.DestinationID)
comparison := CompareSharedRootOwner(source, scope, DestinationStatus{State: &singleOwner, HasContents: true})
if comparison.Outcome != OutcomeDestinationOlder {
t.Fatalf("comparison = %#v, want destination older", comparison)
}
}
func validSharedRootStateJSON(t *testing.T) string { func validSharedRootStateJSON(t *testing.T) string {
t.Helper() t.Helper()
data, err := json.MarshalIndent(validSharedRootState(t), "", " ") data, err := json.MarshalIndent(validSharedRootState(t), "", " ")