Add catalog workflow planning

2026-06-19 15:36:26 +00:00
parent 20129bfff5
commit 77cde40296
10 changed files with 488 additions and 1936 deletions
--- a/internal/publish/plan.go
+++ b/internal/publish/plan.go
@@ -3,6 +3,7 @@ package publish
 import (
 	"context"
 	"fmt"
+	"time"

 	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
@@ -24,6 +25,8 @@ const (
 	ActionFailUnmanaged        Action = "fail_unmanaged"
 	ActionForceReplace         Action = "force_replace"
 	ActionReplaceTakeover      Action = "replace_takeover"
+	ActionUpsertAdditive       Action = "upsert_additive"
+	ActionReplaceCatalog       Action = "replace_catalog"
 )

 type Request struct {
@@ -37,44 +40,17 @@ type Request struct {
 	Publish               config.PublishPolicy
 	Transform             config.Transform
 	Links                 *config.Links
-	State                 config.StatePolicy
-	Reconciliation        config.ReconciliationPolicy
-	Takeover              config.TakeoverPolicy
+	Workflow              string
 	Transformers          TransformerResolver
-	Transfer              config.TransferPolicy
 	DistributorVersion    string
 	Force                 bool
+	Now                   time.Time
 }

 type TransformerResolver interface {
 	Get(name string) (transform.Transformer, bool)
 }

-type Plan struct {
-	PipelineID            string
-	DestinationID         string
-	BundleID              string
-	BundlePath            string
-	DestinationBundlePath string
-	PathMapping           string
-	Action                Action
-	Reason                string
-	Force                 bool
-	PrimaryURL            string
-	StateMode             string
-	OwnerScope            state.OwnerScope
-	Reconciliation        config.ReconciliationPolicy
-	TakeoverMode          string
-	Outputs               []Output
-	ExistingState         *state.DistributorState
-	ExistingSharedRoot    *state.SharedRootState
-	OtherOwnerOutputs     []state.SharedRootOutputFile
-	TakenOverOwnerOutputs []state.SharedRootOutputFile
-	RetainedOwnerOutputs  []state.SharedRootOutputFile
-	OwnerOutputsToDelete  []state.SharedRootOutputFile
-	OwnerOutputsToWrite   []Output
-}
-
 type Output struct {
 	SourcePath      string
 	DestinationPath string
@@ -86,6 +62,49 @@ type Output struct {
 	Size            int64
 }

+type Plan struct {
+	PipelineID             string
+	DestinationID          string
+	BundleID               string
+	BundlePath             string
+	DestinationBundlePath  string
+	PathMapping            string
+	Action                 Action
+	Reason                 string
+	Force                  bool
+	PrimaryURL             string
+	Workflow               string
+	OwnerScope             state.OwnerScope
+	Outputs                []Output
+	ExistingCatalog        *state.CatalogState
+	SupersededLegacy       *state.SupersededLegacyState
+	CatalogOutputsToWrite  []state.CatalogOutputFile
+	CatalogOutputsToRetain []state.CatalogOutputFile
+	CatalogOutputsToDelete []state.CatalogOutputFile
+	ClearDestinationRoot   bool
+
+	// Retained while the executor is migrated to catalog state.
+	StateMode             string
+	Reconciliation        config.ReconciliationPolicy
+	TakeoverMode          string
+	ExistingState         *state.DistributorState
+	ExistingSharedRoot    *state.SharedRootState
+	OtherOwnerOutputs     []state.SharedRootOutputFile
+	TakenOverOwnerOutputs []state.SharedRootOutputFile
+	RetainedOwnerOutputs  []state.SharedRootOutputFile
+	OwnerOutputsToDelete  []state.SharedRootOutputFile
+	OwnerOutputsToWrite   []Output
+}
+
+type catalogPlanDetails struct {
+	Action                 Action
+	Reason                 string
+	CatalogOutputsToWrite  []state.CatalogOutputFile
+	CatalogOutputsToRetain []state.CatalogOutputFile
+	CatalogOutputsToDelete []state.CatalogOutputFile
+	ClearDestinationRoot   bool
+}
+
 func Build(ctx context.Context, req Request) (Plan, error) {
 	if err := validateRequest(req); err != nil {
 		return Plan{}, err
@@ -102,14 +121,9 @@ func Build(ctx context.Context, req Request) (Plan, error) {
 	if err != nil {
 		return Plan{}, err
 	}
-	reconciliation := normalizeReconciliation(req.Reconciliation)
-	stateMode := normalizeState(req.State).Mode
-	comparison := compareDestination(req, status)
-	action, reason := actionForComparison(comparison, req.Transfer, req.Force)
-	if takeoverActionAllowed(req, status, comparison, stateMode, action) {
-		action = ActionReplaceTakeover
-		reason = comparison.Reason
-	}
+	workflow := normalizeWorkflow(req.Workflow)
+	scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID)
+	now := requestTime(req)
 	plan := Plan{
 		PipelineID:            req.PipelineID,
 		DestinationID:         req.DestinationID,
@@ -117,36 +131,37 @@ func Build(ctx context.Context, req Request) (Plan, error) {
 		BundlePath:            req.SourceBundle.RootRelativePath,
 		DestinationBundlePath: req.DestinationBundlePath,
 		PathMapping:           req.PathMapping,
-		Action:                action,
-		Reason:                reason,
-		Force:                 action == ActionForceReplace,
 		PrimaryURL:            primaryURL,
-		StateMode:             stateMode,
-		OwnerScope:            state.CurrentOwnerScope(req.PipelineID, req.DestinationID),
-		Reconciliation:        reconciliation,
-		TakeoverMode:          normalizeTakeover(req.Takeover).Mode,
+		Workflow:              workflow,
+		OwnerScope:            scope,
 		Outputs:               outputs,
-		ExistingState:         status.State,
-		ExistingSharedRoot:    status.SharedRoot,
+		ExistingCatalog:       status.Catalog,
+		SupersededLegacy:      status.SupersededLegacy,
 	}
-	if stateMode == config.StateModeSharedRoot {
-		sharedDetails, err := planSharedRootOwner(ctx, req, status, action, reconciliation, outputs)
-		plan.Action = sharedDetails.Action
-		if sharedDetails.Reason != "" {
-			plan.Reason = sharedDetails.Reason
-		}
-		plan.OtherOwnerOutputs = sharedDetails.OtherOwnerOutputs
-		plan.TakenOverOwnerOutputs = sharedDetails.TakenOverOwnerOutputs
-		plan.RetainedOwnerOutputs = sharedDetails.RetainedOwnerOutputs
-		plan.OwnerOutputsToDelete = sharedDetails.OwnerOutputsToDelete
-		plan.OwnerOutputsToWrite = sharedDetails.OwnerOutputsToWrite
-		if err != nil {
-			return plan, err
-		}
-	}
-	if plan.Action == ActionFailConflict || plan.Action == ActionFailUnmanaged {
+	if status.StateErr != nil {
+		plan.Action = ActionFailConflict
+		plan.Reason = status.StateErr.Error()
 		return plan, fmt.Errorf("%s: %s", plan.Action, plan.Reason)
 	}
+
+	var details catalogPlanDetails
+	switch {
+	case status.Catalog != nil:
+		details, err = planExistingCatalog(ctx, req, *status.Catalog, outputs, workflow, scope, now)
+	case status.SupersededLegacy != nil:
+		details = planSupersededLegacy(req, outputs, workflow, scope, now)
+	default:
+		details, err = planWithoutCatalog(ctx, req, outputs, workflow, scope, now)
+	}
+	plan.Action = details.Action
+	plan.Reason = details.Reason
+	plan.CatalogOutputsToWrite = details.CatalogOutputsToWrite
+	plan.CatalogOutputsToRetain = details.CatalogOutputsToRetain
+	plan.CatalogOutputsToDelete = details.CatalogOutputsToDelete
+	plan.ClearDestinationRoot = details.ClearDestinationRoot
+	if err != nil {
+		return plan, err
+	}
 	return plan, nil
 }

@@ -166,165 +181,134 @@ func validateRequest(req Request) error {
 	if err := config.ValidatePublishTransformPolicy(req.Publish, req.Transform); err != nil {
 		return fmt.Errorf("publish/transform policy: %w", err)
 	}
-	switch normalizeReconciliation(req.Reconciliation).Mode {
-	case config.ReconciliationModeReplace, config.ReconciliationModeMerge:
+	switch normalizeWorkflow(req.Workflow) {
+	case config.WorkflowAdditive, config.WorkflowReplacement:
 	default:
-		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)
-	}
-	switch normalizeTakeover(req.Takeover).Mode {
-	case config.TakeoverModeSamePipeline, config.TakeoverModeSameSource, config.TakeoverModeAnyManaged, config.TakeoverModeNever:
-	default:
-		return fmt.Errorf("takeover.mode must be %s, %s, %s, or %s", config.TakeoverModeSamePipeline, config.TakeoverModeSameSource, config.TakeoverModeAnyManaged, config.TakeoverModeNever)
+		return fmt.Errorf("destination.workflow must be %s or %s", config.WorkflowAdditive, config.WorkflowReplacement)
 	}
 	return nil
 }

-func normalizeReconciliation(policy config.ReconciliationPolicy) config.ReconciliationPolicy {
-	if policy.Mode == "" {
-		policy.Mode = config.ReconciliationModeReplace
+func normalizeWorkflow(workflow string) string {
+	if workflow == "" {
+		return config.WorkflowAdditive
 	}
-	return policy
+	return workflow
 }

-func normalizeState(policy config.StatePolicy) config.StatePolicy {
-	if policy.Mode == "" {
-		policy.Mode = config.StateModeSingleOwner
+func requestTime(req Request) time.Time {
+	if req.Now.IsZero() {
+		return time.Now().UTC()
 	}
-	return policy
+	return req.Now.UTC()
 }

-func normalizeTakeover(policy config.TakeoverPolicy) config.TakeoverPolicy {
-	if policy.Mode == "" {
-		policy.Mode = config.TakeoverModeSamePipeline
+func planExistingCatalog(ctx context.Context, req Request, catalog state.CatalogState, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) (catalogPlanDetails, error) {
+	if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, catalog.Outputs, outputs); err != nil {
+		return catalogPlanDetails{
+			Action: ActionFailUnmanaged,
+			Reason: err.Error(),
+		}, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
 	}
-	return policy
-}
-
-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",
-				Detail: state.ComparisonDetail{
-					Kind:                state.ComparisonDetailDestinationNewer,
-					CurrentSourceID:     req.SourceBundle.Manifest.ID,
-					DestinationSourceID: destinationManifest.ID,
-				},
-			}
-		}
-		return comparison
+	planned := outputPathSet(outputs)
+	details := catalogPlanDetails{
+		Action:                actionForWorkflow(workflow),
+		CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, catalog.Outputs, scope, now),
 	}
-	comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status)
-	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
-	TakenOverOwnerOutputs []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 action == ActionForceReplace {
-		details.OwnerOutputsToWrite = append([]Output(nil), outputs...)
-		return details, nil
-	}
-	conflicts := sharedRootPathOwnershipConflicts(status, scope, plannedPaths)
-	if len(conflicts) > 0 {
-		conflictAction := ActionReplaceTakeover
-		for _, conflict := range conflicts {
-			if sharedRootTakeoverAllowed(req, status, conflict) {
-				continue
-			}
-			if req.Transfer.OnConflict == config.TransferActionReplace {
-				conflictAction = ActionReplaceConflict
-				continue
-			}
-			reason := sharedRootOwnershipConflictReason(conflict)
-			details.Action = ActionFailConflict
-			details.Reason = reason
-			return details, fmt.Errorf("%s: %s", ActionFailConflict, reason)
-		}
-		details.Action = conflictAction
-		details.Reason = sharedRootOwnershipConflictReason(conflicts[0])
-		details.TakenOverOwnerOutputs = sharedRootConflictOutputs(status.SharedRoot, conflicts)
-	}
-	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)
-	}
-
-	takenOverPaths := sharedRootOutputPathSet(details.TakenOverOwnerOutputs)
-	details.OtherOwnerOutputs = otherOwnerOutputsExcept(status, scope, takenOverPaths)
-	ownerOutputs := currentOwnerOutputs(status, scope)
-	planned := make(map[string]struct{}, len(plannedPaths))
-	for _, path := range plannedPaths {
-		planned[path] = struct{}{}
-	}
-	for _, output := range ownerOutputs {
+	for _, output := range catalog.Outputs {
 		if _, exists := planned[output.Path]; exists {
 			continue
 		}
-		if details.Action == ActionReplaceTakeover || details.Action == ActionReplaceConflict || (isReconciliationReplacementAction(details.Action) && reconciliation.Mode == config.ReconciliationModeReplace) {
-			details.OwnerOutputsToDelete = append(details.OwnerOutputsToDelete, output)
+		if workflow == config.WorkflowReplacement && output.PipelineID == scope.PipelineID && output.DestinationID == scope.DestinationID {
+			details.CatalogOutputsToDelete = append(details.CatalogOutputsToDelete, output)
 			continue
 		}
-		if isReconciliationReplacementAction(details.Action) && reconciliation.Mode == config.ReconciliationModeMerge {
-			details.RetainedOwnerOutputs = append(details.RetainedOwnerOutputs, output)
-		}
+		details.CatalogOutputsToRetain = append(details.CatalogOutputsToRetain, output)
 	}
-	details.OwnerOutputsToWrite = append([]Output(nil), outputs...)
 	return details, nil
 }

-func isWriteAction(action Action) bool {
-	switch action {
-	case ActionPublishNew, ActionReplaceOlder, ActionReplaceConflict, ActionReplaceNewer, ActionReplaceTakeover, ActionForceReplace:
-		return true
-	default:
-		return false
+func planSupersededLegacy(req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) catalogPlanDetails {
+	details := catalogPlanDetails{
+		Action:                actionForWorkflow(workflow),
+		CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, nil, scope, now),
 	}
+	if workflow == config.WorkflowReplacement {
+		details.ClearDestinationRoot = true
+	}
+	return details
+}
+
+func planWithoutCatalog(ctx context.Context, req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) (catalogPlanDetails, error) {
+	if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, nil, outputs); err != nil {
+		return catalogPlanDetails{
+			Action: ActionFailUnmanaged,
+			Reason: err.Error(),
+		}, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
+	}
+	return catalogPlanDetails{
+		Action:                ActionPublishNew,
+		CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, nil, scope, now),
+	}, nil
+}
+
+func actionForWorkflow(workflow string) Action {
+	if workflow == config.WorkflowReplacement {
+		return ActionReplaceCatalog
+	}
+	return ActionUpsertAdditive
+}
+
+func catalogOutputsForPlan(req Request, outputs []Output, existing []state.CatalogOutputFile, scope state.OwnerScope, now time.Time) []state.CatalogOutputFile {
+	files := make([]state.CatalogOutputFile, 0, len(outputs))
+	source := state.CatalogSourceIdentity{
+		ID:      req.SourceBundle.Manifest.ID,
+		Digest:  req.SourceBundle.Manifest.Digest,
+		Created: req.SourceBundle.Manifest.Created,
+	}
+	for _, output := range outputs {
+		createdAt := now
+		if existingOutput, ok := state.FindCatalogOutputByPath(existing, output.DestinationPath); ok {
+			createdAt = existingOutput.CreatedAt
+		}
+		file := state.CatalogOutputFile{
+			Path:          output.DestinationPath,
+			PipelineID:    scope.PipelineID,
+			DestinationID: scope.DestinationID,
+			Source:        source,
+			Kind:          output.Kind,
+			URL:           output.URL,
+			SHA256:        output.SHA256,
+			Size:          output.Size,
+			CreatedAt:     createdAt,
+			UpdatedAt:     now,
+		}
+		if output.Kind == state.OutputKindGenerated {
+			file.SourcePath = output.SourcePath
+			file.Transform = output.Transform
+		}
+		files = append(files, file)
+	}
+	return files
+}
+
+func rejectCatalogUnmanagedCollisions(ctx context.Context, backend storage.Backend, bundlePath string, existing []state.CatalogOutputFile, outputs []Output) error {
+	managed := catalogOutputPathSet(existing)
+	for _, output := range outputs {
+		if _, exists := managed[output.DestinationPath]; exists {
+			continue
+		}
+		destinationPath, err := storage.Join(bundlePath, output.DestinationPath)
+		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 catalog state", storage.DisplayPath(output.DestinationPath))
+		} else if !storage.IsNotFound(err) {
+			return err
+		}
+	}
+	return nil
 }

 func outputPaths(outputs []Output) []string {
@@ -335,127 +319,31 @@ func outputPaths(outputs []Output) []string {
 	return paths
 }

-func sharedRootOwnershipConflictReason(conflict state.PathOwnershipConflict) string {
-	return fmt.Sprintf("destination output path %s is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID)
+func outputPathSet(outputs []Output) map[string]struct{} {
+	paths := make(map[string]struct{}, len(outputs))
+	for _, output := range outputs {
+		paths[output.DestinationPath] = struct{}{}
+	}
+	return paths
 }

-func sharedRootPathOwnershipConflicts(status state.DestinationStatus, scope state.OwnerScope, paths []string) []state.PathOwnershipConflict {
-	if status.SharedRoot == nil {
-		return nil
+func catalogOutputPathSet(outputs []state.CatalogOutputFile) map[string]struct{} {
+	paths := make(map[string]struct{}, len(outputs))
+	for _, output := range outputs {
+		paths[output.Path] = struct{}{}
 	}
-	conflicts := make([]state.PathOwnershipConflict, 0)
-	seen := make(map[string]struct{}, len(paths))
-	for _, path := range paths {
-		if _, exists := seen[path]; exists {
-			continue
-		}
-		seen[path] = struct{}{}
-		owner, exists := status.SharedRoot.OutputOwner(path)
-		if !exists || owner == scope {
-			continue
-		}
-		conflicts = append(conflicts, state.PathOwnershipConflict{
-			Path:         path,
-			Owner:        owner,
-			CurrentOwner: scope,
-			Detail: state.ComparisonDetail{
-				Kind:             state.ComparisonDetailSharedRootOutputOwner,
-				Path:             path,
-				CurrentOwner:     scope,
-				ConflictingOwner: owner,
-			},
-		})
-	}
-	return conflicts
+	return paths
 }

-func sharedRootConflictOutputs(sharedRoot *state.SharedRootState, conflicts []state.PathOwnershipConflict) []state.SharedRootOutputFile {
-	if sharedRoot == nil || len(conflicts) == 0 {
-		return nil
+func normalizeReconciliation(policy config.ReconciliationPolicy) config.ReconciliationPolicy {
+	if policy.Mode == "" {
+		policy.Mode = config.ReconciliationModeReplace
 	}
-	paths := make(map[string]struct{}, len(conflicts))
-	for _, conflict := range conflicts {
-		paths[conflict.Path] = struct{}{}
-	}
-	outputs := make([]state.SharedRootOutputFile, 0, len(conflicts))
-	for _, output := range sharedRoot.Outputs {
-		if _, exists := paths[output.Path]; exists {
-			outputs = append(outputs, output)
-		}
-	}
-	return outputs
+	return policy
 }

-func sharedRootTakeoverAllowed(req Request, status state.DestinationStatus, conflict state.PathOwnershipConflict) bool {
-	if status.SharedRoot == nil {
-		return false
-	}
-	takeover := normalizeTakeover(req.Takeover)
-	switch takeover.Mode {
-	case config.TakeoverModeSamePipeline:
-		return conflict.Owner.PipelineID == req.PipelineID
-	case config.TakeoverModeSameSource:
-		owner, ok := status.SharedRoot.Owner(conflict.Owner)
-		return ok && owner.Source.Manifest.ID == req.SourceBundle.Manifest.ID
-	case config.TakeoverModeAnyManaged:
-		_, ok := status.SharedRoot.Owner(conflict.Owner)
-		return ok
-	case config.TakeoverModeNever:
-		return false
-	default:
-		return 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 {
-	return otherOwnerOutputsExcept(status, scope, nil)
-}
-
-func otherOwnerOutputsExcept(status state.DestinationStatus, scope state.OwnerScope, exclude map[string]struct{}) []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 {
-			continue
-		}
-		if _, skip := exclude[output.Path]; skip {
-			continue
-		}
-		outputs = append(outputs, output)
-	}
-	return outputs
+func isReconciliationReplacementAction(action Action) bool {
+	return action == ActionReplaceOlder || action == ActionReplaceNewer
 }

 func sharedRootOutputPathSet(outputs []state.SharedRootOutputFile) map[string]struct{} {
@@ -465,109 +353,3 @@ func sharedRootOutputPathSet(outputs []state.SharedRootOutputFile) map[string]st
 	}
 	return paths
 }
-
-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) {
-	switch comparison.Outcome {
-	case state.OutcomeDestinationAbsent:
-		return ActionPublishNew, comparison.Reason
-	case state.OutcomeDestinationUnmanaged:
-		if force {
-			return ActionForceReplace, "forced replacement of unmanaged destination content"
-		}
-		return ActionFailUnmanaged, comparison.Reason
-	case state.OutcomeInvalidState:
-		return ActionFailConflict, comparison.Reason
-	case state.OutcomeIdentityMismatch, state.OutcomeSameCreatedConflict, state.OutcomeDifferentSourceConflict:
-		if transfer.OnConflict == config.TransferActionReplace {
-			return ActionReplaceConflict, comparison.Reason
-		}
-		return ActionFailConflict, comparison.Reason
-	case state.OutcomeSameSource:
-		if transfer.OnDestinationSame == config.TransferActionFail {
-			return ActionFailConflict, "destination matches source and transfer policy requires failure"
-		}
-		return ActionSkipSame, comparison.Reason
-	case state.OutcomeDestinationOlder:
-		if transfer.OnDestinationOlder == config.TransferActionFail {
-			return ActionFailConflict, "destination is older and transfer policy requires failure"
-		}
-		return ActionReplaceOlder, comparison.Reason
-	case state.OutcomeDestinationNewer:
-		if transfer.OnDestinationNewer == config.TransferActionReplace {
-			return ActionReplaceNewer, comparison.Reason
-		}
-		if transfer.OnDestinationNewer == config.TransferActionFail {
-			return ActionFailConflict, "destination is newer and transfer policy requires failure"
-		}
-		return ActionSkipDestinationNewer, comparison.Reason
-	default:
-		return ActionFailConflict, "unsupported comparison outcome"
-	}
-}
-
-func isReconciliationReplacementAction(action Action) bool {
-	return action == ActionReplaceOlder || action == ActionReplaceNewer
-}
-
-func takeoverActionAllowed(req Request, status state.DestinationStatus, comparison state.Comparison, stateMode string, action Action) bool {
-	if stateMode != config.StateModeSingleOwner || status.State == nil {
-		return false
-	}
-	if action == ActionForceReplace {
-		return false
-	}
-	switch comparison.Detail.Kind {
-	case state.ComparisonDetailPipelineIDMismatch,
-		state.ComparisonDetailDestinationIDMismatch,
-		state.ComparisonDetailDifferentSourceID:
-	default:
-		return false
-	}
-	takeover := normalizeTakeover(req.Takeover)
-	switch takeover.Mode {
-	case config.TakeoverModeSamePipeline:
-		return status.State.PipelineID == req.PipelineID
-	case config.TakeoverModeSameSource:
-		return status.State.Source.Manifest.ID == req.SourceBundle.Manifest.ID
-	case config.TakeoverModeAnyManaged:
-		return true
-	case config.TakeoverModeNever:
-		return false
-	default:
-		return false
-	}
-}