From 9143a00bff139dd929c70f579b85648c1f8e6561 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 4 Jun 2026 00:28:23 +0000 Subject: [PATCH] Extract destination run processing --- docs/internal/app.md | 2 + internal/app/run.go | 121 +++++------------------- internal/app/run_destination.go | 163 ++++++++++++++++++++++++++++++++ internal/app/run_test.go | 44 +++++++++ 4 files changed, 232 insertions(+), 98 deletions(-) create mode 100644 internal/app/run_destination.go diff --git a/docs/internal/app.md b/docs/internal/app.md index 283ad6b..49d573e 100644 --- a/docs/internal/app.md +++ b/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. diff --git a/internal/app/run.go b/internal/app/run.go index 91e2a54..e8e2d0a 100644 --- a/internal/app/run.go +++ b/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) - if isFixedPathDestination(destination) { - summary.recordFixedPath() - if options.DryRun { - warning := fixedPathSelectionWarning(pipeline.ID, destination.ID, selections, len(bundles)) - report.addWarning(warning) - report.Pipelines[pipelineIndex].events = append(report.Pipelines[pipelineIndex].events, warningEvent(warning)) - } - } - if len(selections) == 0 { - continue - } - 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() + processDestination(ctx, runDestinationRequest{ + options: options, + notifier: notifier, + backends: backends, + transforms: transforms, + pipeline: pipeline, + pipelineIndex: pipelineIndex, + sourceBackend: sourceBackend, + bundles: bundles, + destination: destination, + recorder: &recorder, + }) } 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) diff --git a/internal/app/run_destination.go b/internal/app/run_destination.go new file mode 100644 index 0000000..7bebad9 --- /dev/null +++ b/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 +} diff --git a/internal/app/run_test.go b/internal/app/run_test.go index 1cb5b57..1be3d1d 100644 --- a/internal/app/run_test.go +++ b/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()