diff --git a/docs/internal/app.md b/docs/internal/app.md index 63c0ee8..534bfba 100644 --- a/docs/internal/app.md +++ b/docs/internal/app.md @@ -79,6 +79,43 @@ destination planning and execution loop. Destination code receives the normal storage backend and bundle values and does not depend on how the source root was prepared. +## Upload Coordination + +`UploadCoordinator` owns in-memory coordination for asynchronous upload +processing. It admits uploads for configured `http_upload` pipelines, generates +run IDs, tracks status records, stages accepted archives through +`internal/ingest`, and executes the selected pipeline through +`RunPipelineWithLocalSource`. + +Upload run IDs use: + +```text +.. +``` + +The timestamp uses `YYYYMMDDThhmmssZ` UTC format and the suffix is filesystem +safe. + +The coordinator records these statuses: + +- `accepted` +- `queued` +- `running` +- `succeeded` +- `failed` +- `expired` + +Admission is bounded by `server.http.queue_size`. Full queues are rejected +before the upload body is staged. Execution is bounded by +`server.http.max_concurrency`, and only one upload for a given pipeline may run +at a time. Later uploads for the same pipeline remain queued until the active +run finishes. + +Completed records retain the final run report or error text until +`server.http.retention` elapses. Expiration removes completed status records and +their committed staged bundle directories. The coordinator is memory-only and +does not persist queue state, status records, or run reports. + ## Coordination `PipelineRunCoordinator` wraps `RunPipeline` with in-memory admission control. @@ -133,6 +170,7 @@ Run helpers are grouped by responsibility: - `run_warnings.go`: secret and SSH warning records. - `run_notify.go`: notification event projection and action filtering. - `run_coordinator.go`: in-memory run admission, run IDs, status records, and duplicate-run errors. +- `upload_coordinator.go`: in-memory upload admission, queueing, status tracking, staging handoff, and staged-source execution. - `backends.go`: app-level backend factory wiring. - `transforms.go`: app-level transform registry wiring. - `source_select.go`: configured-source selection shared by `validate` and `inspect`. diff --git a/internal/app/upload_coordinator.go b/internal/app/upload_coordinator.go new file mode 100644 index 0000000..33a53a6 --- /dev/null +++ b/internal/app/upload_coordinator.go @@ -0,0 +1,382 @@ +package app + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "io" + "os" + "sync" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/ingest" +) + +const DefaultUploadMaxFileCount = 4096 + +type UploadRunID string + +type UploadStatus string + +const ( + UploadStatusAccepted UploadStatus = "accepted" + UploadStatusQueued UploadStatus = "queued" + UploadStatusRunning UploadStatus = "running" + UploadStatusSucceeded UploadStatus = "succeeded" + UploadStatusFailed UploadStatus = "failed" + UploadStatusExpired UploadStatus = "expired" +) + +type UploadRunRecord struct { + ID UploadRunID `json:"run_id"` + PipelineID string `json:"pipeline_id"` + Status UploadStatus `json:"status"` + AcceptedAt time.Time `json:"accepted_at"` + StartedAt *time.Time `json:"started_at,omitempty"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + Report *RunReport `json:"report,omitempty"` + Error string `json:"error,omitempty"` + StagedRoot string `json:"-"` +} + +type UploadRequest struct { + PipelineID string + ContentType string + Body io.Reader + DryRun bool + Force bool + MaxFileCount int +} + +type UploadQueueFullError struct { + QueueSize int +} + +func (err UploadQueueFullError) Error() string { + return fmt.Sprintf("upload queue is full with capacity %d", err.QueueSize) +} + +func IsUploadQueueFull(err error) bool { + var full UploadQueueFullError + return errors.As(err, &full) +} + +type UploadCoordinator struct { + ctx context.Context + cfg config.Config + stage uploadStageFunc + run uploadRunFunc + now func() time.Time + randomSuffix func() (string, error) + retention time.Duration + + mu sync.Mutex + signal chan struct{} + queueSize int + maxConcurrency int + runningCount int + activePipeline map[string]bool + pending []*uploadJob + records map[UploadRunID]UploadRunRecord +} + +type uploadStageFunc func(context.Context, ingest.StageOptions) (ingest.StagedBundle, error) + +type uploadRunFunc func(context.Context, config.Config, RunPipelineWithLocalSourceOptions) (RunReport, error) + +type uploadJob struct { + recordID UploadRunID + request UploadRequest + pipeline config.Pipeline +} + +type uploadCoordinatorHooks struct { + stage uploadStageFunc + run uploadRunFunc + now func() time.Time + randomSuffix func() (string, error) +} + +func NewUploadCoordinator(ctx context.Context, cfg config.Config) *UploadCoordinator { + return newUploadCoordinator(ctx, cfg, uploadCoordinatorHooks{}) +} + +func newUploadCoordinator(ctx context.Context, cfg config.Config, hooks uploadCoordinatorHooks) *UploadCoordinator { + if ctx == nil { + ctx = context.Background() + } + config.ApplyDefaults(&cfg) + stage := hooks.stage + if stage == nil { + stage = ingest.StageArchive + } + run := hooks.run + if run == nil { + run = runPipelineConfigWithLocalSource + } + now := hooks.now + if now == nil { + now = time.Now + } + randomSuffix := hooks.randomSuffix + if randomSuffix == nil { + randomSuffix = randomRunIDSuffix + } + coordinator := &UploadCoordinator{ + ctx: ctx, + cfg: cfg, + stage: stage, + run: run, + now: now, + randomSuffix: randomSuffix, + retention: cfg.Server.HTTP.Retention.AsDuration(), + signal: make(chan struct{}, 1), + queueSize: cfg.Server.HTTP.QueueSize, + maxConcurrency: cfg.Server.HTTP.MaxConcurrency, + activePipeline: map[string]bool{}, + records: map[UploadRunID]UploadRunRecord{}, + } + go coordinator.dispatchLoop() + return coordinator +} + +func (coordinator *UploadCoordinator) Submit(ctx context.Context, request UploadRequest) (UploadRunRecord, error) { + if ctx == nil { + ctx = context.Background() + } + if err := ctx.Err(); err != nil { + return UploadRunRecord{}, err + } + if request.Body == nil { + return UploadRunRecord{}, fmt.Errorf("upload body is required") + } + pipeline, ok := findPipeline(coordinator.cfg, request.PipelineID) + if !ok { + return UploadRunRecord{}, PipelineNotFoundError{ID: request.PipelineID} + } + if pipeline.Source.Backend != config.BackendHTTPUpload { + return UploadRunRecord{}, fmt.Errorf("pipeline %s source backend %s is not configured for uploads", pipeline.ID, pipeline.Source.Backend) + } + runID, err := coordinator.newRunID(pipeline.ID) + if err != nil { + return UploadRunRecord{}, err + } + + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + coordinator.expireLocked(coordinator.now().UTC()) + if len(coordinator.pending) >= coordinator.queueSize { + return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize} + } + record := UploadRunRecord{ + ID: runID, + PipelineID: pipeline.ID, + Status: UploadStatusAccepted, + AcceptedAt: coordinator.now().UTC(), + } + coordinator.records[runID] = record + coordinator.pending = append(coordinator.pending, &uploadJob{ + recordID: runID, + request: request, + pipeline: pipeline, + }) + coordinator.notify() + return record, nil +} + +func (coordinator *UploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + coordinator.expireLocked(coordinator.now().UTC()) + record, ok := coordinator.records[runID] + return record, ok +} + +func (coordinator *UploadCoordinator) Expire() []UploadRunRecord { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + return coordinator.expireLocked(coordinator.now().UTC()) +} + +func (coordinator *UploadCoordinator) QueueDepth() int { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + return len(coordinator.pending) +} + +func (coordinator *UploadCoordinator) RunningCount() int { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + return coordinator.runningCount +} + +func (coordinator *UploadCoordinator) newRunID(pipelineID string) (UploadRunID, error) { + suffix, err := coordinator.randomSuffix() + if err != nil { + return "", err + } + timestamp := coordinator.now().UTC().Format("20060102T150405Z") + return UploadRunID(pipelineID + "." + timestamp + "." + suffix), nil +} + +func (coordinator *UploadCoordinator) dispatchLoop() { + for { + select { + case <-coordinator.ctx.Done(): + return + case <-coordinator.signal: + for coordinator.startNext() { + } + } + } +} + +func (coordinator *UploadCoordinator) startNext() bool { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + if coordinator.runningCount >= coordinator.maxConcurrency { + coordinator.markPendingQueuedLocked() + return false + } + index := -1 + for candidateIndex, job := range coordinator.pending { + if coordinator.activePipeline[job.pipeline.ID] { + record := coordinator.records[job.recordID] + if record.Status == UploadStatusAccepted { + record.Status = UploadStatusQueued + coordinator.records[job.recordID] = record + } + continue + } + index = candidateIndex + break + } + if index < 0 { + return false + } + + job := coordinator.pending[index] + coordinator.pending = append(coordinator.pending[:index], coordinator.pending[index+1:]...) + now := coordinator.now().UTC() + record := coordinator.records[job.recordID] + record.Status = UploadStatusRunning + record.StartedAt = &now + coordinator.records[job.recordID] = record + coordinator.runningCount++ + coordinator.activePipeline[job.pipeline.ID] = true + + go coordinator.runJob(job) + return true +} + +func (coordinator *UploadCoordinator) markPendingQueuedLocked() { + for _, job := range coordinator.pending { + record := coordinator.records[job.recordID] + if record.Status == UploadStatusAccepted { + record.Status = UploadStatusQueued + coordinator.records[job.recordID] = record + } + } +} + +func (coordinator *UploadCoordinator) runJob(job *uploadJob) { + record := coordinator.currentRecord(job.recordID) + maxFileCount := job.request.MaxFileCount + if maxFileCount <= 0 { + maxFileCount = DefaultUploadMaxFileCount + } + staged, err := coordinator.stage(coordinator.ctx, ingest.StageOptions{ + Body: job.request.Body, + ContentType: job.request.ContentType, + PipelineStagingPath: job.pipeline.Source.Upload.StagingPath, + RunID: string(record.ID), + MaxUploadSize: int64(*job.pipeline.Source.Upload.MaxUploadSize), + MaxExtractedSize: int64(*job.pipeline.Source.Upload.MaxUploadSize), + MaxFileCount: maxFileCount, + }) + if err == nil { + coordinator.setStagedRoot(job.recordID, staged.Root) + var report RunReport + report, err = coordinator.run(coordinator.ctx, coordinator.cfg, RunPipelineWithLocalSourceOptions{ + PipelineID: job.pipeline.ID, + SourceRoot: staged.Root, + DryRun: job.request.DryRun, + Force: job.request.Force, + }) + coordinator.complete(job, &report, err) + return + } + coordinator.complete(job, nil, err) +} + +func (coordinator *UploadCoordinator) currentRecord(runID UploadRunID) UploadRunRecord { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + return coordinator.records[runID] +} + +func (coordinator *UploadCoordinator) setStagedRoot(runID UploadRunID, root string) { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + record := coordinator.records[runID] + record.StagedRoot = root + coordinator.records[runID] = record +} + +func (coordinator *UploadCoordinator) complete(job *uploadJob, report *RunReport, runErr error) { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + record := coordinator.records[job.recordID] + finishedAt := coordinator.now().UTC() + record.FinishedAt = &finishedAt + record.Report = report + if runErr != nil { + record.Status = UploadStatusFailed + record.Error = runErr.Error() + } else { + record.Status = UploadStatusSucceeded + } + coordinator.records[job.recordID] = record + coordinator.runningCount-- + delete(coordinator.activePipeline, job.pipeline.ID) + coordinator.notify() +} + +func (coordinator *UploadCoordinator) expireLocked(now time.Time) []UploadRunRecord { + var expired []UploadRunRecord + for runID, record := range coordinator.records { + if record.FinishedAt == nil || record.Status == UploadStatusExpired { + continue + } + if now.Before(record.FinishedAt.Add(coordinator.retention)) { + continue + } + if record.StagedRoot != "" { + _ = os.RemoveAll(record.StagedRoot) + } + record.Status = UploadStatusExpired + record.Report = nil + record.Error = "" + expired = append(expired, record) + delete(coordinator.records, runID) + } + return expired +} + +func (coordinator *UploadCoordinator) notify() { + select { + case coordinator.signal <- struct{}{}: + default: + } +} + +func randomRunIDSuffix() (string, error) { + var data [4]byte + if _, err := rand.Read(data[:]); err != nil { + return "", fmt.Errorf("generate run id suffix: %w", err) + } + return hex.EncodeToString(data[:]), nil +} diff --git a/internal/app/upload_coordinator_test.go b/internal/app/upload_coordinator_test.go new file mode 100644 index 0000000..c540c2f --- /dev/null +++ b/internal/app/upload_coordinator_test.go @@ -0,0 +1,373 @@ +package app + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/ingest" +) + +func TestUploadCoordinatorGeneratesRunIDAndAcceptedStatus(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + clock := newUploadTestClock(time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC)) + coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{ + pipelineIDs: []string{"weather-daily"}, + }), uploadCoordinatorHooks{ + now: clock.Now, + randomSuffix: uploadTestSuffixes("ab12cd34"), + stage: successfulUploadStage, + run: successfulUploadRun, + }) + + record, err := coordinator.Submit(context.Background(), UploadRequest{ + PipelineID: "weather-daily", + ContentType: ingest.ContentTypeTar, + Body: strings.NewReader("archive"), + }) + if err != nil { + t.Fatalf("Submit() error = %v", err) + } + if got, want := record.ID, UploadRunID("weather-daily.20260603T120000Z.ab12cd34"); got != want { + t.Fatalf("run id = %q, want %q", got, want) + } + if got, want := record.Status, UploadStatusAccepted; got != want { + t.Fatalf("initial status = %q, want %q", got, want) + } + waitForUploadStatus(t, coordinator, record.ID, UploadStatusSucceeded) +} + +func TestUploadCoordinatorRejectsFullQueueBeforeReadingBody(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + release := make(chan struct{}) + var reads atomic.Int64 + coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{ + pipelineIDs: []string{"reports"}, + queueSize: 1, + maxConcurrency: 1, + }), uploadCoordinatorHooks{ + randomSuffix: uploadTestSuffixes("00000001", "00000002", "00000003"), + stage: successfulUploadStage, + run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + <-release + return RunReport{}, nil + }, + }) + + first, err := coordinator.Submit(context.Background(), UploadRequest{ + PipelineID: "reports", + ContentType: ingest.ContentTypeTar, + Body: strings.NewReader("first"), + }) + if err != nil { + t.Fatalf("first Submit() error = %v", err) + } + waitForUploadStatus(t, coordinator, first.ID, UploadStatusRunning) + + second, err := coordinator.Submit(context.Background(), UploadRequest{ + PipelineID: "reports", + ContentType: ingest.ContentTypeTar, + Body: strings.NewReader("second"), + }) + if err != nil { + t.Fatalf("second Submit() error = %v", err) + } + waitForUploadStatus(t, coordinator, second.ID, UploadStatusQueued) + + _, err = coordinator.Submit(context.Background(), UploadRequest{ + PipelineID: "reports", + ContentType: ingest.ContentTypeTar, + Body: readerFunc(func(data []byte) (int, error) { + reads.Add(1) + return 0, io.EOF + }), + }) + if err == nil || !IsUploadQueueFull(err) { + t.Fatalf("third Submit() error = %v, want full queue", err) + } + if got := reads.Load(); got != 0 { + t.Fatalf("rejected body reads = %d, want 0", got) + } + close(release) + waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded) + waitForUploadStatus(t, coordinator, second.ID, UploadStatusSucceeded) +} + +func TestUploadCoordinatorSerializesSamePipelineUploads(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + release := make(chan struct{}) + coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{ + pipelineIDs: []string{"reports"}, + queueSize: 4, + maxConcurrency: 2, + }), uploadCoordinatorHooks{ + randomSuffix: uploadTestSuffixes("00000001", "00000002"), + stage: successfulUploadStage, + run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + <-release + return RunReport{}, nil + }, + }) + + first, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("first")}) + if err != nil { + t.Fatalf("first Submit() error = %v", err) + } + second, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("second")}) + if err != nil { + t.Fatalf("second Submit() error = %v", err) + } + waitForUploadStatus(t, coordinator, first.ID, UploadStatusRunning) + waitForUploadStatus(t, coordinator, second.ID, UploadStatusQueued) + if got := coordinator.RunningCount(); got != 1 { + t.Fatalf("running count = %d, want 1", got) + } + close(release) + waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded) + waitForUploadStatus(t, coordinator, second.ID, UploadStatusSucceeded) +} + +func TestUploadCoordinatorRunsDifferentPipelinesConcurrentlyUpToLimit(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + release := make(chan struct{}) + coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{ + pipelineIDs: []string{"reports-one", "reports-two"}, + queueSize: 4, + maxConcurrency: 2, + }), uploadCoordinatorHooks{ + randomSuffix: uploadTestSuffixes("00000001", "00000002"), + stage: successfulUploadStage, + run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + <-release + return RunReport{}, nil + }, + }) + + first, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports-one", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("first")}) + if err != nil { + t.Fatalf("first Submit() error = %v", err) + } + second, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports-two", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("second")}) + if err != nil { + t.Fatalf("second Submit() error = %v", err) + } + waitForUploadStatus(t, coordinator, first.ID, UploadStatusRunning) + waitForUploadStatus(t, coordinator, second.ID, UploadStatusRunning) + if got := coordinator.RunningCount(); got != 2 { + t.Fatalf("running count = %d, want 2", got) + } + close(release) + waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded) + waitForUploadStatus(t, coordinator, second.ID, UploadStatusSucceeded) +} + +func TestUploadCoordinatorRecordsFailureDetails(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + runErr := errors.New("publish failed") + coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{ + pipelineIDs: []string{"reports"}, + }), uploadCoordinatorHooks{ + randomSuffix: uploadTestSuffixes("00000001"), + stage: successfulUploadStage, + run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + return RunReport{DryRun: options.DryRun}, runErr + }, + }) + + record, err := coordinator.Submit(context.Background(), UploadRequest{ + PipelineID: "reports", + ContentType: ingest.ContentTypeTar, + Body: strings.NewReader("archive"), + DryRun: true, + }) + if err != nil { + t.Fatalf("Submit() error = %v", err) + } + failed := waitForUploadStatus(t, coordinator, record.ID, UploadStatusFailed) + if failed.Error != runErr.Error() { + t.Fatalf("error = %q, want %q", failed.Error, runErr.Error()) + } + if failed.Report == nil || !failed.Report.DryRun { + t.Fatalf("report = %#v, want retained dry-run report", failed.Report) + } +} + +func TestUploadCoordinatorExpiresCompletedRecordsAndStagingDirectories(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + clock := newUploadTestClock(time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC)) + coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{ + pipelineIDs: []string{"reports"}, + retention: time.Second, + }), uploadCoordinatorHooks{ + now: clock.Now, + randomSuffix: uploadTestSuffixes("00000001"), + stage: successfulUploadStage, + run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + return RunReport{DryRun: true}, nil + }, + }) + + record, err := coordinator.Submit(context.Background(), UploadRequest{ + PipelineID: "reports", + ContentType: ingest.ContentTypeTar, + Body: strings.NewReader("archive"), + }) + if err != nil { + t.Fatalf("Submit() error = %v", err) + } + succeeded := waitForUploadStatus(t, coordinator, record.ID, UploadStatusSucceeded) + if succeeded.Report == nil || !succeeded.Report.DryRun { + t.Fatalf("report = %#v, want retained dry-run report", succeeded.Report) + } + if _, err := os.Stat(succeeded.StagedRoot); err != nil { + t.Fatalf("staged root stat before expiry = %v", err) + } + + clock.Advance(2 * time.Second) + expired := coordinator.Expire() + if got, want := len(expired), 1; got != want { + t.Fatalf("expired count = %d, want %d", got, want) + } + if expired[0].Status != UploadStatusExpired || expired[0].Report != nil || expired[0].Error != "" { + t.Fatalf("expired record = %#v, want expired without report/error", expired[0]) + } + if _, ok := coordinator.Status(record.ID); ok { + t.Fatal("Status() ok = true after expiry, want removed status") + } + if _, err := os.Stat(succeeded.StagedRoot); !os.IsNotExist(err) { + t.Fatalf("staged root stat after expiry = %v, want not exist", err) + } +} + +type readerFunc func([]byte) (int, error) + +func (fn readerFunc) Read(data []byte) (int, error) { + return fn(data) +} + +func successfulUploadStage(ctx context.Context, opts ingest.StageOptions) (ingest.StagedBundle, error) { + root := filepath.Join(opts.PipelineStagingPath, opts.RunID) + if err := os.MkdirAll(root, 0o755); err != nil { + return ingest.StagedBundle{}, err + } + return ingest.StagedBundle{Root: root}, nil +} + +func successfulUploadRun(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) { + return RunReport{}, nil +} + +type uploadCoordinatorConfigOptions struct { + pipelineIDs []string + queueSize int + maxConcurrency int + retention time.Duration +} + +func uploadCoordinatorConfig(t *testing.T, opts uploadCoordinatorConfigOptions) config.Config { + t.Helper() + queueSize := opts.queueSize + if queueSize == 0 { + queueSize = 4 + } + maxConcurrency := opts.maxConcurrency + if maxConcurrency == 0 { + maxConcurrency = 1 + } + retentionValue := opts.retention + if retentionValue == 0 { + retentionValue = time.Minute + } + retention := config.Duration(retentionValue) + maxUploadSize := config.ByteSize(1024) + cfg := config.Config{ + Server: config.Server{HTTP: config.HTTPServer{ + StagingRoot: t.TempDir(), + MaxUploadSize: &maxUploadSize, + QueueSize: queueSize, + MaxConcurrency: maxConcurrency, + Retention: &retention, + }}, + } + for _, pipelineID := range opts.pipelineIDs { + cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{ + ID: pipelineID, + Source: config.Backend{ + Backend: config.BackendHTTPUpload, + Upload: config.HTTPUpload{TokenEnv: strings.ToUpper(strings.ReplaceAll(pipelineID, "-", "_")) + "_TOKEN"}, + }, + Destinations: []config.Destination{{ + ID: "archive", + Backend: config.BackendLocal, + Path: t.TempDir(), + }}, + }) + } + return cfg +} + +type uploadTestClock struct { + mu sync.Mutex + now time.Time +} + +func newUploadTestClock(now time.Time) *uploadTestClock { + return &uploadTestClock{now: now} +} + +func (clock *uploadTestClock) Now() time.Time { + clock.mu.Lock() + defer clock.mu.Unlock() + return clock.now +} + +func (clock *uploadTestClock) Advance(duration time.Duration) { + clock.mu.Lock() + defer clock.mu.Unlock() + clock.now = clock.now.Add(duration) +} + +func uploadTestSuffixes(values ...string) func() (string, error) { + var mu sync.Mutex + index := 0 + return func() (string, error) { + mu.Lock() + defer mu.Unlock() + if index >= len(values) { + return fmt.Sprintf("%08d", index+1), nil + } + value := values[index] + index++ + return value, nil + } +} + +func waitForUploadStatus(t *testing.T, coordinator *UploadCoordinator, runID UploadRunID, status UploadStatus) UploadRunRecord { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + record, ok := coordinator.Status(runID) + if ok && record.Status == status { + return record + } + time.Sleep(time.Millisecond) + } + record, ok := coordinator.Status(runID) + t.Fatalf("timed out waiting for status %s; latest ok=%t record=%#v", status, ok, record) + return UploadRunRecord{} +}