Extract destination run processing

docs/internal/app.md

@@ -201,6 +201,8 @@ Run helpers are grouped by responsibility:
 - `runtime.go`: runtime config path resolution, config loading, secret loading,
   environment resolver handoff, and secret-conflict warning projection.
 - `run.go`: `Run`, `RunPipeline`, and shared run orchestration.
+- `run_destination.go`: destination-scoped planning, execution, action
+  recording, and failure bookkeeping.
 - `run_output.go`: `RunReport`, action/output records, and text/JSON report projection.
 - `run_summary.go`: summary counters.
 - `run_failures.go`: destination failure aggregation and partial-result detection.

internal/app/run.go

@@ -8,7 +8,6 @@ import (
 	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
 	"gitea.maximumdirect.net/eric/distributor/internal/notify"
-	"gitea.maximumdirect.net/eric/distributor/internal/publish"
 	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 )
 
@@ -213,6 +212,11 @@ func buildRunReportWithSetup(ctx context.Context, setup runtimeSetup, options Ru
 		Actions:   []RunActionRecord{},
 	}
 	var failures runFailures
+	recorder := runReportRecorder{
+		report:   &report,
+		summary:  &summary,
+		failures: &failures,
+	}
 	report.PreambleWarnings = append(report.PreambleWarnings, setup.Warnings...)
 	report.addWarnings(setup.Warnings)
 	backends := provider(setup.Environment)
@@ -234,103 +238,18 @@ func buildRunReportWithSetup(ctx context.Context, setup runtimeSetup, options Ru
 		})
 		pipelineIndex := len(report.Pipelines) - 1
 		for _, destination := range pipeline.Destinations {
-			selections := selectDestinationBundles(destination, bundles)
+			processDestination(ctx, runDestinationRequest{
-			if isFixedPathDestination(destination) {
+				options:       options,
-				summary.recordFixedPath()
+				notifier:      notifier,
-				if options.DryRun {
+				backends:      backends,
-					warning := fixedPathSelectionWarning(pipeline.ID, destination.ID, selections, len(bundles))
+				transforms:    transforms,
-					report.addWarning(warning)
+				pipeline:      pipeline,
-					report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning))
+				pipelineIndex: pipelineIndex,
-				}
+				sourceBackend: sourceBackend,
-			}
+				bundles:       bundles,
-			if len(selections) == 0 {
+				destination:   destination,
-				continue
+				recorder:      &recorder,
-			}
+			})
-			destinationBackend, err := backends.openDestination(ctx, destination)
-			if err != nil {
-				for _, selection := range selections {
-					failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(selection.SourceBundle.RootRelativePath), err)
-					summary.recordFailure()
-					report.Actions = append(report.Actions, errorAction(pipeline.ID, destination.ID, destination.Backend, selection.SourceBundle.RootRelativePath, err))
-					report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, actionEvent(len(report.Actions)-1))
-				}
-				continue
-			}
-			closeDestination := true
-			deferCloseDestination := func() {
-				if closeDestination {
-					closeBackend(destinationBackend)
-					closeDestination = false
-				}
-			}
-			for _, selection := range selections {
-				sourceBundle := selection.SourceBundle
-				req := publish.Request{
-					PipelineID:            pipeline.ID,
-					DestinationID:         destination.ID,
-					SourceBundle:          sourceBundle,
-					SourceBackend:         sourceBackend,
-					DestinationBackend:    destinationBackend,
-					DestinationBundlePath: selection.DestinationBundlePath,
-					PathMapping:           destination.PathMap.Mode,
-					Publish:               *destination.Publish,
-					Transform:             destination.Transform,
-					Links:                 destination.Links,
-					Transformers:          transforms,
-					Transfer:              destination.Transfer,
-					DistributorVersion:    Version,
-					Force:                 options.Force,
-				}
-				plan, err := publish.Build(ctx, req)
-				if err != nil {
-					if plan.PipelineID == "" {
-						plan.PipelineID = pipeline.ID
-					}
-					if plan.DestinationID == "" {
-						plan.DestinationID = destination.ID
-					}
-					if plan.BundleID == "" {
-						plan.BundleID = sourceBundle.Manifest.ID
-					}
-					if plan.BundlePath == "" {
-						plan.BundlePath = sourceBundle.RootRelativePath
-					}
-					if plan.DestinationBundlePath == "" {
-						plan.DestinationBundlePath = selection.DestinationBundlePath
-					}
-				}
-				if isFixedPathDestination(destination) {
-					plan.PathMapping = config.PathMappingFixed
-					if options.DryRun && isDestructiveFixedPathAction(plan.Action) {
-						warning := fixedPathReplacementWarning(plan)
-						report.addWarning(warning)
-						report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning))
-					}
-				}
-				report.Actions = append(report.Actions, runActionFromPlan(destination.Backend, plan, err))
-				report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, actionEvent(len(report.Actions)-1))
-				if err != nil {
-					failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
-					summary.recordFailure()
-					continue
-				}
-				summary.recordPlan(plan.Action)
-				if !options.DryRun {
-					if err := publish.Execute(ctx, req, plan); err != nil {
-						failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
-						summary.recordFailure()
-						continue
-					}
-					if shouldNotify(plan.Action) {
-						if err := notifier.Notify(ctx, notifyEvent(plan)); err != nil {
-							failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
-							summary.recordFailure()
-							continue
-						}
-					}
-				}
-			}
-			deferCloseDestination()
 		}
 		closeBackend(sourceBackend)
 	}
