diff --git a/pkg/preparation/internal/worker/group.go b/pkg/preparation/internal/worker/group.go new file mode 100644 index 00000000..65b7561f --- /dev/null +++ b/pkg/preparation/internal/worker/group.go @@ -0,0 +1,128 @@ +package worker + +import ( + "context" + "sync" +) + +// Group runs a collection of tasks that return [TaskError]. Non-fatal errors +// are accumulated; a fatal error from any task cancels the context derived by +// [WithContext] so siblings watching that context can exit early. It's +// analogous to [golang.org/x/sync/errgroup.Group], with two deliberate +// differences: (1) it distinguishes fatal from non-fatal errors, and (2) it +// accumulates every task's errors rather than keeping only the first. +// +// If the group is cancelled, the cause will be the fatal error. +type Group struct { + ctx context.Context + cancel context.CancelCauseFunc + wg sync.WaitGroup + sem chan struct{} + + mu sync.Mutex + err taskError +} + +// WithContext returns a new [Group] and a derived context. The context is +// cancelled the first time a task reports a fatal error, or the first time +// [Group.Wait] returns, whichever occurs first. +func WithContext(ctx context.Context) (*Group, context.Context) { + ctx, cancel := context.WithCancelCause(ctx) + return &Group{ctx: ctx, cancel: cancel}, ctx +} + +// SetLimit limits the number of concurrently-running tasks to n. Must be +// called before any call to [Group.Go]. A value <= 0 removes any limit. +func (g *Group) SetLimit(n int) { + if n <= 0 { + g.sem = nil + return + } + g.sem = make(chan struct{}, n) +} + +// Go runs the given task in a new goroutine. If a concurrency limit is set, Go +// blocks until a slot is available. The task's [TaskError] is folded into the +// group's accumulated result; if the task returns a fatal error, the derived +// context is cancelled. +// +// If the group's context is cancelled when Go is called or while waiting for a +// slot, the task is not started and Go returns immediately. +func (g *Group) Go(task func() TaskError) { + if g.sem != nil { + select { + case g.sem <- struct{}{}: + case <-g.ctx.Done(): + return + } + } + g.launch(task) +} + +// TryGo runs the given task in a new goroutine if a slot is available. It +// returns true if the task was started, false otherwise. If no concurrency +// limit is set, TryGo always starts the task and returns true. +func (g *Group) TryGo(task func() TaskError) bool { + if g.sem != nil { + select { + case g.sem <- struct{}{}: + default: + return false + } + } + g.launch(task) + return true +} + +func (g *Group) launch(task func() TaskError) { + g.wg.Add(1) + go func() { + defer func() { + if g.sem != nil { + <-g.sem + } + g.wg.Done() + }() + select { + case <-g.ctx.Done(): + // Group has cancelled before this task started. Skip it silently: + // the real fatal is already in the accumulator, and an un-started + // task has nothing to report. Tasks already executing when cancel + // fires still run to completion and report whatever they want. + return + default: + } + g.collect(task()) + }() +} + +// Wait blocks until all goroutines started with [Group.Go] or [Group.TryGo] +// have returned, then returns the accumulated [TaskError]. It returns nil iff +// no task reported a fatal error and no task reported any non-fatal errors. +func (g *Group) Wait() TaskError { + g.wg.Wait() + + // Clean up resources: a `cancel` must always be called eventually. We've just + // `Wait()`ed for all tasks to finish, so this won't stop anything. + g.cancel(nil) + + g.mu.Lock() + defer g.mu.Unlock() + if g.err.isEmpty() { + return nil + } + result := g.err + return &result +} + +func (g *Group) collect(result TaskError) { + if result == nil { + return + } + g.mu.Lock() + g.err.add(result) + g.mu.Unlock() + if result.FatalError() != nil { + g.cancel(result.FatalError()) + } +} diff --git a/pkg/preparation/internal/worker/group_test.go b/pkg/preparation/internal/worker/group_test.go new file mode 100644 index 00000000..1530ccec --- /dev/null +++ b/pkg/preparation/internal/worker/group_test.go @@ -0,0 +1,78 @@ +package worker_test + +import ( + "errors" + "sync/atomic" + "testing" + + "github.com/storacha/guppy/pkg/preparation/internal/worker" + "github.com/stretchr/testify/require" +) + +func TestGroupLaunchSkipsIfCancelled(t *testing.T) { + t.Run("task is skipped if group ctx is already cancelled before launch", func(t *testing.T) { + fatalErr := errors.New("fatal") + + g, gctx := worker.WithContext(t.Context()) + + // First task cancels the group by returning a fatal. + g.Go(func() worker.TaskError { + return worker.NewFatalError(fatalErr) + }) + + // Wait for cancellation to propagate before enqueuing the second + // task, so the second task's goroutine sees a cancelled ctx on entry. + <-gctx.Done() + + var ranSecond atomic.Bool + g.Go(func() worker.TaskError { + ranSecond.Store(true) + return worker.NewFatalError(errors.New("should not be reported")) + }) + + res := g.Wait() + require.False(t, ranSecond.Load(), "task enqueued after cancellation should be skipped") + require.NotNil(t, res) + require.True(t, res.IsFatal()) + require.ErrorIs(t, res.FatalError(), fatalErr) + require.NotContains(t, res.FatalError().Error(), "should not be reported") + }) + + t.Run("in-flight task can still report errors after cancellation", func(t *testing.T) { + firstFatal := errors.New("first fatal") + secondFatal := errors.New("second fatal from in-flight task") + + g, gctx := worker.WithContext(t.Context()) + + started := make(chan struct{}) + release := make(chan struct{}) + + // In-flight task: signals it has started, waits to be released, then + // returns a fatal. It's already executing when the group cancels. + g.Go(func() worker.TaskError { + close(started) + <-release + return worker.NewFatalError(secondFatal) + }) + + // Wait for the in-flight task to be running. + <-started + + // Fire a fatal from another task to cancel the group. + g.Go(func() worker.TaskError { + return worker.NewFatalError(firstFatal) + }) + + // Wait for gctx to be cancelled. + <-gctx.Done() + + // Release the in-flight task so it can complete. + close(release) + + res := g.Wait() + require.NotNil(t, res) + require.True(t, res.IsFatal()) + require.ErrorIs(t, res.FatalError(), firstFatal, "first fatal should be recorded") + require.ErrorIs(t, res.FatalError(), secondFatal, "in-flight task's fatal should also be recorded") + }) +} diff --git a/pkg/preparation/internal/worker/types.go b/pkg/preparation/internal/worker/types.go new file mode 100644 index 00000000..1da80af4 --- /dev/null +++ b/pkg/preparation/internal/worker/types.go @@ -0,0 +1,133 @@ +package worker + +import ( + "context" + "errors" + "fmt" + "strings" +) + +// TaskError is the error type returned by a [Task]. A TaskError may carry a +// fatal error (which causes the worker to cancel dispatch of remaining tasks) +// and/or a set of non-fatal errors (which are collected and reported after +// all tasks complete). +type TaskError interface { + error + NonFatalErrors() []error + FatalError() error + IsFatal() bool +} + +// NewFatalError returns a [TaskError] carrying the given error as a fatal +// error. If err is nil, it returns nil. +func NewFatalError(err error) TaskError { + if err == nil { + return nil + } + return &taskError{fatalError: err} +} + +// NewNonFatalError returns a [TaskError] carrying the given errors as +// non-fatal errors. Any nil entries are dropped; if the result would be empty, +// it returns nil. +func NewNonFatalError(errs ...error) TaskError { + var filtered []error + for _, err := range errs { + if err != nil { + filtered = append(filtered, err) + } + } + if len(filtered) == 0 { + return nil + } + return &taskError{nonFatalErrors: filtered} +} + +type taskError struct { + nonFatalErrors []error + fatalError error +} + +// Task is a worker task. It returns a [TaskError] to signal the result: +// - return nil for success, +// - return [NewFatalError] (or any TaskError whose IsFatal is true) to abort +// further dispatch, +// - return [NewNonFatalError] to report errors that should be collected but +// not abort dispatch. +type Task func(context.Context) TaskError + +func (e *taskError) NonFatalErrors() []error { + return e.nonFatalErrors +} + +func (e *taskError) FatalError() error { + return e.fatalError +} + +func (e *taskError) IsFatal() bool { + return e.fatalError != nil +} + +func (e *taskError) Error() string { + nonFatalString := "" + if len(e.nonFatalErrors) > 0 { + nonFatalString = "non-fatal errors:" + } + for _, err := range e.nonFatalErrors { + nonFatalString += fmt.Sprintf("\n- %s", err) + } + + fatalString := "" + if e.fatalError != nil { + fatalString = fmt.Sprintf("fatal error: %s", e.fatalError) + } + + return fmt.Sprintf("worker encountered %s", strings.Join([]string{nonFatalString, fatalString}, "\n")) +} + +func (e *taskError) Unwrap() []error { + errs := make([]error, len(e.nonFatalErrors)) + copy(errs, e.nonFatalErrors) + if e.fatalError != nil { + errs = append(errs, e.fatalError) + } + return errs +} + +// add folds another [TaskError] into the receiver in place. +func (e *taskError) add(other TaskError) { + if other == nil { + return + } + e.nonFatalErrors = append(e.nonFatalErrors, other.NonFatalErrors()...) + if other.FatalError() != nil { + e.fatalError = errors.Join(e.fatalError, other.FatalError()) + } +} + +// isEmpty reports whether the taskError carries neither a fatal error nor any +// non-fatal errors. +func (e *taskError) isEmpty() bool { + return e.fatalError == nil && len(e.nonFatalErrors) == 0 +} + +// Join combines two [TaskError]s. Non-fatal errors are concatenated; fatal +// errors are joined with [errors.Join]. Either operand may be nil. +func Join(a TaskError, b TaskError) TaskError { + switch { + case a == nil && b == nil: + return nil + case a == nil: + return b + case b == nil: + return a + } + joined := &taskError{ + nonFatalErrors: append(a.NonFatalErrors(), b.NonFatalErrors()...), + fatalError: errors.Join(a.FatalError(), b.FatalError()), + } + if joined.isEmpty() { + return nil + } + return joined +} diff --git a/pkg/preparation/internal/worker/types_test.go b/pkg/preparation/internal/worker/types_test.go new file mode 100644 index 00000000..55aa0b8e --- /dev/null +++ b/pkg/preparation/internal/worker/types_test.go @@ -0,0 +1,94 @@ +package worker_test + +import ( + "errors" + "testing" + + "github.com/storacha/guppy/pkg/preparation/internal/worker" + "github.com/stretchr/testify/require" +) + +func TestNewFatalError(t *testing.T) { + t.Run("returns nil when err is nil", func(t *testing.T) { + require.Nil(t, worker.NewFatalError(nil)) + }) + + t.Run("wraps a non-nil error as fatal", func(t *testing.T) { + err := errors.New("boom") + te := worker.NewFatalError(err) + require.NotNil(t, te) + require.True(t, te.IsFatal()) + require.ErrorIs(t, te.FatalError(), err) + require.Empty(t, te.NonFatalErrors()) + }) +} + +func TestNewNonFatalError(t *testing.T) { + t.Run("returns nil when no errors are supplied", func(t *testing.T) { + require.Nil(t, worker.NewNonFatalError()) + }) + + t.Run("returns nil when all supplied errors are nil", func(t *testing.T) { + require.Nil(t, worker.NewNonFatalError(nil, nil)) + }) + + t.Run("drops nil entries and keeps non-nil ones", func(t *testing.T) { + a := errors.New("a") + b := errors.New("b") + te := worker.NewNonFatalError(nil, a, nil, b) + require.NotNil(t, te) + require.False(t, te.IsFatal()) + require.Nil(t, te.FatalError()) + require.Len(t, te.NonFatalErrors(), 2) + require.ErrorIs(t, te.NonFatalErrors()[0], a) + require.ErrorIs(t, te.NonFatalErrors()[1], b) + }) +} + +func TestTaskErrorUnwrap(t *testing.T) { + t.Run("exposes non-fatal and fatal errors via errors.Is", func(t *testing.T) { + nonFatal := errors.New("non-fatal") + fatal := errors.New("fatal") + te := worker.Join( + worker.NewNonFatalError(nonFatal), + worker.NewFatalError(fatal), + ) + require.NotNil(t, te) + require.ErrorIs(t, te, nonFatal) + require.ErrorIs(t, te, fatal) + }) +} + +func TestJoin(t *testing.T) { + t.Run("returns nil when both operands are nil", func(t *testing.T) { + require.Nil(t, worker.Join(nil, nil)) + }) + + t.Run("returns the non-nil operand when the other is nil", func(t *testing.T) { + err := errors.New("err") + te := worker.NewFatalError(err) + require.Equal(t, te, worker.Join(te, nil)) + require.Equal(t, te, worker.Join(nil, te)) + }) + + t.Run("concatenates non-fatal errors and joins fatals", func(t *testing.T) { + nonFatalA := errors.New("non-fatal A") + nonFatalB := errors.New("non-fatal B") + fatalA := errors.New("fatal A") + fatalB := errors.New("fatal B") + + a := worker.Join(worker.NewNonFatalError(nonFatalA), worker.NewFatalError(fatalA)) + b := worker.Join(worker.NewNonFatalError(nonFatalB), worker.NewFatalError(fatalB)) + + joined := worker.Join(a, b) + require.NotNil(t, joined) + require.True(t, joined.IsFatal()) + + require.Len(t, joined.NonFatalErrors(), 2) + require.ErrorIs(t, joined.NonFatalErrors()[0], nonFatalA) + require.ErrorIs(t, joined.NonFatalErrors()[1], nonFatalB) + + require.ErrorIs(t, joined.FatalError(), fatalA) + require.ErrorIs(t, joined.FatalError(), fatalB) + }) +} diff --git a/pkg/preparation/internal/worker/worker.go b/pkg/preparation/internal/worker/worker.go new file mode 100644 index 00000000..27a8866c --- /dev/null +++ b/pkg/preparation/internal/worker/worker.go @@ -0,0 +1,81 @@ +package worker + +import ( + "context" + "fmt" + + "github.com/storacha/guppy/internal/ctxutil" +) + +// Run executes a worker loop. The loop runs tasks in parallel. Run blocks until +// the `workAvailable` channel closes and all tasks are complete. +// +// The loop waits for a signal on the `workAvailable` channel, then calls +// `findWork` to get a batch of tasks to run and runs them with a maximum +// parallelism of `parallelism`, queuing any additional tasks. Any time the +// queue becomes empty, the loop waits for the next signal to find more work, +// until the `workAvailable` channel is closed. When the channel is closed, and +// all work is complete, the loop calls `finalize` and returns. +// +// If any task returns a [TaskError] whose [TaskError.IsFatal] is true, all +// running tasks are cancelled and no new tasks are started. +// +// Returns nil iff no task reported any errors, fatal or non-fatal. Otherwise, +// returns a [TaskError] describing what accumulated. +func Run( + ctx context.Context, + workAvailable <-chan struct{}, + parallelism int, + findWork func(ctx context.Context) ([]Task, error), + finalize func() error, +) TaskError { + g, gctx := WithContext(ctx) + g.SetLimit(parallelism) + + for { + select { + case <-ctx.Done(): + // External cancellation. The group has already been cancelled, so wait + // and return, but include the cancellation cause as a fatal error. + return Join(g.Wait(), NewFatalError(ctxutil.Cause(ctx))) + + case <-gctx.Done(): + // Internal cancellation. Wait, and return the result. + return g.Wait() + + case _, ok := <-workAvailable: + if !ok { + result := g.Wait() + if result != nil && result.IsFatal() { + return result + } + if finalize != nil { + if ferr := finalize(); ferr != nil { + return Join(result, NewFatalError(fmt.Errorf("worker finalize encountered an error: %w", ferr))) + } + } + return result + } + + tasks, err := findWork(ctx) + if err != nil { + // Cancel in-flight siblings and drain before surfacing the + // fatal. + result := g.Wait() + return Join(result, NewFatalError(fmt.Errorf("worker findWork encountered an error: %w", err))) + } + + for _, task := range tasks { + g.Go(func() TaskError { + select { + case <-gctx.Done(): + // If the context is already cancelled, skip starting the task. + return nil + default: + return task(gctx) + } + }) + } + } + } +} diff --git a/pkg/preparation/internal/worker/worker_test.go b/pkg/preparation/internal/worker/worker_test.go new file mode 100644 index 00000000..4a2c8f19 --- /dev/null +++ b/pkg/preparation/internal/worker/worker_test.go @@ -0,0 +1,200 @@ +package worker_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/storacha/guppy/pkg/preparation/internal/worker" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// e is a shorthand helper function that uses [require.EventuallyWithT] with +// a standard timeout and interval, to keep noise out of the tests. +func e(t *testing.T, condition func(collect *assert.CollectT)) { + t.Helper() + require.EventuallyWithT(t, condition, time.Second, 10*time.Millisecond) +} + +// fatalTask wraps a fn returning error into a Task that reports any error as +// fatal. +func fatalTask(fn func() error) worker.Task { + return func(ctx context.Context) worker.TaskError { + return worker.NewFatalError(fn()) + } +} + +func TestWorker(t *testing.T) { + t.Run("runs the work function for every signal received, then the finalize function when the channel closes", func(t *testing.T) { + signalChan := make(chan struct{}, 1) + resultChan := make(chan worker.TaskError, 1) + var runs int + var finalizes int + + go func() { + defer close(resultChan) + resultChan <- worker.Run(t.Context(), signalChan, 1, + func(ctx context.Context) ([]worker.Task, error) { + runs++ + return nil, nil + }, + func() error { + finalizes++ + return nil + }, + ) + }() + + require.Equal(t, 0, runs, "worker should not run before signal") + signalChan <- struct{}{} + e(t, func(t *assert.CollectT) { require.Equal(t, 1, runs, "worker should run once after signal") }) + signalChan <- struct{}{} + e(t, func(t *assert.CollectT) { require.Equal(t, 2, runs, "worker should run again after second signal") }) + + require.Equal(t, 0, finalizes, "finalize function should not be called until the channel closes") + close(signalChan) + e(t, func(t *assert.CollectT) { + require.Equal(t, 1, finalizes, "finalize function should be called once the channel closes") + }) + + res := <-resultChan + require.Nil(t, res, "result should be nil after a clean run") + }) + + t.Run("immediately responds with any work error, skipping the finalizer", func(t *testing.T) { + workerErr := errors.New("error in doWork") + signalChan := make(chan struct{}, 3) + resultChan := make(chan worker.TaskError, 1) + var runs int + var finalizes int + + go func() { + defer close(resultChan) + resultChan <- worker.Run(t.Context(), signalChan, 1, + func(ctx context.Context) ([]worker.Task, error) { + runs++ + if runs == 2 { + return []worker.Task{fatalTask(func() error { return workerErr })}, nil + } + return nil, nil + }, + func() error { + finalizes++ + return nil + }, + ) + }() + + // Send three signals; the second should cause an error, the third should not run + signalChan <- struct{}{} + signalChan <- struct{}{} + signalChan <- struct{}{} + + res := <-resultChan + require.NotNil(t, res) + require.True(t, res.IsFatal(), "a fatal task error should make the result fatal") + require.ErrorIs(t, res.FatalError(), workerErr) + require.LessOrEqual(t, runs, 3, "worker should have stopped after encountering an error") + require.Equal(t, 0, finalizes, "finalize function should not have been called") + }) + + t.Run("collects non-fatal errors and continues dispatching", func(t *testing.T) { + nonFatalErr := errors.New("non-fatal failure") + signalChan := make(chan struct{}, 3) + resultChan := make(chan worker.TaskError, 1) + var runs int + var finalizes int + + go func() { + defer close(resultChan) + resultChan <- worker.Run(t.Context(), signalChan, 1, + func(ctx context.Context) ([]worker.Task, error) { + runs++ + return []worker.Task{ + func(ctx context.Context) worker.TaskError { + return worker.NewNonFatalError(nonFatalErr) + }, + }, nil + }, + func() error { + finalizes++ + return nil + }, + ) + }() + + signalChan <- struct{}{} + signalChan <- struct{}{} + signalChan <- struct{}{} + close(signalChan) + + res := <-resultChan + require.NotNil(t, res) + require.False(t, res.IsFatal(), "non-fatal errors should not make the result fatal") + require.Len(t, res.NonFatalErrors(), 3, "every non-fatal error should be collected") + for _, err := range res.NonFatalErrors() { + require.ErrorIs(t, err, nonFatalErr) + } + require.Equal(t, 3, runs, "worker should have kept dispatching after non-fatal errors") + require.Equal(t, 1, finalizes, "finalize should still run when only non-fatal errors occurred") + }) + + t.Run("responds with any finalize error", func(t *testing.T) { + finalizerErr := errors.New("error in finalize") + signalChan := make(chan struct{}, 3) + resultChan := make(chan worker.TaskError, 1) + var runs int + + go func() { + defer close(resultChan) + resultChan <- worker.Run(t.Context(), signalChan, 1, + func(ctx context.Context) ([]worker.Task, error) { + runs++ + return nil, nil + }, + func() error { + return finalizerErr + }, + ) + }() + + // Send three signals; all should run + signalChan <- struct{}{} + signalChan <- struct{}{} + signalChan <- struct{}{} + close(signalChan) + + res := <-resultChan + require.NotNil(t, res) + require.True(t, res.IsFatal(), "finalize error should make the result fatal") + require.ErrorContains(t, res.FatalError(), "worker finalize encountered an error: error in finalize") + require.ErrorIs(t, res.FatalError(), finalizerErr) + require.Equal(t, 3, runs, "worker should have run all three times") + }) + + t.Run("ignores a nil finalizer", func(t *testing.T) { + signalChan := make(chan struct{}, 1) + resultChan := make(chan worker.TaskError, 1) + var ran bool + + go func() { + defer close(resultChan) + resultChan <- worker.Run(t.Context(), signalChan, 1, + func(ctx context.Context) ([]worker.Task, error) { + ran = true + return nil, nil + }, + nil, + ) + }() + + require.False(t, ran, "worker should not run before signal") + signalChan <- struct{}{} + e(t, func(t *assert.CollectT) { require.True(t, ran, "worker should run after signal") }) + close(signalChan) + res := <-resultChan + require.Nil(t, res, "result should be nil after a clean run with no finalizer") + }) +} diff --git a/pkg/preparation/preparation.go b/pkg/preparation/preparation.go index 3341e37c..801f61b3 100644 --- a/pkg/preparation/preparation.go +++ b/pkg/preparation/preparation.go @@ -164,34 +164,34 @@ func NewAPI(repo Repo, client StorachaClient, options ...Option) API { blobAddOptions = append(blobAddOptions, clientpkg.WithPutClient(cfg.putHTTPClient)) } storachaAPI := storacha.API{ - Repo: repo, - Client: client, - ReaderForShard: blobsAPI.ReaderForShard, - ReaderForIndex: blobsAPI.ReaderForIndex, - BlobUploadParallelism: cfg.blobUploadParallelism, - Bus: cfg.bus, - Replicas: cfg.replicas, - BlobAddOptions: blobAddOptions, + Repo: repo, + Client: client, + ReaderForShard: blobsAPI.ReaderForShard, + ReaderForIndex: blobsAPI.ReaderForIndex, + Bus: cfg.bus, + Replicas: cfg.replicas, + BlobAddOptions: blobAddOptions, } uploadsAPI = uploads.API{ - Repo: repo, - AssumeUnchangedSources: cfg.assumeUnchangedSources, - ExecuteScan: scansAPI.ExecuteScan, - ExecuteDagScansForUpload: dagsAPI.ExecuteDagScansForUpload, - AddNodesToUploadShards: blobsAPI.AddNodesToUploadShards, - AddShardsToUploadIndexes: blobsAPI.AddShardsToUploadIndexes, - CloseUploadShards: blobsAPI.CloseUploadShards, - CloseUploadIndexes: blobsAPI.CloseUploadIndexes, - AddShardsForUpload: storachaAPI.AddShardsForUpload, - PostProcessUploadedShards: storachaAPI.PostProcessUploadedShards, - PostProcessUploadedIndexes: storachaAPI.PostProcessUploadedIndexes, - AddIndexesForUpload: storachaAPI.AddIndexesForUpload, - AddStorachaUploadForUpload: storachaAPI.AddStorachaUploadForUpload, - RemoveBadFSEntry: scansAPI.RemoveBadFSEntry, - RemoveBadNodes: dagsAPI.RemoveBadNodes, - RemoveShard: blobsAPI.RemoveShard, - Publisher: cfg.bus, + Repo: repo, + AssumeUnchangedSources: cfg.assumeUnchangedSources, + ExecuteScan: scansAPI.ExecuteScan, + ExecuteDagScansForUpload: dagsAPI.ExecuteDagScansForUpload, + AddNodesToUploadShards: blobsAPI.AddNodesToUploadShards, + AddShardsToUploadIndexes: blobsAPI.AddShardsToUploadIndexes, + CloseUploadShards: blobsAPI.CloseUploadShards, + CloseUploadIndexes: blobsAPI.CloseUploadIndexes, + FindShardAddTasksForUpload: storachaAPI.FindShardAddTasksForUpload, + FindIndexAddTasksForUpload: storachaAPI.FindIndexAddTasksForUpload, + BlobUploadParallelism: cfg.blobUploadParallelism, + FindShardPostProcessTasksForUpload: storachaAPI.FindShardPostProcessTasksForUpload, + FindIndexPostProcessTasksForUpload: storachaAPI.FindIndexPostProcessTasksForUpload, + AddStorachaUploadForUpload: storachaAPI.AddStorachaUploadForUpload, + RemoveBadFSEntry: scansAPI.RemoveBadFSEntry, + RemoveBadNodes: dagsAPI.RemoveBadNodes, + RemoveShard: blobsAPI.RemoveShard, + Publisher: cfg.bus, } return API{ diff --git a/pkg/preparation/preparation_test.go b/pkg/preparation/preparation_test.go index 100c8a2c..2e1d1be0 100644 --- a/pkg/preparation/preparation_test.go +++ b/pkg/preparation/preparation_test.go @@ -405,7 +405,7 @@ func TestExecuteUpload(t *testing.T) { // We don't know exactly how many successful PUTs there were, but we know it // should be at least 2 and at most 6. require.GreaterOrEqual(t, putBlobs.Size(), 2, "expected at least 2/5 shards to be added so far") - require.Less(t, putBlobs.Size(), 6, "expected at most 4/5 shards + 1 index to be added so far") + require.LessOrEqual(t, putBlobs.Size(), 6, "expected at most 5/5 shards + 1 index to be added so far") require.Len(t, uploadAddCaps, 0, "expected `upload/add` not to have been called yet") t.Log("Retrying the upload after error...") diff --git a/pkg/preparation/storacha/storacha.go b/pkg/preparation/storacha/storacha.go index b51806b1..e1aa4d9b 100644 --- a/pkg/preparation/storacha/storacha.go +++ b/pkg/preparation/storacha/storacha.go @@ -15,25 +15,24 @@ import ( "github.com/multiformats/go-multicodec" filecoincap "github.com/storacha/go-libstoracha/capabilities/filecoin" spaceblobcap "github.com/storacha/go-libstoracha/capabilities/space/blob" - "github.com/storacha/go-libstoracha/capabilities/types" + captypes "github.com/storacha/go-libstoracha/capabilities/types" "github.com/storacha/go-libstoracha/capabilities/upload" "github.com/storacha/go-ucanto/core/delegation" "github.com/storacha/go-ucanto/core/receipt/fx" "github.com/storacha/go-ucanto/did" - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/trace" - "golang.org/x/sync/errgroup" - "github.com/storacha/guppy/pkg/bus" "github.com/storacha/guppy/pkg/bus/events" "github.com/storacha/guppy/pkg/client" "github.com/storacha/guppy/pkg/internal/util" "github.com/storacha/guppy/pkg/preparation/blobs/model" "github.com/storacha/guppy/pkg/preparation/internal/meteredwriter" - gtypes "github.com/storacha/guppy/pkg/preparation/types" + "github.com/storacha/guppy/pkg/preparation/internal/worker" + "github.com/storacha/guppy/pkg/preparation/types" "github.com/storacha/guppy/pkg/preparation/types/id" "github.com/storacha/guppy/pkg/preparation/uploads" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" ) var ( @@ -48,7 +47,7 @@ type Client interface { SpaceIndexAdd(ctx context.Context, indexCID cid.Cid, indexSize uint64, rootCID cid.Cid, space did.DID) error FilecoinOffer(ctx context.Context, space did.DID, content ipld.Link, piece ipld.Link, opts ...client.FilecoinOfferOption) (filecoincap.OfferOk, error) UploadAdd(ctx context.Context, space did.DID, root ipld.Link, shards []ipld.Link) (upload.AddOk, error) - SpaceBlobReplicate(ctx context.Context, space did.DID, blob types.Blob, replicaCount uint, locationCommitment delegation.Delegation) (spaceblobcap.ReplicateOk, fx.Effects, error) + SpaceBlobReplicate(ctx context.Context, space did.DID, blob captypes.Blob, replicaCount uint, locationCommitment delegation.Delegation) (spaceblobcap.ReplicateOk, fx.Effects, error) } var _ Client = (*client.Client)(nil) @@ -58,164 +57,123 @@ type ReaderForIndexFunc func(ctx context.Context, indexID id.IndexID) (io.ReadCl // API provides methods to interact with Storacha. type API struct { - Repo Repo - Client Client - ReaderForShard ReaderForShardFunc - ReaderForIndex ReaderForIndexFunc + Repo Repo + Client Client + ReaderForShard ReaderForShardFunc + ReaderForIndex ReaderForIndexFunc + + // TK: Rm BlobUploadParallelism int Bus bus.Publisher Replicas uint BlobAddOptions []client.SpaceBlobAddOption } -var _ uploads.AddShardsForUploadFunc = API{}.AddShardsForUpload -var _ uploads.AddIndexesForUploadFunc = API{}.AddIndexesForUpload +var _ uploads.FindShardAddTasksForUploadFunc = API{}.FindShardAddTasksForUpload +var _ uploads.FindIndexAddTasksForUploadFunc = API{}.FindIndexAddTasksForUpload +var _ uploads.FindShardPostProcessTasksForUploadFunc = API{}.FindShardPostProcessTasksForUpload +var _ uploads.FindIndexPostProcessTasksForUploadFunc = API{}.FindIndexPostProcessTasksForUpload var _ uploads.AddStorachaUploadForUploadFunc = API{}.AddStorachaUploadForUpload -func (a API) AddShardsForUpload(ctx context.Context, uploadID id.UploadID, spaceDID did.DID, shardUploadedCb func(shard *model.Shard) error) error { - ctx, span := tracer.Start(ctx, "add-shards-for-upload") +func (a API) FindShardAddTasksForUpload(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]types.IDTask, error) { + ctx, span := tracer.Start(ctx, "find-shard-add-tasks-for-upload") defer span.End() + closedShards, err := a.Repo.ShardsForUploadByState(ctx, uploadID, model.BlobStateClosed) if err != nil { - return fmt.Errorf("failed to get closed shards for upload %s: %w", uploadID, err) + return nil, fmt.Errorf("failed to get closed shards for upload %s: %w", uploadID, err) } span.AddEvent("found closed shards", trace.WithAttributes(attribute.Int("shards", len(closedShards)))) - blobs := make([]model.Blob, len(closedShards)) - for i, shard := range closedShards { - blobs[i] = shard + tasks := make([]types.IDTask, 0, len(closedShards)) + for _, shard := range closedShards { + tasks = append(tasks, types.IDTask{ + ID: shard.ID(), + Run: func(ctx context.Context) worker.TaskError { + if err := a.addBlob(ctx, shard, spaceDID); err != nil { + err = fmt.Errorf("failed to add shard %s: %w", shard, err) + // [types.BlobUploadError]s are non-fatal. + var errBlobUpload types.BlobUploadError + if errors.As(err, &errBlobUpload) { + return worker.NewNonFatalError(err) + } + log.Errorf("%v", err) + return worker.NewFatalError(err) + } + return nil + }, + }) } - return a.addBlobs(ctx, blobs, spaceDID, func(blob model.Blob) error { - if shardUploadedCb != nil { - return shardUploadedCb(blob.(*model.Shard)) - } - return nil - }) + return tasks, nil } -func (a API) PostProcessUploadedShards(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) error { - ctx, span := tracer.Start(ctx, "post-process-uploaded-shards") +func (a API) FindIndexAddTasksForUpload(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]types.IDTask, error) { + ctx, span := tracer.Start(ctx, "find-index-add-tasks-for-upload") defer span.End() - uploadedShards, err := a.Repo.ShardsForUploadByState(ctx, uploadID, model.BlobStateUploaded) - if err != nil { - return fmt.Errorf("failed to get uploaded shards for post processing %s: %w", uploadID, err) - } - span.AddEvent("found uploaded shards", trace.WithAttributes(attribute.Int("shards", len(uploadedShards)))) - blobs := make([]model.Blob, len(uploadedShards)) - for i, shard := range uploadedShards { - blobs[i] = shard - } - return a.postProcessBlobs(ctx, blobs, spaceDID, func(blob model.Blob) error { - var opts []client.FilecoinOfferOption - if blob.PDPAccept() != nil { - opts = append(opts, client.WithPDPAcceptInvocation(blob.PDPAccept())) - } - if err := a.filecoinOffer(ctx, blob, spaceDID, opts...); err != nil { - return gtypes.NewBlobUploadError(blob.ID(), err) - } - return nil - }) -} - -// addBlobs adds the given blobs to the space, in parallel. For each blob, it -// will `space/blob/add` if it hasn't been added yet, then call the `afterUploaded` callback if successful. -// `SpaceBlobAdded()` will be called after `space/blob/add`. `Added()` will be -// called at the very end. If any of these steps fail, an error will be -// returned. -func (a API) addBlobs(ctx context.Context, blobs []model.Blob, spaceDID did.DID, afterUploaded func(blob model.Blob) error) error { - // Ensure at least 1 parallelism - if a.BlobUploadParallelism < 1 { - a.BlobUploadParallelism = 1 + closedIndexes, err := a.Repo.IndexesForUploadByState(ctx, uploadID, model.BlobStateClosed) + if err != nil { + return nil, fmt.Errorf("failed to get closed indexes for upload %s: %w", uploadID, err) } + span.AddEvent("found closed indexes", trace.WithAttributes(attribute.Int("indexes", len(closedIndexes)))) - sem := make(chan struct{}, a.BlobUploadParallelism) - blobUploadErrorCh := make(chan gtypes.BlobUploadError, len(blobs)) - eg, gctx := errgroup.WithContext(ctx) - for _, blob := range blobs { - sem <- struct{}{} - eg.Go(func() error { - defer func() { <-sem }() - if err := a.addBlob(gctx, blob, spaceDID); err != nil { - err = fmt.Errorf("failed to add blob %s: %w", blob, err) - var errBlobUpload gtypes.BlobUploadError - if errors.As(err, &errBlobUpload) { - blobUploadErrorCh <- errBlobUpload - return nil - } - log.Errorf("%v", err) - return err - } - if afterUploaded != nil { - if err := afterUploaded(blob); err != nil { - return fmt.Errorf("failed to call after uploaded callback for blob %s: %w", blob.ID(), err) + tasks := make([]types.IDTask, 0, len(closedIndexes)) + for _, index := range closedIndexes { + tasks = append(tasks, types.IDTask{ + ID: index.ID(), + Run: func(ctx context.Context) worker.TaskError { + if err := a.addBlob(ctx, index, spaceDID); err != nil { + err = fmt.Errorf("failed to add index %s: %w", index, err) + // [types.BlobUploadError]s are non-fatal. + var errBlobUpload types.BlobUploadError + if errors.As(err, &errBlobUpload) { + return worker.NewNonFatalError(err) + } + log.Errorf("%v", err) + return worker.NewFatalError(err) } - } - log.Infof("Successfully added blob %s", blob.ID()) - return nil + return nil + }, }) } - - terminalErr := eg.Wait() - close(blobUploadErrorCh) - - if terminalErr != nil { - return terminalErr - } - - var blobUploadErrors []gtypes.BlobUploadError - for err := range blobUploadErrorCh { - blobUploadErrors = append(blobUploadErrors, err) - } - if len(blobUploadErrors) > 0 { - return gtypes.NewBlobUploadErrors(blobUploadErrors) - } - return nil + return tasks, nil } -func (a API) postProcessBlobs(ctx context.Context, blobs []model.Blob, spaceDID did.DID, afterAdded func(blob model.Blob) error) error { - // Ensure at least 1 parallelism - if a.BlobUploadParallelism < 1 { - a.BlobUploadParallelism = 1 +func (a API) FindShardPostProcessTasksForUpload(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]types.IDTask, error) { + ctx, span := tracer.Start(ctx, "find-shard-post-process-tasks-for-upload") + defer span.End() + + uploadedShards, err := a.Repo.ShardsForUploadByState(ctx, uploadID, model.BlobStateUploaded) + if err != nil { + return nil, fmt.Errorf("failed to get uploaded shards for upload %s: %w", uploadID, err) } + span.AddEvent("found uploaded shards", trace.WithAttributes(attribute.Int("shards", len(uploadedShards)))) - sem := make(chan struct{}, a.BlobUploadParallelism) - blobUploadErrorCh := make(chan gtypes.BlobUploadError, len(blobs)) - eg, gctx := errgroup.WithContext(ctx) - for _, blob := range blobs { - sem <- struct{}{} - eg.Go(func() error { - defer func() { <-sem }() - if err := a.postProcessBlob(gctx, blob, spaceDID, afterAdded); err != nil { - err = fmt.Errorf("failed to add blob %s: %w", blob, err) - var errBlobUpload gtypes.BlobUploadError - if errors.As(err, &errBlobUpload) { - blobUploadErrorCh <- errBlobUpload + tasks := make([]types.IDTask, 0, len(uploadedShards)) + for _, shard := range uploadedShards { + tasks = append(tasks, types.IDTask{ + ID: shard.ID(), + Run: func(ctx context.Context) worker.TaskError { + err := a.postProcessBlob(ctx, shard, spaceDID, func(blob model.Blob) error { + var opts []client.FilecoinOfferOption + if blob.PDPAccept() != nil { + opts = append(opts, client.WithPDPAcceptInvocation(blob.PDPAccept())) + } + if err := a.filecoinOffer(ctx, blob, spaceDID, opts...); err != nil { + return fmt.Errorf("failed to `filecoin/offer` shard %s: %w", blob, err) + } return nil + }) + if err != nil { + log.Errorf("failed to post-process shard %s: %v", shard, err) + return worker.NewFatalError(fmt.Errorf("failed to post-process shard %s: %w", shard, err)) } - log.Errorf("%v", err) - return err - } - log.Infof("Successfully post-processed blob %s", blob.ID()) - return nil + log.Infof("Successfully post-processed shard %s", shard.ID()) + return nil + }, }) } - - terminalErr := eg.Wait() - close(blobUploadErrorCh) - - if terminalErr != nil { - return terminalErr - } - - var blobUploadErrors []gtypes.BlobUploadError - for err := range blobUploadErrorCh { - blobUploadErrors = append(blobUploadErrors, err) - } - if len(blobUploadErrors) > 0 { - return gtypes.NewBlobUploadErrors(blobUploadErrors) - } - return nil + return tasks, nil } func (a API) readerForBlob(ctx context.Context, blob model.Blob) (io.ReadCloser, error) { @@ -289,7 +247,7 @@ func (a API) addBlob(ctx context.Context, blob model.Blob, spaceDID did.DID) err })) addedBlob, err := a.spaceBlobAdd(ctx, addReader, spaceDID, opts...) if err != nil { - return gtypes.NewBlobUploadError(blob.ID(), fmt.Errorf("failed to add blob %s to space %s: %w", blob, spaceDID, err)) + return types.NewBlobUploadError(blob.ID(), fmt.Errorf("failed to add blob %s to space %s: %w", blob, spaceDID, err)) } if err := blob.SpaceBlobAdded(addedBlob); err != nil { @@ -306,7 +264,7 @@ func (a API) addBlob(ctx context.Context, blob model.Blob, spaceDID did.DID) err } } - if err := a.updateBlob(ctx, blob); err != nil { + if err := a.updateBlob(context.WithoutCancel(ctx), blob); err != nil { return fmt.Errorf("failed to update blob %s after `space/blob/add`: %w", blob, err) } return nil @@ -314,7 +272,7 @@ func (a API) addBlob(ctx context.Context, blob model.Blob, spaceDID did.DID) err func (a API) postProcessBlob(ctx context.Context, blob model.Blob, spaceDID did.DID, afterAdded func(blob model.Blob) error) error { if err := a.spaceBlobReplicate(ctx, blob, spaceDID, blob.Location()); err != nil { - return gtypes.NewBlobUploadError(blob.ID(), fmt.Errorf("failed to replicate blob %s: %w", blob, err)) + return types.NewBlobUploadError(blob.ID(), fmt.Errorf("failed to replicate blob %s: %w", blob, err)) } if afterAdded != nil { @@ -326,7 +284,7 @@ func (a API) postProcessBlob(ctx context.Context, blob model.Blob, spaceDID did. if err := blob.Added(); err != nil { return fmt.Errorf("failed to mark blob %s as added: %w", blob, err) } - if err := a.updateBlob(ctx, blob); err != nil { + if err := a.updateBlob(context.WithoutCancel(ctx), blob); err != nil { return fmt.Errorf("failed to update blob %s after adding to space: %w", blob, err) } @@ -353,7 +311,7 @@ func (a API) spaceBlobReplicate(ctx context.Context, blob model.Blob, spaceDID d _, _, err := a.Client.SpaceBlobReplicate( ctx, spaceDID, - types.Blob{ + captypes.Blob{ Digest: blob.Digest(), Size: blob.Size(), }, @@ -371,8 +329,8 @@ func (a API) filecoinOffer(ctx context.Context, blob model.Blob, spaceDID did.DI switch { case blob.Size() == 0: return fmt.Errorf("blob %s has no set size yet", blob) - case blob.Size() < gtypes.MinPiecePayload: - log.Warnf("skipping `filecoin/offer` for blob %s: size %d is below minimum %d", blob, blob.Size(), gtypes.MinPiecePayload) + case blob.Size() < types.MinPiecePayload: + log.Warnf("skipping `filecoin/offer` for blob %s: size %d is below minimum %d", blob, blob.Size(), types.MinPiecePayload) return nil case blob.Size() > commp.MaxPiecePayload: log.Warnf("skipping `filecoin/offer` for blob %s: size %d is above maximum %d", blob, blob.Size(), commp.MaxPiecePayload) @@ -391,52 +349,37 @@ func (a API) filecoinOffer(ctx context.Context, blob model.Blob, spaceDID did.DI return nil } -// AddIndexesForUpload adds the given indexes to the space, in parallel. The -// upload must have a root CID set. -func (a API) AddIndexesForUpload(ctx context.Context, uploadID id.UploadID, spaceDID did.DID, indexCB func(index *model.Index) error) error { - ctx, span := tracer.Start(ctx, "add-indexes-for-upload") - defer span.End() - - closedIndexes, err := a.Repo.IndexesForUploadByState(ctx, uploadID, model.BlobStateClosed) - if err != nil { - return fmt.Errorf("failed to get closed indexes for upload %s: %w", uploadID, err) - } - span.AddEvent("found closed indexes", trace.WithAttributes(attribute.Int("indexes", len(closedIndexes)))) - - blobs := make([]model.Blob, len(closedIndexes)) - for i, shard := range closedIndexes { - blobs[i] = shard - } - return a.addBlobs(ctx, blobs, spaceDID, func(blob model.Blob) error { - if indexCB != nil { - return indexCB(blob.(*model.Index)) - } - return nil - }) -} - -// PostProcessUploadedIndexes runs post-processing for uploaded indexes, including -// adding them to the space via `space/index/add`. -func (a API) PostProcessUploadedIndexes(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) error { - ctx, span := tracer.Start(ctx, "post-process-uploaded-indexes") +func (a API) FindIndexPostProcessTasksForUpload(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]types.IDTask, error) { + ctx, span := tracer.Start(ctx, "find-index-post-process-tasks-for-upload") defer span.End() uploadedIndexes, err := a.Repo.IndexesForUploadByState(ctx, uploadID, model.BlobStateUploaded) if err != nil { - return fmt.Errorf("failed to get uploaded indexes for upload %s: %w", uploadID, err) + return nil, fmt.Errorf("failed to get uploaded indexes for upload %s: %w", uploadID, err) } span.AddEvent("found uploaded indexes", trace.WithAttributes(attribute.Int("indexes", len(uploadedIndexes)))) - blobs := make([]model.Blob, len(uploadedIndexes)) - for i, shard := range uploadedIndexes { - blobs[i] = shard + tasks := make([]types.IDTask, 0, len(uploadedIndexes)) + for _, index := range uploadedIndexes { + tasks = append(tasks, types.IDTask{ + ID: index.ID(), + Run: func(ctx context.Context) worker.TaskError { + err := a.postProcessBlob(ctx, index, spaceDID, func(blob model.Blob) error { + // Use a placeholder for the root because it doesn't matter what it is, + // and we don't want to wait for it to be known. It shouldn't really be + // something the index knows at all. + return a.Client.SpaceIndexAdd(ctx, blob.CID(), blob.Size(), util.PlaceholderCID, spaceDID) + }) + if err != nil { + log.Errorf("failed to post-process index %s: %v", index, err) + return worker.NewFatalError(fmt.Errorf("failed to post-process index %s: %w", index, err)) + } + log.Infof("Successfully post-processed index %s", index.ID()) + return nil + }, + }) } - return a.postProcessBlobs(ctx, blobs, spaceDID, func(blob model.Blob) error { - // Use a placeholder for the root because it doesn't matter what it is, - // and we don't want to wait for it to be known. It shouldn't really be - // something the index knows at all. - return a.Client.SpaceIndexAdd(ctx, blob.CID(), blob.Size(), util.PlaceholderCID, spaceDID) - }) + return tasks, nil } func (a API) AddStorachaUploadForUpload(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) error { diff --git a/pkg/preparation/storacha/storacha_test.go b/pkg/preparation/storacha/storacha_test.go index ee5bb431..9e8c7ed6 100644 --- a/pkg/preparation/storacha/storacha_test.go +++ b/pkg/preparation/storacha/storacha_test.go @@ -35,7 +35,7 @@ import ( // padding to every "CAR" to make sure it's definitely long enough. var padding = bytes.Repeat([]byte{0}, 127) -func TestAddShardsForUpload(t *testing.T) { +func TestFindShardAddTasksForUpload(t *testing.T) { t.Run("`space/blob/add`s, `space/blob/replicate`s, and `filecoin/offer`s a CAR for each shard", func(t *testing.T) { db := testdb.CreateTestDB(t) repo := stestutil.Must(sqlrepo.New(db))(t) @@ -53,11 +53,10 @@ func TestAddShardsForUpload(t *testing.T) { } api := storacha.API{ - Repo: repo, - Client: &client, - ReaderForShard: carForShard, - BlobUploadParallelism: 1, - Replicas: 3, + Repo: repo, + Client: &client, + ReaderForShard: carForShard, + Replicas: 3, } blobsApi := blobs.API{ @@ -83,8 +82,11 @@ func TestAddShardsForUpload(t *testing.T) { secondShard := shards[0] // Upload shards that are ready to go. - err = api.AddShardsForUpload(t.Context(), upload.ID(), spaceDID, nil) + tasks, err := api.FindShardAddTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range tasks { + require.Nil(t, task.Run(t.Context())) + } // Reload shards firstShard, err = repo.GetShardByID(t.Context(), firstShard.ID()) @@ -101,8 +103,11 @@ func TestAddShardsForUpload(t *testing.T) { require.Equal(t, spaceDID, client.SpaceBlobAddInvocations[0].Space) // Now run post processing. - err = api.PostProcessUploadedShards(t.Context(), upload.ID(), spaceDID) + ppTasks, err := api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range ppTasks { + require.Nil(t, task.Run(t.Context())) + } // Reload shards firstShard, err = repo.GetShardByID(t.Context(), firstShard.ID()) require.NoError(t, err) @@ -131,8 +136,11 @@ func TestAddShardsForUpload(t *testing.T) { // Now close the upload shards and run it again. err = blobsApi.CloseUploadShards(t.Context(), upload.ID(), nil) require.NoError(t, err) - err = api.AddShardsForUpload(t.Context(), upload.ID(), spaceDID, nil) + tasks, err = api.FindShardAddTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range tasks { + require.Nil(t, task.Run(t.Context())) + } // Reload second shard secondShard, err = repo.GetShardByID(t.Context(), secondShard.ID()) @@ -145,8 +153,11 @@ func TestAddShardsForUpload(t *testing.T) { require.Equal(t, spaceDID, client.SpaceBlobAddInvocations[1].Space) // Now run post processing. - err = api.PostProcessUploadedShards(t.Context(), upload.ID(), spaceDID) + ppTasks, err = api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range ppTasks { + require.Nil(t, task.Run(t.Context())) + } // Reload second shard secondShard, err = repo.GetShardByID(t.Context(), secondShard.ID()) @@ -182,11 +193,10 @@ func TestAddShardsForUpload(t *testing.T) { } api := storacha.API{ - Repo: repo, - Client: &client, - ReaderForShard: carForShard, - BlobUploadParallelism: 1, - Replicas: 3, + Repo: repo, + Client: &client, + ReaderForShard: carForShard, + Replicas: 3, } blobsApi := blobs.API{ @@ -203,10 +213,20 @@ func TestAddShardsForUpload(t *testing.T) { client.SpaceBlobAddError = fmt.Errorf("simulated SpaceBlobAdd error") - err = api.AddShardsForUpload(t.Context(), upload.ID(), spaceDID, nil) - require.ErrorContains(t, err, "simulated SpaceBlobAdd error") - err = api.PostProcessUploadedShards(t.Context(), upload.ID(), spaceDID) + tasks, err := api.FindShardAddTasksForUpload(t.Context(), upload.ID(), spaceDID) + require.NoError(t, err) + for _, task := range tasks { + te := task.Run(t.Context()) + require.NotNil(t, te) + require.False(t, te.IsFatal(), "BlobUploadError should be non-fatal") + require.Len(t, te.NonFatalErrors(), 1) + require.ErrorContains(t, te.NonFatalErrors()[0], "simulated SpaceBlobAdd error") + } + ppTasks, err := api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range ppTasks { + require.Nil(t, task.Run(t.Context())) + } // It should have `space/blob/add`ed (and failed)... require.Len(t, client.SpaceBlobAddInvocations, 1) @@ -216,7 +236,7 @@ func TestAddShardsForUpload(t *testing.T) { require.Len(t, client.FilecoinOfferInvocations, 0) // It should have closed the first shard's reader. - require.Len(t, shardReadersClosed, 1, "expected shard readerto be closed, even though it failed") + require.Len(t, shardReadersClosed, 1, "expected shard reader to be closed, even though it failed") // reset the shard readers closed map for shardID := range shardReadersClosed { delete(shardReadersClosed, shardID) @@ -225,10 +245,19 @@ func TestAddShardsForUpload(t *testing.T) { // Now retry: `space/blob/add` succeeds but `space/blob/replicate` fails. client.SpaceBlobAddError = nil client.SpaceBlobReplicateError = fmt.Errorf("simulated SpaceBlobReplicate error") - err = api.AddShardsForUpload(t.Context(), upload.ID(), spaceDID, nil) + tasks, err = api.FindShardAddTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) - err = api.PostProcessUploadedShards(t.Context(), upload.ID(), spaceDID) - require.ErrorContains(t, err, "simulated SpaceBlobReplicate error") + for _, task := range tasks { + require.Nil(t, task.Run(t.Context())) + } + ppTasks, err = api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) + require.NoError(t, err) + for _, task := range ppTasks { + te := task.Run(t.Context()) + require.NotNil(t, te) + require.True(t, te.IsFatal(), "post-process error should be fatal") + require.ErrorContains(t, te.FatalError(), "simulated SpaceBlobReplicate error") + } // It should have `space/blob/add`ed again... require.Len(t, client.SpaceBlobAddInvocations, 2) @@ -243,10 +272,19 @@ func TestAddShardsForUpload(t *testing.T) { // Now retry: `space/blob/replicate` succeeds but `filecoin/offer` fails. client.SpaceBlobReplicateError = nil client.FilecoinOfferError = fmt.Errorf("simulated FilecoinOffer error") - err = api.AddShardsForUpload(t.Context(), upload.ID(), spaceDID, nil) + tasks, err = api.FindShardAddTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) - err = api.PostProcessUploadedShards(t.Context(), upload.ID(), spaceDID) - require.ErrorContains(t, err, "simulated FilecoinOffer error") + for _, task := range tasks { + require.Nil(t, task.Run(t.Context())) + } + ppTasks, err = api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) + require.NoError(t, err) + for _, task := range ppTasks { + te := task.Run(t.Context()) + require.NotNil(t, te) + require.True(t, te.IsFatal(), "post-process error should be fatal") + require.ErrorContains(t, te.FatalError(), "simulated FilecoinOffer error") + } // It should NOT `space/blob/add` again... require.Len(t, client.SpaceBlobAddInvocations, 2) @@ -274,11 +312,10 @@ func TestAddShardsForUpload(t *testing.T) { } api := storacha.API{ - Repo: repo, - Client: &client, - ReaderForShard: carForShard, - BlobUploadParallelism: 1, - Replicas: 3, + Repo: repo, + Client: &client, + ReaderForShard: carForShard, + Replicas: 3, } blobsApi := blobs.API{ @@ -293,10 +330,16 @@ func TestAddShardsForUpload(t *testing.T) { err = blobsApi.CloseUploadShards(t.Context(), upload.ID(), nil) require.NoError(t, err) - err = api.AddShardsForUpload(t.Context(), upload.ID(), spaceDID, nil) + tasks, err := api.FindShardAddTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) - err = api.PostProcessUploadedShards(t.Context(), upload.ID(), spaceDID) + for _, task := range tasks { + require.Nil(t, task.Run(t.Context())) + } + ppTasks, err := api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range ppTasks { + require.Nil(t, task.Run(t.Context())) + } // It should `space/blob/add`... require.Len(t, client.SpaceBlobAddInvocations, 1) @@ -309,7 +352,7 @@ func TestAddShardsForUpload(t *testing.T) { }) } -func TestAddIndexesForUpload(t *testing.T) { +func TestFindIndexAddTasksForUpload(t *testing.T) { t.Run("`space/blob/add`s and `space/blob/replicate`s index CARs", func(t *testing.T) { logging.SetLogLevel("preparation/storacha", "warn") db := testdb.CreateTestDB(t) @@ -327,11 +370,10 @@ func TestAddIndexesForUpload(t *testing.T) { } api := storacha.API{ - Repo: repo, - Client: &client, - ReaderForIndex: carForIndex, - BlobUploadParallelism: 1, - Replicas: 3, + Repo: repo, + Client: &client, + ReaderForIndex: carForIndex, + Replicas: 3, } blobsApi := blobs.API{ @@ -368,8 +410,11 @@ func TestAddIndexesForUpload(t *testing.T) { require.Len(t, shards, 3) require.Len(t, indexes, 1) - err = api.AddIndexesForUpload(t.Context(), upload.ID(), spaceDID, nil) + tasks, err := api.FindIndexAddTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range tasks { + require.Nil(t, task.Run(t.Context())) + } // Reload first shard firstIndex, err := repo.GetIndexByID(t.Context(), indexes[0].ID()) @@ -383,8 +428,11 @@ func TestAddIndexesForUpload(t *testing.T) { require.Equal(t, spaceDID, client.SpaceBlobAddInvocations[0].Space) // Now run post processing. - err = api.PostProcessUploadedIndexes(t.Context(), upload.ID(), spaceDID) + ppTasks, err := api.FindIndexPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range ppTasks { + require.Nil(t, task.Run(t.Context())) + } // Reload first shard firstIndex, err = repo.GetIndexByID(t.Context(), indexes[0].ID()) @@ -416,8 +464,11 @@ func TestAddIndexesForUpload(t *testing.T) { err = blobsApi.CloseUploadIndexes(t.Context(), upload.ID(), recordClosedIndex) require.NoError(t, err) require.Len(t, indexes, 2) - err = api.AddIndexesForUpload(t.Context(), upload.ID(), spaceDID, nil) + tasks, err = api.FindIndexAddTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range tasks { + require.Nil(t, task.Run(t.Context())) + } // Reload second shard secondIndex, err := repo.GetIndexByID(t.Context(), indexes[1].ID()) @@ -430,8 +481,11 @@ func TestAddIndexesForUpload(t *testing.T) { require.Equal(t, spaceDID, client.SpaceBlobAddInvocations[1].Space) // Now run post processing. - err = api.PostProcessUploadedIndexes(t.Context(), upload.ID(), spaceDID) + ppTasks, err = api.FindIndexPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range ppTasks { + require.Nil(t, task.Run(t.Context())) + } // Reload second shard secondIndex, err = repo.GetIndexByID(t.Context(), indexes[1].ID()) @@ -460,10 +514,9 @@ func TestAddStorachaUploadForUpload(t *testing.T) { mclient := mockclient.MockClient{} api := storacha.API{ - Repo: repo, - Client: &mclient, - BlobUploadParallelism: 1, - Replicas: 3, + Repo: repo, + Client: &mclient, + Replicas: 3, } upload, _ := testutil.CreateUpload(t, repo, spaceDID, spacesmodel.WithShardSize(1<<16)) diff --git a/pkg/preparation/types/errors.go b/pkg/preparation/types/errors.go index 1716d2dd..4e10dab9 100644 --- a/pkg/preparation/types/errors.go +++ b/pkg/preparation/types/errors.go @@ -169,17 +169,16 @@ func (e BlobUploadError) ID() id.ID { } type BlobUploadErrors struct { - errs []BlobUploadError + errs []error } -func NewBlobUploadErrors(errs []BlobUploadError) error { +func NewBlobUploadErrors(errs []error) error { + if len(errs) == 0 { + return nil + } return RetriableError{err: BlobUploadErrors{errs: errs}} } -func (e BlobUploadErrors) Errs() []BlobUploadError { - return e.errs -} - func (e BlobUploadErrors) Error() string { var messages []string for _, err := range e.errs { @@ -190,9 +189,5 @@ func (e BlobUploadErrors) Error() string { } func (e BlobUploadErrors) Unwrap() []error { - errs := make([]error, len(e.errs)) - for i, err := range e.errs { - errs[i] = err - } - return errs + return e.errs } diff --git a/pkg/preparation/types/idtask.go b/pkg/preparation/types/idtask.go new file mode 100644 index 00000000..687bbdc3 --- /dev/null +++ b/pkg/preparation/types/idtask.go @@ -0,0 +1,13 @@ +package types + +import ( + "github.com/storacha/guppy/pkg/preparation/internal/worker" + "github.com/storacha/guppy/pkg/preparation/types/id" +) + +// IDTask represents a worker task that can be identified and deduplicated by an +// [id.ID]. +type IDTask struct { + ID id.ID + Run worker.Task +} diff --git a/pkg/preparation/uploads/uploads.go b/pkg/preparation/uploads/uploads.go index 619e9f59..9bb3dad9 100644 --- a/pkg/preparation/uploads/uploads.go +++ b/pkg/preparation/uploads/uploads.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "github.com/ipfs/go-cid" logging "github.com/ipfs/go-log/v2" @@ -17,6 +18,7 @@ import ( "github.com/storacha/guppy/pkg/preparation/bettererrgroup" blobsmodel "github.com/storacha/guppy/pkg/preparation/blobs/model" dagmodel "github.com/storacha/guppy/pkg/preparation/dags/model" + "github.com/storacha/guppy/pkg/preparation/internal/worker" scanmodel "github.com/storacha/guppy/pkg/preparation/scans/model" "github.com/storacha/guppy/pkg/preparation/types" "github.com/storacha/guppy/pkg/preparation/types/id" @@ -34,28 +36,29 @@ type AddNodeToUploadShardsFunc func(ctx context.Context, uploadID id.UploadID, s type AddShardsToUploadIndexesFunc func(ctx context.Context, uploadID id.UploadID, indexCB func(index *blobsmodel.Index) error) error type CloseUploadShardsFunc func(ctx context.Context, uploadID id.UploadID, shardCB func(shard *blobsmodel.Shard) error) error type CloseUploadIndexesFunc func(ctx context.Context, uploadID id.UploadID, indexCB func(index *blobsmodel.Index) error) error -type AddShardsForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID, shardCB func(shard *blobsmodel.Shard) error) error +type FindShardAddTasksForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]types.IDTask, error) +type FindIndexAddTasksForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]types.IDTask, error) type AddNodesToUploadShardsFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID, shardCB func(shard *blobsmodel.Shard) error) error -type AddIndexesForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID, indexCB func(index *blobsmodel.Index) error) error -type PostProcessUploadedShardsFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) error -type PostProcessUploadedIndexesFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) error +type FindShardPostProcessTasksForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]types.IDTask, error) +type FindIndexPostProcessTasksForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]types.IDTask, error) type AddStorachaUploadForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) error type RemoveBadFSEntryFunc func(ctx context.Context, spaceDID did.DID, fsEntryID id.FSEntryID) error type RemoveBadNodesFunc func(ctx context.Context, spaceDID did.DID, nodeCIDs []cid.Cid) error type RemoveShardFunc func(ctx context.Context, shardID id.ShardID) error type API struct { - Repo Repo - ExecuteScan ExecuteScanFunc - ExecuteDagScansForUpload ExecuteDagScansForUploadFunc - AddShardsForUpload AddShardsForUploadFunc - PostProcessUploadedShards PostProcessUploadedShardsFunc - PostProcessUploadedIndexes PostProcessUploadedIndexesFunc - AddIndexesForUpload AddIndexesForUploadFunc - AddStorachaUploadForUpload AddStorachaUploadForUploadFunc - RemoveBadFSEntry RemoveBadFSEntryFunc - RemoveBadNodes RemoveBadNodesFunc - RemoveShard RemoveShardFunc + Repo Repo + ExecuteScan ExecuteScanFunc + ExecuteDagScansForUpload ExecuteDagScansForUploadFunc + BlobUploadParallelism int + FindShardAddTasksForUpload FindShardAddTasksForUploadFunc + FindIndexAddTasksForUpload FindIndexAddTasksForUploadFunc + FindShardPostProcessTasksForUpload FindShardPostProcessTasksForUploadFunc + FindIndexPostProcessTasksForUpload FindIndexPostProcessTasksForUploadFunc + AddStorachaUploadForUpload AddStorachaUploadForUploadFunc + RemoveBadFSEntry RemoveBadFSEntryFunc + RemoveBadNodes RemoveBadNodesFunc + RemoveShard RemoveShardFunc // AddNodesToUploadShards assigns all unsharded nodes for an upload to shards. AddNodesToUploadShards AddNodesToUploadShardsFunc @@ -220,6 +223,7 @@ func (a API) ExecuteUpload(ctx context.Context, uploadID id.UploadID, spaceDID d signal(dagScansAvailable) signal(nodeUploadsAvailable) signal(closedShardsAvailable) + signal(closedIndexesAvailable) signal(uploadedShardsAvailable) signal(uploadedIndexesAvailable) close(scansAvailable) @@ -239,12 +243,12 @@ func (a API) ExecuteUpload(ctx context.Context, uploadID id.UploadID, spaceDID d return cid.Undef, fmt.Errorf("handling bad FS entries worker error [%w]: %w", workersErr, err) } case errors.As(workersErr, &blobUploadErrors): - err := a.handleBadBlobUploads(ctx, uploadID, spaceDID, blobUploadErrors) + err := a.handleBadBlobUploads(ctx, uploadID, blobUploadErrors) if err != nil { return cid.Undef, fmt.Errorf("handling bad shard uploads worker error [%w]: %w", workersErr, err) } case errors.As(workersErr, &badNodesErr): - err := a.handleBadNodes(ctx, uploadID, spaceDID, badNodesErr) + err := a.handleBadNodes(ctx, uploadID, badNodesErr) if err != nil { return cid.Undef, fmt.Errorf("handling bad nodes worker error [%w]: %w", workersErr, err) } @@ -286,13 +290,13 @@ func (a API) handleBadFSEntries(ctx context.Context, uploadID id.UploadID, badFS return nil } -func (a API) handleBadBlobUploads(ctx context.Context, uploadID id.UploadID, spaceDID did.DID, blobUploadErrors types.BlobUploadErrors) error { +func (a API) handleBadBlobUploads(ctx context.Context, uploadID id.UploadID, blobUploadErrors types.BlobUploadErrors) error { // when there's a bad shard upload, it's not based on a problem locally usually, unless bad nodes were read during upload - for _, e := range blobUploadErrors.Errs() { + for _, e := range blobUploadErrors.Unwrap() { // bad nodes error can happen from reading car during upload var badNodesErr types.BadNodesError - if errors.As(e.Unwrap(), &badNodesErr) { - err := a.handleBadNodes(ctx, uploadID, spaceDID, badNodesErr) + if errors.As(e, &badNodesErr) { + err := a.handleBadNodes(ctx, uploadID, badNodesErr) if err != nil { return err } @@ -302,7 +306,7 @@ func (a API) handleBadBlobUploads(ctx context.Context, uploadID id.UploadID, spa return nil } -func (a API) handleBadNodes(ctx context.Context, uploadID id.UploadID, spaceDID did.DID, badNodesErr types.BadNodesError) error { +func (a API) handleBadNodes(ctx context.Context, uploadID id.UploadID, badNodesErr types.BadNodesError) error { upload, err := a.Repo.GetUploadByID(ctx, uploadID) if err != nil { return fmt.Errorf("getting upload %s after finding bad nodes: %w", uploadID, err) @@ -371,39 +375,44 @@ func runScanWorker( span.End() }() - return Worker( + if te := worker.Run( ctx, scansAvailable, + 1, + + // findWork + func(ctx context.Context) ([]worker.Task, error) { + return []worker.Task{ + func(ctx context.Context) worker.TaskError { + if api.AssumeUnchangedSources { + upload, err := api.Repo.GetUploadByID(ctx, uploadID) + if err != nil { + return worker.NewFatalError(fmt.Errorf("checking upload for existing scan: %w", err)) + } + if upload.HasRootFSEntryID() { + log.Infow("Skipping FS rescan (--assume-unchanged-sources): scan already exists", "upload", uploadID) + return nil + } + log.Infow("No existing scan found, performing FS scan despite --assume-unchanged-sources", "upload", uploadID) + } + + err := api.ExecuteScan(ctx, uploadID, func(entry scanmodel.FSEntry) error { + _, isDirectory := entry.(*scanmodel.Directory) + _, err := api.Repo.CreateDAGScan(ctx, entry.ID(), isDirectory, uploadID, spaceDID) + if err != nil { + return fmt.Errorf("creating DAG scan: %w", err) + } + signal(dagScansAvailable) + return nil + }) + + if err != nil { + return worker.NewFatalError(fmt.Errorf("running scans: %w", err)) + } - // doWork - func() error { - if api.AssumeUnchangedSources { - upload, err := api.Repo.GetUploadByID(ctx, uploadID) - if err != nil { - return fmt.Errorf("checking upload for existing scan: %w", err) - } - if upload.HasRootFSEntryID() { - log.Infow("Skipping FS rescan (--assume-unchanged-sources): scan already exists", "upload", uploadID) return nil - } - log.Infow("No existing scan found, performing FS scan despite --assume-unchanged-sources", "upload", uploadID) - } - - err := api.ExecuteScan(ctx, uploadID, func(entry scanmodel.FSEntry) error { - _, isDirectory := entry.(*scanmodel.Directory) - _, err := api.Repo.CreateDAGScan(ctx, entry.ID(), isDirectory, uploadID, spaceDID) - if err != nil { - return fmt.Errorf("creating DAG scan: %w", err) - } - signal(dagScansAvailable) - return nil - }) - - if err != nil { - return fmt.Errorf("running scans: %w", err) - } - - return nil + }, + }, nil }, // finalize @@ -411,7 +420,10 @@ func runScanWorker( close(dagScansAvailable) return nil }, - ) + ); te != nil { + err = te + } + return err } // runDAGScanWorker runs the worker that scans files and directories into blocks, @@ -447,22 +459,27 @@ func runDAGScanWorker( span.End() }() - return Worker( + if te := worker.Run( ctx, dagScansAvailable, + 1, - // doWork - func() error { - err := api.ExecuteDagScansForUpload(ctx, uploadID, func(node dagmodel.Node, data []byte) error { - signal(nodeUploadsAvailable) - return nil - }) + // findWork + func(ctx context.Context) ([]worker.Task, error) { + return []worker.Task{ + func(ctx context.Context) worker.TaskError { + err := api.ExecuteDagScansForUpload(ctx, uploadID, func(node dagmodel.Node, data []byte) error { + signal(nodeUploadsAvailable) + return nil + }) - if err != nil { - return fmt.Errorf("running dag scans for upload %s: %w", uploadID, err) - } + if err != nil { + return worker.NewFatalError(fmt.Errorf("running dag scans for upload %s: %w", uploadID, err)) + } - return nil + return nil + }, + }, nil }, // finalize @@ -487,7 +504,10 @@ func runDAGScanWorker( close(nodeUploadsAvailable) return nil }, - ) + ); te != nil { + err = te + } + return err } // runShardingWorker runs the worker that assigns nodes to shards. @@ -527,17 +547,22 @@ func runShardingWorker( return nil } - return Worker( + if te := worker.Run( ctx, nodeUploadsAvailable, - - // doWork - func() error { - err := api.AddNodesToUploadShards(ctx, uploadID, spaceDID, handleClosedShard) - if err != nil { - return fmt.Errorf("adding nodes to shards for upload %s: %w", uploadID, err) - } - return nil + 1, + + // findWork + func(ctx context.Context) ([]worker.Task, error) { + return []worker.Task{ + func(ctx context.Context) worker.TaskError { + err := api.AddNodesToUploadShards(ctx, uploadID, spaceDID, handleClosedShard) + if err != nil { + return worker.NewFatalError(fmt.Errorf("adding nodes to shards for upload %s: %w", uploadID, err)) + } + return nil + }, + }, nil }, // finalize @@ -551,7 +576,10 @@ func runShardingWorker( return nil }, - ) + ); te != nil { + err = te + } + return err } func runIndexingWorker( @@ -590,17 +618,22 @@ func runIndexingWorker( return nil } - return Worker( + if te := worker.Run( ctx, shardsNeedIndexing, - - // doWork - func() error { - err := api.AddShardsToUploadIndexes(ctx, uploadID, handleClosedIndex) - if err != nil { - return fmt.Errorf("adding shards to indexes for upload %s: %w", uploadID, err) - } - return nil + 1, + + // findWork + func(ctx context.Context) ([]worker.Task, error) { + return []worker.Task{ + func(ctx context.Context) worker.TaskError { + err := api.AddShardsToUploadIndexes(ctx, uploadID, handleClosedIndex) + if err != nil { + return worker.NewFatalError(fmt.Errorf("adding shards to indexes for upload %s: %w", uploadID, err)) + } + return nil + }, + }, nil }, // finalize @@ -614,7 +647,10 @@ func runIndexingWorker( return nil }, - ) + ); te != nil { + err = te + } + return err } // runShardUploadWorker runs the worker that adds shards to Storacha. @@ -649,21 +685,35 @@ func runShardUploadWorker( span.End() }() - return Worker( + var inFlightShards sync.Map + + te := worker.Run( ctx, closedShardsAvailable, + api.BlobUploadParallelism, - // doWork - func() error { - err := api.AddShardsForUpload(ctx, uploadID, spaceDID, func(shard *blobsmodel.Shard) error { - signal(uploadedShardsAvailable) - return nil - }) + // findWork + func(ctx context.Context) ([]worker.Task, error) { + rawTasks, err := api.FindShardAddTasksForUpload(ctx, uploadID, spaceDID) if err != nil { - return fmt.Errorf("`space/blob/add`ing shards for upload %s: %w", uploadID, err) + return nil, err } - - return nil + var tasks []worker.Task + for _, raw := range rawTasks { + // Ignore tasks that are already in flight. + if _, loaded := inFlightShards.LoadOrStore(raw.ID, struct{}{}); loaded { + continue + } + tasks = append(tasks, func(ctx context.Context) worker.TaskError { + defer inFlightShards.Delete(raw.ID) + taskErr := raw.Run(ctx) + if taskErr == nil { + signal(uploadedShardsAvailable) + } + return taskErr + }) + } + return tasks, nil }, // finalize @@ -672,6 +722,15 @@ func runShardUploadWorker( return nil }, ) + + var nonFatals []error + var fatal error + if te != nil { + nonFatals = te.NonFatalErrors() + fatal = te.FatalError() + } + err = errors.Join(fatal, types.NewBlobUploadErrors(nonFatals)) + return err } func runPostProcessShardWorker( @@ -704,17 +763,30 @@ func runPostProcessShardWorker( span.End() }() - return Worker( + var inFlightShards sync.Map + + if te := worker.Run( ctx, uploadedShardsAvailable, + api.BlobUploadParallelism, - // doWork - func() error { - err := api.PostProcessUploadedShards(ctx, uploadID, spaceDID) + // findWork + func(ctx context.Context) ([]worker.Task, error) { + rawTasks, err := api.FindShardPostProcessTasksForUpload(ctx, uploadID, spaceDID) if err != nil { - return fmt.Errorf("`post-processing shards for upload %s: %w", uploadID, err) + return nil, err } - return nil + var tasks []worker.Task + for _, raw := range rawTasks { + if _, loaded := inFlightShards.LoadOrStore(raw.ID, struct{}{}); loaded { + continue + } + tasks = append(tasks, func(ctx context.Context) worker.TaskError { + defer inFlightShards.Delete(raw.ID) + return raw.Run(ctx) + }) + } + return tasks, nil }, // finalize @@ -723,10 +795,12 @@ func runPostProcessShardWorker( if err != nil { return fmt.Errorf("`upload/add`ing upload %s: %w", uploadID, err) } - return nil }, - ) + ); te != nil { + err = te + } + return err } // runIndexUploadWorker runs the worker that adds indexes to Storacha. @@ -761,20 +835,35 @@ func runIndexUploadWorker( span.End() }() - return Worker( + var inFlightIndexes sync.Map + + te := worker.Run( ctx, closedIndexesAvailable, + api.BlobUploadParallelism, - // doWork - func() error { - err := api.AddIndexesForUpload(ctx, uploadID, spaceDID, func(index *blobsmodel.Index) error { - signal(uploadedIndexesAvailable) - return nil - }) + // findWork + func(ctx context.Context) ([]worker.Task, error) { + rawTasks, err := api.FindIndexAddTasksForUpload(ctx, uploadID, spaceDID) if err != nil { - return fmt.Errorf("`space/blob/add`ing indexes for upload %s: %w", uploadID, err) + return nil, err } - return nil + var tasks []worker.Task + for _, raw := range rawTasks { + // Ignore tasks that are already in flight. + if _, loaded := inFlightIndexes.LoadOrStore(raw.ID, struct{}{}); loaded { + continue + } + tasks = append(tasks, func(ctx context.Context) worker.TaskError { + defer inFlightIndexes.Delete(raw.ID) + taskErr := raw.Run(ctx) + if taskErr == nil { + signal(uploadedIndexesAvailable) + } + return taskErr + }) + } + return tasks, nil }, // finalize @@ -783,6 +872,15 @@ func runIndexUploadWorker( return nil }, ) + + var nonFatals []error + var fatal error + if te != nil { + nonFatals = te.NonFatalErrors() + fatal = te.FatalError() + } + err = errors.Join(fatal, types.NewBlobUploadErrors(nonFatals)) + return err } func runPostProcessIndexWorker( @@ -815,20 +913,36 @@ func runPostProcessIndexWorker( span.End() }() - return Worker( + var inFlightIndexes sync.Map + + if te := worker.Run( ctx, uploadedIndexesAvailable, + api.BlobUploadParallelism, - // doWork - func() error { - err := api.PostProcessUploadedIndexes(ctx, uploadID, spaceDID) + // findWork + func(ctx context.Context) ([]worker.Task, error) { + rawTasks, err := api.FindIndexPostProcessTasksForUpload(ctx, uploadID, spaceDID) if err != nil { - return fmt.Errorf("`post-processing indexes for upload %s: %w", uploadID, err) + return nil, err } - return nil + var tasks []worker.Task + for _, raw := range rawTasks { + if _, loaded := inFlightIndexes.LoadOrStore(raw.ID, struct{}{}); loaded { + continue + } + tasks = append(tasks, func(ctx context.Context) worker.TaskError { + defer inFlightIndexes.Delete(raw.ID) + return raw.Run(ctx) + }) + } + return tasks, nil }, // finalize nil, - ) + ); te != nil { + err = te + } + return err } diff --git a/pkg/preparation/uploads/worker.go b/pkg/preparation/uploads/worker.go deleted file mode 100644 index fe1147d4..00000000 --- a/pkg/preparation/uploads/worker.go +++ /dev/null @@ -1,29 +0,0 @@ -package uploads - -import ( - "context" - "fmt" - - "github.com/storacha/guppy/internal/ctxutil" -) - -func Worker(ctx context.Context, in <-chan struct{}, doWork func() error, finalize func() error) error { - for { - select { - case <-ctx.Done(): - return ctxutil.Cause(ctx) - case _, ok := <-in: - if !ok { - if finalize != nil { - if err := finalize(); err != nil { - return fmt.Errorf("worker finalize encountered an error: %w", err) - } - } - return nil - } - if err := doWork(); err != nil { - return fmt.Errorf("worker encountered an error: %w", err) - } - } - } -} diff --git a/pkg/preparation/uploads/worker_test.go b/pkg/preparation/uploads/worker_test.go deleted file mode 100644 index ae4b1a50..00000000 --- a/pkg/preparation/uploads/worker_test.go +++ /dev/null @@ -1,145 +0,0 @@ -package uploads_test - -import ( - "errors" - "testing" - "time" - - "github.com/storacha/guppy/pkg/preparation/uploads" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -type unwrappableError interface { - error - Unwrap() error -} - -// e is a shorthand helper function that uses [require.EventuallyWithT] with -// a standard timeout and interval, to keep noise out of the tests. -func e(t *testing.T, condition func(collect *assert.CollectT)) { - t.Helper() - require.EventuallyWithT(t, condition, time.Second, 10*time.Millisecond) -} - -func TestWorker(t *testing.T) { - t.Run("runs the work function for every signal received, then the finalize function when the channel closes", func(t *testing.T) { - signalChan := make(chan struct{}, 1) - resultChan := make(chan error, 1) - var runs int - var finalizes int - - go func() { - defer close(resultChan) - - resultChan <- uploads.Worker(t.Context(), signalChan, func() error { - runs++ - return nil - }, func() error { - finalizes++ - return nil - }) - }() - - require.Equal(t, 0, runs, "worker should not run before signal") - signalChan <- struct{}{} - e(t, func(t *assert.CollectT) { require.Equal(t, 1, runs, "worker should run once after signal") }) - signalChan <- struct{}{} - e(t, func(t *assert.CollectT) { require.Equal(t, 2, runs, "worker should run again after second signal") }) - - require.Equal(t, 0, finalizes, "finalize function should be called until the channel closes") - close(signalChan) - e(t, func(t *assert.CollectT) { - require.Equal(t, 1, finalizes, "finalize function should be called once the channel closes") - }) - - result := <-resultChan - require.Nil(t, result, "result should be nil after successful runs") - }) - - t.Run("immediately responds with any work error, skipping the finalizer", func(t *testing.T) { - workerErr := errors.New("error in doWork") - signalChan := make(chan struct{}, 3) - resultChan := make(chan error, 1) - var runs int - var finalizes int - - go func() { - defer close(resultChan) - resultChan <- uploads.Worker(t.Context(), signalChan, func() error { - runs++ - // Fail on the second run - if runs == 2 { - return workerErr - } - return nil - }, func() error { - finalizes++ - return nil - }) - }() - - // Send three signals; the second should cause an error, the third should not run - signalChan <- struct{}{} - signalChan <- struct{}{} - signalChan <- struct{}{} - - result, ok := (<-resultChan).(unwrappableError) - require.True(t, ok, "result should be a wrapped error") - require.ErrorContains(t, result, "worker encountered an error: error in doWork") - require.Equal(t, workerErr, result.Unwrap(), "worker should send back the error it encountered, wrapped") - require.Equal(t, 2, runs, "worker should have stopped after encountering an error") - require.Equal(t, 0, finalizes, "finalize function should not have be called") - }) - - t.Run("responds with any finalize error", func(t *testing.T) { - finalizerErr := errors.New("error in finalize") - signalChan := make(chan struct{}, 3) - resultChan := make(chan error, 1) - var runs int - - go func() { - defer close(resultChan) - resultChan <- uploads.Worker(t.Context(), signalChan, func() error { - runs++ - return nil - }, func() error { - return finalizerErr - }) - }() - - // Send three signals; all should run - signalChan <- struct{}{} - signalChan <- struct{}{} - signalChan <- struct{}{} - close(signalChan) - - result, ok := (<-resultChan).(unwrappableError) - require.True(t, ok, "result should be a wrapped error") - require.ErrorContains(t, result, "worker finalize encountered an error: error in finalize") - require.Equal(t, finalizerErr, result.Unwrap(), "worker should send back the error it encountered, wrapped") - require.Equal(t, 3, runs, "worker should have run all three times") - }) - - t.Run("ignores a nil finalizer", func(t *testing.T) { - signalChan := make(chan struct{}, 1) - resultChan := make(chan error, 1) - var ran bool - - go func() { - defer close(resultChan) - - resultChan <- uploads.Worker(t.Context(), signalChan, func() error { - ran = true - return nil - }, nil) - }() - - require.False(t, ran, "worker should not run before signal") - signalChan <- struct{}{} - e(t, func(t *assert.CollectT) { require.True(t, ran, "worker should run after signal") }) - close(signalChan) - result := <-resultChan - require.Nil(t, result, "result should be nil after successful runs and no finalizer") - }) -}