@@ -342,6 +261,12 @@ func buildRunReportWithSetup(ctx context.Context, setup runtimeSetup, options Ru
 	return report, nil
 }
 
+type runReportRecorder struct {
+	report   *RunReport
+	summary  *runSummary
+	failures *runFailures
+}
+
 func openPipelineSource(ctx context.Context, backends *backendFactory, pipeline config.Pipeline, sourceRoot *localSourceRoot) (storage.Backend, []bundle.Bundle, string, error) {
 	if sourceRoot != nil && sourceRoot.pipelineID == pipeline.ID {
 		sourceBackend, err := backends.openLocalPath(ctx, sourceRoot.root)

internal/app/run_destination.go

@@ -0,0 +1,163 @@
+package app
+
+import (
+	"context"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/notify"
+	"gitea.maximumdirect.net/eric/distributor/internal/publish"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+type runDestinationRequest struct {
+	options       RunOptions
+	notifier      notify.Notifier
+	backends      *backendFactory
+	transforms    publish.TransformerResolver
+	pipeline      config.Pipeline
+	pipelineIndex int
+	sourceBackend storage.Backend
+	bundles       []bundle.Bundle
+	destination   config.Destination
+	recorder      *runReportRecorder
+}
+
+func processDestination(ctx context.Context, request runDestinationRequest) {
+	selections := selectDestinationBundles(request.destination, request.bundles)
+	if isFixedPathDestination(request.destination) {
+		request.recorder.summary.recordFixedPath()
+		if request.options.DryRun {
+			warning := fixedPathSelectionWarning(request.pipeline.ID, request.destination.ID, selections, len(request.bundles))
+			request.recorder.addPipelineWarning(request.pipelineIndex, warning)
+		}
+	}
+	if len(selections) == 0 {
+		return
+	}
+	destinationBackend, err := request.backends.openDestination(ctx, request.destination)
+	if err != nil {
+		for _, selection := range selections {
+			sourceBundle := selection.SourceBundle
+			request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{
+				pipelineID:    request.pipeline.ID,
+				destinationID: request.destination.ID,
+				backend:       request.destination.Backend,
+				bundlePath:    sourceBundle.RootRelativePath,
+				err:           err,
+			}, errorAction(request.pipeline.ID, request.destination.ID, request.destination.Backend, sourceBundle.RootRelativePath, err), true)
+		}
+		return
+	}
+	defer closeBackend(destinationBackend)
+
+	for _, selection := range selections {
+		processDestinationSelection(ctx, request, destinationBackend, selection)
+	}
+}
+
+func processDestinationSelection(ctx context.Context, request runDestinationRequest, destinationBackend storage.Backend, selection destinationBundleSelection) {
+	sourceBundle := selection.SourceBundle
+	publishRequest := publish.Request{
+		PipelineID:            request.pipeline.ID,
+		DestinationID:         request.destination.ID,
+		SourceBundle:          sourceBundle,
+		SourceBackend:         request.sourceBackend,
+		DestinationBackend:    destinationBackend,
+		DestinationBundlePath: selection.DestinationBundlePath,
+		PathMapping:           request.destination.PathMap.Mode,
+		Publish:               *request.destination.Publish,
+		Transform:             request.destination.Transform,
+		Links:                 request.destination.Links,
+		Transformers:          request.transforms,
+		Transfer:              request.destination.Transfer,
+		DistributorVersion:    Version,
+		Force:                 request.options.Force,
+	}
+	plan, err := publish.Build(ctx, publishRequest)
+	if err != nil {
+		plan = completePlanIdentity(plan, request.pipeline, request.destination, selection)
+	}
+	if isFixedPathDestination(request.destination) {
+		plan.PathMapping = config.PathMappingFixed
+		if request.options.DryRun && isDestructiveFixedPathAction(plan.Action) {
+			warning := fixedPathReplacementWarning(plan)
+			request.recorder.addPipelineWarning(request.pipelineIndex, warning)
+		}
+	}
+	action := runActionFromPlan(request.destination.Backend, plan, err)
+	if err != nil {
+		request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{
+			pipelineID:    request.pipeline.ID,
+			destinationID: request.destination.ID,
+			backend:       request.destination.Backend,
+			bundlePath:    sourceBundle.RootRelativePath,
+			err:           err,
+		}, action, true)
+		return
+	}
+	request.recorder.addPipelineAction(request.pipelineIndex, action)
+	request.recorder.summary.recordPlan(plan.Action)
+	if request.options.DryRun {
+		return
+	}
+	if err := publish.Execute(ctx, publishRequest, plan); err != nil {
+		request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{
+			pipelineID:    request.pipeline.ID,
+			destinationID: request.destination.ID,
+			backend:       request.destination.Backend,
+			bundlePath:    sourceBundle.RootRelativePath,
+			err:           err,
+		}, RunActionRecord{}, false)
+		return
+	}
+	if shouldNotify(plan.Action) {
+		if err := request.notifier.Notify(ctx, notifyEvent(plan)); err != nil {
+			request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{
+				pipelineID:    request.pipeline.ID,
+				destinationID: request.destination.ID,
+				backend:       request.destination.Backend,
+				bundlePath:    sourceBundle.RootRelativePath,
+				err:           err,
+			}, RunActionRecord{}, false)
+			return
+		}
+	}
+}
+
+func (recorder *runReportRecorder) addPipelineWarning(pipelineIndex int, warning OutputWarning) {
+	recorder.report.addWarning(warning)
+	recorder.report.Pipelines[pipelineIndex].events = append(recorder.report.Pipelines[pipelineIndex].events, warningEvent(warning))
+}
+
+func (recorder *runReportRecorder) addPipelineAction(pipelineIndex int, action RunActionRecord) {
+	recorder.report.Actions = append(recorder.report.Actions, action)
+	recorder.report.Pipelines[pipelineIndex].events = append(recorder.report.Pipelines[pipelineIndex].events, actionEvent(len(recorder.report.Actions)-1))
+}
+
+func (recorder *runReportRecorder) recordDestinationFailure(pipelineIndex int, failure runFailure, action RunActionRecord, includeAction bool) {
+	recorder.failures.add(failure.pipelineID, failure.destinationID, failure.backend, storage.DisplayPath(failure.bundlePath), failure.err)
+	recorder.summary.recordFailure()
+	if includeAction {
+		recorder.addPipelineAction(pipelineIndex, action)
+	}
+}
+
+func completePlanIdentity(plan publish.Plan, pipeline config.Pipeline, destination config.Destination, selection destinationBundleSelection) publish.Plan {
+	if plan.PipelineID == "" {
+		plan.PipelineID = pipeline.ID
+	}
+	if plan.DestinationID == "" {
+		plan.DestinationID = destination.ID
+	}
+	if plan.BundleID == "" {
+		plan.BundleID = selection.SourceBundle.Manifest.ID
+	}
+	if plan.BundlePath == "" {
+		plan.BundlePath = selection.SourceBundle.RootRelativePath
+	}
+	if plan.DestinationBundlePath == "" {
+		plan.DestinationBundlePath = selection.DestinationBundlePath
+	}
+	return plan
+}

internal/app/run_test.go

@@ -922,6 +922,50 @@ func TestBuildRunReportIncludesPartialFailures(t *testing.T) {
 	}
 }
 
+func TestBuildRunReportAlignsDestinationOpenFailuresForSelectedBundles(t *testing.T) {
+	sourceRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "daily/one", testBundleOptions{ID: "reports.one"})
+	writeSourceBundle(t, sourceRoot, "daily/two", testBundleOptions{ID: "reports.two", Created: testutil.DefaultCreated.Add(time.Hour)})
+	cfg := config.Config{Pipelines: []config.Pipeline{{
+		ID:     "reports",
+		Source: config.Backend{Backend: config.BackendLocal, Path: sourceRoot},
+		Destinations: []config.Destination{{
+			ID:       "object-archive",
+			Backend:  config.BackendS3,
+			Endpoint: "http://s3.test",
+			Bucket:   "missing-destination",
+		}},
+	}}}
+	config.ApplyDefaults(&cfg)
+
+	report, err := buildRunReportWithBackendFactory(context.Background(), cfg, RunOptions{}, fakeBackendFactoryProvider(t, nil))
+	if err == nil || !IsPartialResultError(err) {
+		t.Fatalf("buildRunReportWithBackendFactory() error = %v, want partial result error", err)
+	}
+	if report.Summary.Status != "failed" || report.Summary.Planned != 0 || report.Summary.Failed != 2 {
+		t.Fatalf("summary = %#v, want two destination open failures", report.Summary)
+	}
+	if got, want := len(report.Actions), 2; got != want {
+		t.Fatalf("action count = %d, want %d", got, want)
+	}
+	if got, want := len(report.OutputErrors), 2; got != want {
+		t.Fatalf("output error count = %d, want %d", got, want)
+	}
+	if got, want := len(report.Pipelines[0].events), 2; got != want {
+		t.Fatalf("pipeline event count = %d, want %d", got, want)
+	}
+	for index, bundlePath := range []string{"daily/one", "daily/two"} {
+		action := report.Actions[index]
+		if action.PipelineID != "reports" || action.DestinationID != "object-archive" || action.Backend != config.BackendS3 || action.BundlePath != bundlePath || action.Action != "error" {
+			t.Fatalf("action[%d] = %#v, want %s destination open error", index, action, bundlePath)
+		}
+		outputError := report.OutputErrors[index]
+		if outputError.PipelineID != action.PipelineID || outputError.DestinationID != action.DestinationID || outputError.Backend != action.Backend || outputError.BundlePath != action.BundlePath {
+			t.Fatalf("output error[%d] = %#v, action = %#v, want aligned identity", index, outputError, action)
+		}
+	}
+}
+
 func TestRunPipelineRunsOnlyRequestedPipeline(t *testing.T) {
 	firstSource := t.TempDir()
 	secondSource := t.TempDir()