diff --git a/.gitignore b/.gitignore index f9a87baa..c38ad72e 100644 --- a/.gitignore +++ b/.gitignore @@ -8,10 +8,6 @@ guppy # Environment Files .env* -# Config Files -*.yaml -*.toml - # IDE's *.idea/ diff --git a/Dockerfile b/Dockerfile index 7c278eab..0af30c05 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,7 +1,7 @@ # ============================================ -# Build stage (shared) +# Build base stage # ============================================ -FROM golang:1.25.3-trixie AS build +FROM golang:1.26.1-trixie AS build-base # Docker sets TARGETARCH automatically during multi-platform builds ARG TARGETARCH @@ -10,15 +10,32 @@ WORKDIR /go/src/guppy COPY go.* . RUN go mod download -COPY . . +COPY --parents cmd internal pkg Makefile main.go version.json ./ + +# ============================================ +# Production build stage +# ============================================ +FROM build-base AS build-prod + +# Allow the Makefile to look up the git commit, at the expense of busting the +# cache whenever `.git` changes. +COPY --parents .git ./ # Production build - with symbol stripping -RUN CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH} make guppy-prod +RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH} make guppy-prod # ============================================ # Debug build stage # ============================================ -FROM build AS build-debug +FROM build-base AS build-debug + +# Debug build - no optimizations, no inlining +RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH} make guppy-debug + +# ============================================ +# Debug tools +# ============================================ +FROM golang:1.26.1-trixie AS build-debug-tools ARG TARGETARCH @@ -26,9 +43,6 @@ ARG TARGETARCH RUN GOARCH=${TARGETARCH} go install github.com/go-delve/delve/cmd/dlv@latest && \ GOARCH=${TARGETARCH} go install github.com/storacha/randdir@latest -# Debug build - no optimizations, no inlining -RUN CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH} make guppy-debug - # ============================================ # Production image # ============================================ @@ -39,7 +53,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ curl \ && rm -rf /var/lib/apt/lists/* -COPY --from=build /go/src/guppy/guppy /usr/bin/guppy +COPY --from=build-prod /go/src/guppy/guppy /usr/bin/guppy ENTRYPOINT ["/usr/bin/guppy"] @@ -69,8 +83,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ && rm -rf /var/lib/apt/lists/* # Delve debugger and randdir tool -COPY --from=build-debug /go/bin/dlv /usr/bin/dlv -COPY --from=build-debug /go/bin/randdir /usr/bin/randdir +COPY --from=build-debug-tools /go/bin/dlv /usr/bin/dlv +COPY --from=build-debug-tools /go/bin/randdir /usr/bin/randdir # Debug binary (with symbols, no optimizations) COPY --from=build-debug /go/src/guppy/guppy /usr/bin/guppy @@ -87,3 +101,19 @@ RUN echo 'alias ll="ls -la"' >> /etc/bash.bashrc && \ SHELL ["/bin/bash", "-c"] ENTRYPOINT ["/usr/bin/guppy"] + +# ============================================ +# Test image (without interactive tools) +# ============================================ +FROM debian:bookworm-slim AS test + +# Debug binary (with symbols, no optimizations) +COPY --from=build-debug /go/src/guppy/guppy /usr/bin/guppy + +# Create data directories +RUN mkdir -p /root/.storacha/guppy /root/.config/guppy + +WORKDIR /root + +SHELL ["/bin/bash", "-c"] +ENTRYPOINT ["/usr/bin/guppy"] diff --git a/Makefile b/Makefile index 3e7b835d..27fe823c 100644 --- a/Makefile +++ b/Makefile @@ -6,7 +6,7 @@ DB_PATH ?= ~/.storacha/guppy/preparation.db GOOSE := go tool goose VERSION=$(shell awk -F'"' '/"version":/ {print $$4}' version.json) -COMMIT=$(shell git rev-parse --short HEAD) +COMMIT=$(shell git rev-parse --short HEAD 2>/dev/null || echo "unknown") DATE=$(shell date -u -Iseconds) GOFLAGS=-ldflags="-X github.com/storacha/guppy/pkg/build.version=$(VERSION) -X github.com/storacha/guppy/pkg/build.Commit=$(COMMIT) -X github.com/storacha/guppy/pkg/build.Date=$(DATE) -X github.com/storacha/guppy/pkg/build.BuiltBy=make" DOCKER?=$(shell which docker) @@ -59,3 +59,7 @@ docker-prod: docker-setup docker-dev: docker-setup $(DOCKER) buildx build --platform linux/amd64,linux/arm64 --target dev -t guppy:dev . + +test-upload: + @echo "Running upload test..." + ./test/doupload diff --git a/cmd/upload/root.go b/cmd/upload/root.go index b7ecba44..83494f32 100644 --- a/cmd/upload/root.go +++ b/cmd/upload/root.go @@ -107,6 +107,7 @@ var Cmd = &cobra.Command{ preparation.WithAssumeUnchangedSources(rootFlags.assumeUnchangedSources), preparation.WithEventBus(eb), preparation.WithReplicas(cfg.Upload.Replicas), + preparation.WithPutHTTPClient(cmdutil.TracedHTTPClient), ) allUploads, err := api.FindOrCreateUploads(ctx, spaceDID) if err != nil { diff --git a/cmd/upload/source/add.go b/cmd/upload/source/add.go index 34bdd1d9..315649d1 100644 --- a/cmd/upload/source/add.go +++ b/cmd/upload/source/add.go @@ -73,7 +73,7 @@ var AddCmd = &cobra.Command{ return err } - api := preparation.NewAPI(repo, client) + api := preparation.NewAPI(repo, client, preparation.WithPutHTTPClient(cmdutil.TracedHTTPClient)) // Parse shard size if provided var spaceOptions []model.SpaceOption diff --git a/internal/cmdutil/cmdutil.go b/internal/cmdutil/cmdutil.go index 38cd85d9..19ef354e 100644 --- a/internal/cmdutil/cmdutil.go +++ b/internal/cmdutil/cmdutil.go @@ -43,8 +43,10 @@ func envSigner() (principal.Signer, error) { return signer.Parse(str) } -var tracedHttpClient = &http.Client{ - Transport: otelhttp.NewTransport(http.DefaultTransport), +// TracedHTTPClient is an HTTP client with OpenTelemetry tracing and a guppy +// User-Agent header on all outbound requests. +var TracedHTTPClient = &http.Client{ + Transport: newUserAgentTransport(otelhttp.NewTransport(http.DefaultTransport)), } // MustGetClient creates a new client suitable for the CLI, using stored data, @@ -77,7 +79,7 @@ func MustGetClientForNetwork(storePath string, networkCfg config.NetworkConfig, conn, err := uclient.NewConnection( network.UploadID, - uhttp.NewChannel(&network.UploadURL, uhttp.WithClient(tracedHttpClient)), + uhttp.NewChannel(&network.UploadURL, uhttp.WithClient(TracedHTTPClient)), uclient.WithOutboundCodec(car.NewOutboundCodec()), ) if err != nil { @@ -90,7 +92,7 @@ func MustGetClientForNetwork(storePath string, networkCfg config.NetworkConfig, append( options, client.WithConnection(conn), - client.WithReceiptsClient(receiptclient.New(&network.ReceiptsURL, receiptclient.WithHTTPClient(tracedHttpClient))), + client.WithReceiptsClient(receiptclient.New(&network.ReceiptsURL, receiptclient.WithHTTPClient(TracedHTTPClient))), )..., )..., ) @@ -132,7 +134,7 @@ func MustGetIndexClient(networkCfg config.NetworkConfig) (*indexclient.Client, u func MustGetIndexClientForNetwork(networkCfg config.NetworkConfig, flagName string) (*indexclient.Client, ucan.Principal) { network := MustGetNetworkConfig(networkCfg, flagName) - client, err := indexclient.New(network.IndexerID, network.IndexerURL, indexclient.WithHTTPClient(tracedHttpClient)) + client, err := indexclient.New(network.IndexerID, network.IndexerURL, indexclient.WithHTTPClient(TracedHTTPClient)) if err != nil { log.Fatal(err) } diff --git a/internal/cmdutil/useragent.go b/internal/cmdutil/useragent.go new file mode 100644 index 00000000..267d41cc --- /dev/null +++ b/internal/cmdutil/useragent.go @@ -0,0 +1,26 @@ +package cmdutil + +import ( + "fmt" + "net/http" + + "github.com/storacha/guppy/pkg/build" +) + +type userAgentTransport struct { + userAgent string + base http.RoundTripper +} + +func (t *userAgentTransport) RoundTrip(req *http.Request) (*http.Response, error) { + req = req.Clone(req.Context()) + req.Header.Set("User-Agent", t.userAgent) + return t.base.RoundTrip(req) +} + +func newUserAgentTransport(base http.RoundTripper) http.RoundTripper { + return &userAgentTransport{ + userAgent: fmt.Sprintf("guppy/%s", build.Version), + base: base, + } +} diff --git a/pkg/preparation/preparation.go b/pkg/preparation/preparation.go index c3a93b74..227030b4 100644 --- a/pkg/preparation/preparation.go +++ b/pkg/preparation/preparation.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io/fs" + "net/http" "os" "path/filepath" @@ -12,6 +13,7 @@ import ( "github.com/storacha/go-ucanto/did" "github.com/storacha/guppy/pkg/bus" + clientpkg "github.com/storacha/guppy/pkg/client" "github.com/storacha/guppy/pkg/preparation/blobs" "github.com/storacha/guppy/pkg/preparation/dags" "github.com/storacha/guppy/pkg/preparation/dags/nodereader" @@ -61,6 +63,7 @@ type config struct { assumeUnchangedSources bool bus bus.Bus replicas uint + putHTTPClient *http.Client } const ( @@ -156,14 +159,18 @@ func NewAPI(repo Repo, client StorachaClient, options ...Option) API { ShardEncoder: blobs.NewCAREncoder(), } + var blobAddOptions []clientpkg.SpaceBlobAddOption + if cfg.putHTTPClient != nil { + 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, + Repo: repo, + Client: client, + ReaderForShard: blobsAPI.ReaderForShard, + ReaderForIndex: blobsAPI.ReaderForIndex, + Bus: cfg.bus, + Replicas: cfg.replicas, + BlobAddOptions: blobAddOptions, } uploadsAPI = uploads.API{ @@ -175,10 +182,11 @@ func NewAPI(repo Repo, client StorachaClient, options ...Option) API { AddShardsToUploadIndexes: blobsAPI.AddShardsToUploadIndexes, CloseUploadShards: blobsAPI.CloseUploadShards, CloseUploadIndexes: blobsAPI.CloseUploadIndexes, - AddShardsForUpload: storachaAPI.AddShardsForUpload, - PostProcessUploadedShards: storachaAPI.PostProcessUploadedShards, - PostProcessUploadedIndexes: storachaAPI.PostProcessUploadedIndexes, - AddIndexesForUpload: storachaAPI.AddIndexesForUpload, + FindShardAddTasksForUpload: storachaAPI.FindShardAddTasksForUpload, + FindIndexAddTasksForUpload: storachaAPI.FindIndexAddTasksForUpload, + BlobUploadParallelism: cfg.blobUploadParallelism, + FindShardPostProcessTasksForUpload: storachaAPI.FindShardPostProcessTasksForUpload, + FindIndexPostProcessTasksForUpload: storachaAPI.FindIndexPostProcessTasksForUpload, AddStorachaUploadForUpload: storachaAPI.AddStorachaUploadForUpload, RemoveBadFSEntry: scansAPI.RemoveBadFSEntry, RemoveBadNodes: dagsAPI.RemoveBadNodes, @@ -247,6 +255,14 @@ func WithReplicas(replicas uint) Option { } } +// WithPutHTTPClient sets the HTTP client used for blob PUT uploads. +func WithPutHTTPClient(c *http.Client) Option { + return func(cfg *config) error { + cfg.putHTTPClient = c + return nil + } +} + func (a API) FindOrCreateSpace(ctx context.Context, spaceDID did.DID, name string, options ...spacesmodel.SpaceOption) (*spacesmodel.Space, error) { return a.Spaces.FindOrCreateSpace(ctx, spaceDID, name, options...) } 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 3fedf426..7ec66230 100644 --- a/pkg/preparation/storacha/storacha.go +++ b/pkg/preparation/storacha/storacha.go @@ -20,11 +20,6 @@ import ( "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" @@ -34,6 +29,9 @@ import ( gtypes "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 ( @@ -58,163 +56,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) ([]gtypes.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([]gtypes.IDTask, 0, len(closedShards)) + for _, shard := range closedShards { + tasks = append(tasks, gtypes.IDTask{ + ID: shard.ID(), + Run: func(ctx context.Context) (error, error) { + if err := a.addBlob(ctx, shard, spaceDID); err != nil { + err = fmt.Errorf("failed to add shard %s: %w", shard, err) + // [gtypes.BlobUploadError]s are non-fatal. + var errBlobUpload gtypes.BlobUploadError + if errors.As(err, &errBlobUpload) { + return err, nil + } + log.Errorf("%v", err) + return nil, err + } + return nil, 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) ([]gtypes.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([]gtypes.IDTask, 0, len(closedIndexes)) + for _, index := range closedIndexes { + tasks = append(tasks, gtypes.IDTask{ + ID: index.ID(), + Run: func(ctx context.Context) (error, error) { + if err := a.addBlob(ctx, index, spaceDID); err != nil { + err = fmt.Errorf("failed to add index %s: %w", index, err) + // [gtypes.BlobUploadError]s are non-fatal. + var errBlobUpload gtypes.BlobUploadError + if errors.As(err, &errBlobUpload) { + return err, nil + } + log.Errorf("%v", err) + return nil, err } - } - log.Infof("Successfully added blob %s", blob.ID()) - return nil + return nil, 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) ([]gtypes.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([]gtypes.IDTask, 0, len(uploadedShards)) + for _, shard := range uploadedShards { + tasks = append(tasks, gtypes.IDTask{ + ID: shard.ID(), + Run: func(ctx context.Context) (error, error) { + 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 nil, 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, 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) { @@ -305,7 +263,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 @@ -325,7 +283,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) } @@ -336,7 +294,7 @@ func (a API) spaceBlobAdd(ctx context.Context, content io.Reader, spaceDID did.D ctx, span := tracer.Start(ctx, "space-blob-add") defer span.End() - return a.Client.SpaceBlobAdd(ctx, content, spaceDID, opts...) + return a.Client.SpaceBlobAdd(ctx, content, spaceDID, append(a.BlobAddOptions, opts...)...) } func (a API) spaceBlobReplicate(ctx context.Context, blob model.Blob, spaceDID did.DID, locationCommitment delegation.Delegation) error { @@ -390,52 +348,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) ([]gtypes.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([]gtypes.IDTask, 0, len(uploadedIndexes)) + for _, index := range uploadedIndexes { + tasks = append(tasks, gtypes.IDTask{ + ID: index.ID(), + Run: func(ctx context.Context) (error, error) { + 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 nil, fmt.Errorf("failed to post-process index %s: %w", index, err) + } + log.Infof("Successfully post-processed index %s", index.ID()) + return nil, 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..0cca97b3 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,13 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // Reload shards firstShard, err = repo.GetShardByID(t.Context(), firstShard.ID()) @@ -101,8 +105,13 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // Reload shards firstShard, err = repo.GetShardByID(t.Context(), firstShard.ID()) require.NoError(t, err) @@ -131,8 +140,13 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // Reload second shard secondShard, err = repo.GetShardByID(t.Context(), secondShard.ID()) @@ -145,8 +159,13 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // Reload second shard secondShard, err = repo.GetShardByID(t.Context(), secondShard.ID()) @@ -182,11 +201,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 +221,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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, err) + require.ErrorContains(t, nonFatal, "simulated SpaceBlobAdd error") + } + ppTasks, err := api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range ppTasks { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // It should have `space/blob/add`ed (and failed)... require.Len(t, client.SpaceBlobAddInvocations, 1) @@ -216,7 +244,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 +253,20 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } + ppTasks, err = api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) + require.NoError(t, err) + for _, task := range ppTasks { + nonFatal, fatalErr := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.ErrorContains(t, fatalErr, "simulated SpaceBlobReplicate error") + } // It should have `space/blob/add`ed again... require.Len(t, client.SpaceBlobAddInvocations, 2) @@ -243,10 +281,20 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } + ppTasks, err = api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) + require.NoError(t, err) + for _, task := range ppTasks { + nonFatal, fatalErr := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.ErrorContains(t, fatalErr, "simulated FilecoinOffer error") + } // It should NOT `space/blob/add` again... require.Len(t, client.SpaceBlobAddInvocations, 2) @@ -274,11 +322,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 +340,20 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } + ppTasks, err := api.FindShardPostProcessTasksForUpload(t.Context(), upload.ID(), spaceDID) require.NoError(t, err) + for _, task := range ppTasks { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // It should `space/blob/add`... require.Len(t, client.SpaceBlobAddInvocations, 1) @@ -309,7 +366,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 +384,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 +424,13 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // Reload first shard firstIndex, err := repo.GetIndexByID(t.Context(), indexes[0].ID()) @@ -383,8 +444,13 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // Reload first shard firstIndex, err = repo.GetIndexByID(t.Context(), indexes[0].ID()) @@ -416,8 +482,13 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // Reload second shard secondIndex, err := repo.GetIndexByID(t.Context(), indexes[1].ID()) @@ -430,8 +501,13 @@ 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 { + nonFatal, err := task.Run(t.Context()) + require.NoError(t, nonFatal) + require.NoError(t, err) + } // Reload second shard secondIndex, err = repo.GetIndexByID(t.Context(), indexes[1].ID()) @@ -460,10 +536,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..5a3573c0 100644 --- a/pkg/preparation/types/errors.go +++ b/pkg/preparation/types/errors.go @@ -1,6 +1,7 @@ package types import ( + "context" "errors" "fmt" "strings" @@ -169,17 +170,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 +190,14 @@ 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 +} + +// IDTask represents a task that can be identified and deduplicated by an +// [id.ID]. The task's Run function returns two errors: a non-fatal error that +// can be collected and reported after all tasks have completed, and a fatal +// error that should cause immediate cancellation of all other tasks. +type IDTask struct { + ID id.ID + Run func(context.Context) (error, error) } diff --git a/pkg/preparation/uploads/uploads.go b/pkg/preparation/uploads/uploads.go index 619e9f59..f54b22d9 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" @@ -19,6 +20,7 @@ import ( dagmodel "github.com/storacha/guppy/pkg/preparation/dags/model" scanmodel "github.com/storacha/guppy/pkg/preparation/scans/model" "github.com/storacha/guppy/pkg/preparation/types" + gtypes "github.com/storacha/guppy/pkg/preparation/types" "github.com/storacha/guppy/pkg/preparation/types/id" "github.com/storacha/guppy/pkg/preparation/uploads/model" ) @@ -34,11 +36,11 @@ 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) ([]gtypes.IDTask, error) +type FindIndexAddTasksForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]gtypes.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) ([]gtypes.IDTask, error) +type FindIndexPostProcessTasksForUploadFunc func(ctx context.Context, uploadID id.UploadID, spaceDID did.DID) ([]gtypes.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 @@ -48,10 +50,11 @@ type API struct { Repo Repo ExecuteScan ExecuteScanFunc ExecuteDagScansForUpload ExecuteDagScansForUploadFunc - AddShardsForUpload AddShardsForUploadFunc - PostProcessUploadedShards PostProcessUploadedShardsFunc - PostProcessUploadedIndexes PostProcessUploadedIndexesFunc - AddIndexesForUpload AddIndexesForUploadFunc + BlobUploadParallelism int + FindShardAddTasksForUpload FindShardAddTasksForUploadFunc + FindIndexAddTasksForUpload FindIndexAddTasksForUploadFunc + FindShardPostProcessTasksForUpload FindShardPostProcessTasksForUploadFunc + FindIndexPostProcessTasksForUpload FindIndexPostProcessTasksForUploadFunc AddStorachaUploadForUpload AddStorachaUploadForUploadFunc RemoveBadFSEntry RemoveBadFSEntryFunc RemoveBadNodes RemoveBadNodesFunc @@ -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) @@ -288,10 +292,10 @@ func (a API) handleBadFSEntries(ctx context.Context, uploadID id.UploadID, badFS func (a API) handleBadBlobUploads(ctx context.Context, uploadID id.UploadID, spaceDID did.DID, 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) { + if errors.As(e, &badNodesErr) { err := a.handleBadNodes(ctx, uploadID, spaceDID, badNodesErr) if err != nil { return err @@ -371,39 +375,44 @@ func runScanWorker( span.End() }() - return Worker( + _, err = Worker( ctx, scansAvailable, - - // 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 + 1, + + // findWork + func(ctx context.Context) ([]func(context.Context) (error, error), error) { + return []func(context.Context) (error, error){ + func(ctx context.Context) (error, error) { + if api.AssumeUnchangedSources { + upload, err := api.Repo.GetUploadByID(ctx, uploadID) + if err != nil { + return nil, 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, 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 nil, fmt.Errorf("running scans: %w", err) + } + + return nil, nil + }, + }, nil }, // finalize @@ -412,6 +421,7 @@ func runScanWorker( return nil }, ) + return err } // runDAGScanWorker runs the worker that scans files and directories into blocks, @@ -447,22 +457,27 @@ func runDAGScanWorker( span.End() }() - return Worker( + _, err = Worker( ctx, dagScansAvailable, - - // doWork - func() error { - 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) - } - - return nil + 1, + + // findWork + func(ctx context.Context) ([]func(context.Context) (error, error), error) { + return []func(context.Context) (error, error){ + func(ctx context.Context) (error, error) { + err := api.ExecuteDagScansForUpload(ctx, uploadID, func(node dagmodel.Node, data []byte) error { + signal(nodeUploadsAvailable) + return nil + }) + + if err != nil { + return nil, fmt.Errorf("running dag scans for upload %s: %w", uploadID, err) + } + + return nil, nil + }, + }, nil }, // finalize @@ -488,6 +503,7 @@ func runDAGScanWorker( return nil }, ) + return err } // runShardingWorker runs the worker that assigns nodes to shards. @@ -527,17 +543,22 @@ func runShardingWorker( return nil } - return Worker( + _, err = Worker( 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) ([]func(context.Context) (error, error), error) { + return []func(context.Context) (error, error){ + func(ctx context.Context) (error, error) { + err := api.AddNodesToUploadShards(ctx, uploadID, spaceDID, handleClosedShard) + if err != nil { + return nil, fmt.Errorf("adding nodes to shards for upload %s: %w", uploadID, err) + } + return nil, nil + }, + }, nil }, // finalize @@ -552,6 +573,7 @@ func runShardingWorker( return nil }, ) + return err } func runIndexingWorker( @@ -590,17 +612,22 @@ func runIndexingWorker( return nil } - return Worker( + _, err = Worker( 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) ([]func(context.Context) (error, error), error) { + return []func(context.Context) (error, error){ + func(ctx context.Context) (error, error) { + err := api.AddShardsToUploadIndexes(ctx, uploadID, handleClosedIndex) + if err != nil { + return nil, fmt.Errorf("adding shards to indexes for upload %s: %w", uploadID, err) + } + return nil, nil + }, + }, nil }, // finalize @@ -615,6 +642,7 @@ func runIndexingWorker( return nil }, ) + return err } // runShardUploadWorker runs the worker that adds shards to Storacha. @@ -649,21 +677,35 @@ func runShardUploadWorker( span.End() }() - return Worker( + var inFlightShards sync.Map + + nonFatals, err := Worker( 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) ([]func(context.Context) (error, error), 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 []func(context.Context) (error, error) + 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) (error, error) { + defer inFlightShards.Delete(raw.ID) + nonFatal, fatal := raw.Run(ctx) + if nonFatal == nil && fatal == nil { + signal(uploadedShardsAvailable) + } + return nonFatal, fatal + }) + } + return tasks, nil }, // finalize @@ -672,6 +714,8 @@ func runShardUploadWorker( return nil }, ) + + return errors.Join(err, gtypes.NewBlobUploadErrors(nonFatals)) } func runPostProcessShardWorker( @@ -704,17 +748,30 @@ func runPostProcessShardWorker( span.End() }() - return Worker( + var inFlightShards sync.Map + + _, err = Worker( ctx, uploadedShardsAvailable, + api.BlobUploadParallelism, - // doWork - func() error { - err := api.PostProcessUploadedShards(ctx, uploadID, spaceDID) + // findWork + func(ctx context.Context) ([]func(context.Context) (error, error), 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 []func(context.Context) (error, error) + for _, raw := range rawTasks { + if _, loaded := inFlightShards.LoadOrStore(raw.ID, struct{}{}); loaded { + continue + } + tasks = append(tasks, func(ctx context.Context) (error, error) { + defer inFlightShards.Delete(raw.ID) + return raw.Run(ctx) + }) + } + return tasks, nil }, // finalize @@ -723,10 +780,10 @@ func runPostProcessShardWorker( if err != nil { return fmt.Errorf("`upload/add`ing upload %s: %w", uploadID, err) } - return nil }, ) + return err } // runIndexUploadWorker runs the worker that adds indexes to Storacha. @@ -761,20 +818,35 @@ func runIndexUploadWorker( span.End() }() - return Worker( + var inFlightIndexes sync.Map + + nonFatals, err := Worker( 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) ([]func(context.Context) (error, error), 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 []func(context.Context) (error, error) + 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) (error, error) { + defer inFlightIndexes.Delete(raw.ID) + nonFatal, fatal := raw.Run(ctx) + if nonFatal == nil && fatal == nil { + signal(uploadedIndexesAvailable) + } + return nonFatal, fatal + }) + } + return tasks, nil }, // finalize @@ -783,6 +855,8 @@ func runIndexUploadWorker( return nil }, ) + + return errors.Join(err, gtypes.NewBlobUploadErrors(nonFatals)) } func runPostProcessIndexWorker( @@ -815,20 +889,34 @@ func runPostProcessIndexWorker( span.End() }() - return Worker( + var inFlightIndexes sync.Map + + _, err = Worker( ctx, uploadedIndexesAvailable, + api.BlobUploadParallelism, - // doWork - func() error { - err := api.PostProcessUploadedIndexes(ctx, uploadID, spaceDID) + // findWork + func(ctx context.Context) ([]func(context.Context) (error, error), 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 []func(context.Context) (error, error) + for _, raw := range rawTasks { + if _, loaded := inFlightIndexes.LoadOrStore(raw.ID, struct{}{}); loaded { + continue + } + tasks = append(tasks, func(ctx context.Context) (error, error) { + defer inFlightIndexes.Delete(raw.ID) + return raw.Run(ctx) + }) + } + return tasks, nil }, // finalize nil, ) + return err } diff --git a/pkg/preparation/uploads/worker.go b/pkg/preparation/uploads/worker.go index fe1147d4..d94a6bd0 100644 --- a/pkg/preparation/uploads/worker.go +++ b/pkg/preparation/uploads/worker.go @@ -3,26 +3,98 @@ package uploads import ( "context" "fmt" + "sync" "github.com/storacha/guppy/internal/ctxutil" + "golang.org/x/sync/errgroup" ) -func Worker(ctx context.Context, in <-chan struct{}, doWork func() error, finalize func() error) error { +func Worker( + ctx context.Context, + workAvailable <-chan struct{}, + parallelism int, + findWork func(ctx context.Context) ([]func(context.Context) (error, error), error), + finalize func() error, +) ([]error, error) { + var ( + queue []func(context.Context) (error, error) + nonFatalErrorsMu sync.Mutex + nonFatalErrors []error + ) + + // gctx is cancelled when any task returns a fatal error, allowing the outer + // loop to detect failure and stop dispatching. Tasks receive the outer ctx + // (not gctx) so sibling tasks are not cancelled when one fails. + sem := make(chan struct{}, parallelism) + eg, gctx := errgroup.WithContext(ctx) + + dispatchNext := func() bool { + if len(queue) == 0 { + return false + } + task := queue[0] + queue = queue[1:] + select { + case sem <- struct{}{}: + case <-gctx.Done(): + return false + } + eg.Go(func() error { + defer func() { <-sem }() + nonFatal, fatal := task(ctx) + if fatal != nil { + return fmt.Errorf("worker task encountered a fatal error: %w", fatal) + } + if nonFatal != nil { + nonFatalErrorsMu.Lock() + nonFatalErrors = append(nonFatalErrors, nonFatal) + nonFatalErrorsMu.Unlock() + } + return nil + }) + return true + } + for { select { - case <-ctx.Done(): - return ctxutil.Cause(ctx) - case _, ok := <-in: + case <-gctx.Done(): + // gctx is cancelled either because ctx was cancelled (external stop) + // or because a task returned a fatal error (internal failure). Wait + // for all in-flight tasks to finish before determining which it was. + fatalErr := eg.Wait() + if fatalErr != nil { + return nonFatalErrors, fatalErr + } + // No task error โ€” must be an external cancellation. + return nonFatalErrors, ctxutil.Cause(ctx) + case _, ok := <-workAvailable: if !ok { + // Drain the queue before finalizing. + for dispatchNext() { + } + if fatalErr := eg.Wait(); fatalErr != nil { + return nonFatalErrors, fatalErr + } if finalize != nil { if err := finalize(); err != nil { - return fmt.Errorf("worker finalize encountered an error: %w", err) + return nonFatalErrors, fmt.Errorf("worker finalize encountered an error: %w", err) } } - return nil + if len(nonFatalErrors) > 0 { + return nonFatalErrors, nil + } + return nil, nil + } + + tasks, err := findWork(ctx) + if err != nil { + _ = eg.Wait() + return nonFatalErrors, fmt.Errorf("worker findWork encountered an error: %w", err) } - if err := doWork(); err != nil { - return fmt.Errorf("worker encountered an error: %w", err) + queue = append(queue, tasks...) + + // Fill available parallelism slots. + for dispatchNext() { } } } diff --git a/pkg/preparation/uploads/worker_test.go b/pkg/preparation/uploads/worker_test.go index ae4b1a50..84d8fcb1 100644 --- a/pkg/preparation/uploads/worker_test.go +++ b/pkg/preparation/uploads/worker_test.go @@ -1,6 +1,7 @@ package uploads_test import ( + "context" "errors" "testing" "time" @@ -10,11 +11,6 @@ import ( "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)) { @@ -22,23 +18,36 @@ func e(t *testing.T, condition func(collect *assert.CollectT)) { require.EventuallyWithT(t, condition, time.Second, 10*time.Millisecond) } +func task(fn func() error) func(context.Context) (error, error) { + return func(ctx context.Context) (error, error) { + return nil, 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 error, 1) + type result struct { + nonFatals []error + err error + } + resultChan := make(chan result, 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 - }) + nonFatals, err := uploads.Worker(t.Context(), signalChan, 1, + func(ctx context.Context) ([]func(context.Context) (error, error), error) { + runs++ + return nil, nil + }, + func() error { + finalizes++ + return nil + }, + ) + resultChan <- result{nonFatals, err} }() require.Equal(t, 0, runs, "worker should not run before signal") @@ -47,36 +56,46 @@ func TestWorker(t *testing.T) { 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") + 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") }) - result := <-resultChan - require.Nil(t, result, "result should be nil after successful runs") + res := <-resultChan + require.Nil(t, res.err, "error should be nil after successful runs") + require.Empty(t, res.nonFatals, "non-fatal errors should be empty 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) + type result struct { + nonFatals []error + err error + } + resultChan := make(chan result, 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 - }) + nonFatals, err := uploads.Worker(t.Context(), signalChan, 1, + func(ctx context.Context) ([]func(context.Context) (error, error), error) { + runs++ + if runs == 2 { + return []func(context.Context) (error, error){ + task(func() error { return workerErr }), + }, nil + } + return nil, nil + }, + func() error { + finalizes++ + return nil + }, + ) + resultChan <- result{nonFatals, err} }() // Send three signals; the second should cause an error, the third should not run @@ -84,28 +103,35 @@ func TestWorker(t *testing.T) { 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") + res := <-resultChan + require.ErrorContains(t, res.err, "worker task encountered a fatal error: error in doWork") + require.ErrorIs(t, res.err, 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("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) + type result struct { + nonFatals []error + err error + } + resultChan := make(chan result, 1) var runs int go func() { defer close(resultChan) - resultChan <- uploads.Worker(t.Context(), signalChan, func() error { - runs++ - return nil - }, func() error { - return finalizerErr - }) + nonFatals, err := uploads.Worker(t.Context(), signalChan, 1, + func(ctx context.Context) ([]func(context.Context) (error, error), error) { + runs++ + return nil, nil + }, + func() error { + return finalizerErr + }, + ) + resultChan <- result{nonFatals, err} }() // Send three signals; all should run @@ -114,32 +140,39 @@ func TestWorker(t *testing.T) { 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") + res := <-resultChan + require.ErrorContains(t, res.err, "worker finalize encountered an error: error in finalize") + require.ErrorIs(t, res.err, 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 error, 1) + type result struct { + nonFatals []error + err error + } + resultChan := make(chan result, 1) var ran bool go func() { defer close(resultChan) - - resultChan <- uploads.Worker(t.Context(), signalChan, func() error { - ran = true - return nil - }, nil) + nonFatals, err := uploads.Worker(t.Context(), signalChan, 1, + func(ctx context.Context) ([]func(context.Context) (error, error), error) { + ran = true + return nil, nil + }, + nil, + ) + resultChan <- result{nonFatals, err} }() 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") + res := <-resultChan + require.Nil(t, res.err, "error should be nil after successful runs and no finalizer") + require.Empty(t, res.nonFatals) }) } diff --git a/test/.gitignore b/test/.gitignore index 9f623a4b..f7aa4061 100644 --- a/test/.gitignore +++ b/test/.gitignore @@ -1,4 +1,5 @@ doupload-dir baduploads-dir ipfs -mprocs.log \ No newline at end of file +mprocs.log +smelt \ No newline at end of file diff --git a/test/compose.smelt-override.yml b/test/compose.smelt-override.yml new file mode 100644 index 00000000..bc3c6b4d --- /dev/null +++ b/test/compose.smelt-override.yml @@ -0,0 +1,14 @@ +# This file overrides the Smelt Compose configuration. Paths are relative to +# that file, not this one. + +services: + upload: + environment: + - SPRUE_LOG_LEVEL=debug + volumes: + - ../upload-config:/etc/sprue:ro + +networks: + storacha-network: + # Use an isolated network, not the normal Smelt network. + external: false diff --git a/test/compose.yml b/test/compose.yml new file mode 100644 index 00000000..8906e0f4 --- /dev/null +++ b/test/compose.yml @@ -0,0 +1,44 @@ +name: guppy-test + +include: + - path: + - ./smelt/compose.yml + - ./compose.smelt-override.yml + +services: + guppy-doupload: + extends: + service: guppy + file: ./smelt/systems/guppy/compose.yml + build: + context: .. + dockerfile: Dockerfile + target: dev + image: guppy:dev + environment: + - IN_CONTAINER=true + entrypoint: ["/usr/local/bin/doupload"] + volumes: + - ./doupload:/usr/local/bin/doupload + depends_on: + email-clicker: + condition: service_started + piri-0: + condition: service_healthy + piri-1: + condition: service_healthy + piri-2: + condition: service_healthy + + email-clicker: + image: docker:cli + environment: + - COMPOSE_PROJECT=guppy-test + volumes: + - /var/run/docker.sock:/var/run/docker.sock:ro + - ./email-clicker/click.sh:/usr/local/bin/click.sh:ro + entrypoint: ["/bin/sh", "/usr/local/bin/click.sh"] + depends_on: + - upload + networks: + - storacha-network diff --git a/test/doupload b/test/doupload index 302767f9..b3c59b8a 100755 --- a/test/doupload +++ b/test/doupload @@ -1,4 +1,4 @@ -#!/bin/zsh +#!/bin/bash # Performs an upload and retrieval test using the guppy client CLI. # @@ -12,7 +12,7 @@ # To use a specific schema, add ?search_path= to the URL. # # The test will: -# 1. Create a temporary email inbox using maildrop.cc +# 1. Check out Smelt, initialize it, and move into a container. # 2. Create a new space # 3. Upload some random data to the space # 4. Retrieve the full data @@ -21,308 +21,170 @@ # 7. Retrieve the full data again # 8. Verify that all retrieved data matches the original +# TODO: Postgres support is not yet ported to Smelt/Sprue + set -e set -o pipefail -# Parse arguments -database_url="" -while [[ $# -gt 0 ]]; do - case $1 in - --database-url) - database_url="$2" - shift 2 - ;; - *) - echo "Unknown option: $1" - echo "Usage: test/doupload [--database-url ]" - exit 1 - ;; - esac -done - -# If a database URL was provided, set up the database. -created_db_name="" -if [[ -n "$database_url" ]]; then - # Check for psql - if ! command -v psql &> /dev/null; then - echo "psql could not be found, please install it to use --database-url." - exit 1 - fi - - # Parse the URL to check if a database name is included. - # Split off query string first, then check the path component. - # postgres://user:pass@host:port/dbname?sslmode=disable -> has db name - # postgres://user:pass@host:port?sslmode=disable -> no db name - # postgres://user:pass@host:port/?sslmode=disable -> no db name - # postgres://user:pass@host:port -> no db name - url_base="${database_url%%\?*}" # everything before ? - url_query="" - if [[ "$database_url" == *"?"* ]]; then - url_query="?${database_url#*\?}" # ?sslmode=disable&... - fi - - # After the scheme (postgres://), find the path after the host - after_scheme="${url_base#*://}" # user:pass@host:port/dbname - db_path="${after_scheme#*/}" # dbname (or same as after_scheme if no /) - - # No db name if: no / found (db_path == after_scheme), or path is empty - if [[ "$db_path" == "$after_scheme" ]] || [[ -z "$db_path" ]]; then - # No database name provided; create a temporary one - created_db_name="guppy_test_$(head -c 8 < /dev/urandom | xxd -p)" - # Strip trailing slash from base - url_base="${url_base%/}" - - # We need to connect to the "postgres" database to create a new database. We - # also need to strip out parameters that Go's adapter understands but psql - # doesn't, namely `search_path`. Conveniently, we also don't need to set a - # search_path for this operation. - psql_query=$(echo "$url_query" | sed 's/[?&]search_path=[^&]*//' | sed 's/^&/?/') - postgres_db_url="${url_base}/postgres${psql_query}" - - psql "$postgres_db_url" -c "CREATE DATABASE ${created_db_name};" >/dev/null - database_url="${url_base}/${created_db_name}${url_query}" - echo "Created temporary database: ${created_db_name}" - fi -fi -# Check for dependencies -if ! command -v jq &> /dev/null; then - echo "jq could not be found, please install it to run this script." - exit 1 -fi -if ! command -v htmlq &> /dev/null; then - echo "htmlq could not be found, please install it to run this script." - exit 1 -fi - -# Change to the directory of this script -cd "$(dirname "$0")" - -# Teeing to /dev/fd/3 will show output on stdout while still capturing it. -exec 3>&1 - - -sandbox="doupload-dir" - -go build -gcflags="all=-N -l" .. || { echo "Failed to build guppy"; exit 1; } - -export STORACHA_SERVICE_URL="https://staging.up.warm.storacha.network" -export STORACHA_SERVICE_DID="did:web:staging.up.warm.storacha.network" -export STORACHA_RECEIPTS_URL="https://staging.up.warm.storacha.network/receipt/" -export STORACHA_INDEXING_SERVICE_URL="https://staging.indexer.warm.storacha.network" -export STORACHA_INDEXING_SERVICE_DID="did:web:staging.indexer.warm.storacha.network" -export GUPPY_REPO_DATA_DIR="./$sandbox/storacha" -export GUPPY_REPO_DATABASE_URL="$database_url" - -dataDir="$sandbox/data" -outDir1="$sandbox/out1" -outDir2="$sandbox/out2" -outDir3="$sandbox/out3" - -# Track background processes to kill on exit -cleanup_pids=() - -# Cleanup function to kill background processes and their children -cleanup() { - local pid= - for pid in "${cleanup_pids[@]}"; do - if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then - echo "Cleaning up background process $pid and its children" - # Kill all child processes first - pkill -P "$pid" 2>/dev/null || true - # Then kill the parent - kill "$pid" 2>/dev/null || true - # Wait a moment, then force kill if still alive - sleep 0.1 - if kill -0 "$pid" 2>/dev/null; then - pkill -9 -P "$pid" 2>/dev/null || true - kill -9 "$pid" 2>/dev/null || true - fi - fi - done - - # Drop temporary database if we created one - if [[ -n "$created_db_name" ]]; then - echo "Dropping temporary database: ${created_db_name}" - psql "$postgres_db_url" -c "DROP DATABASE IF EXISTS ${created_db_name};" >/dev/null 2>&1 || true - fi +runGuppyContainer () { + case $1 in + down) + echo "Taking down containers..." + docker compose down + echo "Removing volumes..." + docker volume prune --force --all --filter "label=com.docker.compose.project=guppy-test" + exit $? + ;; + shell) + service="${2:-guppy}" + echo "Starting a shell in the container..." + docker compose run --rm --entrypoint="/bin/sh" "$service" + exit $? + ;; + *) + # Start up the `piri` services first to watch them. + upServices $(docker compose config --services | grep piri- | sort) + + echo "๐Ÿงน Resetting data-dir" + docker volume prune --force --all --filter "label=com.docker.compose.project=guppy-test" --filter "label=com.docker.compose.volume=guppy-data" >/dev/null + + [ -d smelt/.git ] || git clone https://github.com/storacha/smelt.git smelt + + echo "๐Ÿ Initializing smelt" + export SMELT_MANIFEST="$PWD/smelt-manifest.yml" + (cd smelt && make init >/dev/null) + + # docker compose run --rm --entrypoint="true" guppy-doupload + + + + echo "๐Ÿณ Moving into the container..." + echo + docker compose run --rm --build --quiet-build guppy-doupload + exit $? + ;; + esac } -# Set up trap to cleanup on EXIT, INT, TERM -trap cleanup EXIT INT TERM - -# Track the last command for error reporting -last_command="" -trap 'last_command=$ZSH_DEBUG_CMD' DEBUG - -# Error handler -handle_error() { - local exit_code=$1 - local failed_command=$2 - echo - echo "โŒ Command failed with exit code $exit_code: $failed_command" -} +runTests () { + # Teeing to /dev/fd/3 will show output on stdout while still capturing it. + exec 3>&1 -# Handle errors with some printed output -trap 'handle_error $? "$last_command"' ERR + # HTTPS is not available in Smelt, so allow insecure DID resolution. + export GUPPY_NETWORK_INSECURE_DID_RESOLUTION=true -main () { - rm -rf "$sandbox" - mkdir -p "$sandbox" + account="racha@hen.house" - go run github.com/storacha/randdir@latest --output "$dataDir/large-files" --size 50MB --min-file-size 10MB - go run github.com/storacha/randdir@latest --output "$dataDir/small-files" --size 5MB --min-file-size 100KB --max-file-size 1MB + echo + echo "๐Ÿญ Generating random data" + randdir --output "data/large-files" --size 50MB --min-file-size 10MB + randdir --output "data/small-files" --size 5MB --min-file-size 100KB --max-file-size 1MB - # Generate a random maildrop email address - local random_id=$(head -c 10 < /dev/urandom | base32 | tr "[:upper:]" "[:lower:]") - local account="${random_id}@maildrop.cc" + echo echo "๐Ÿ” Logging in as $account" # Log in - log_in "$account" + guppy login "$account" echo echo "๐ŸŽ Generating new space" - space=$(./guppy space generate | tee /dev/fd/3) + space=$(guppy space generate | tee /dev/fd/3) - echo - echo "๐Ÿ“œ Listing space info" - ./guppy space info "$space" + # (Currently not supported in Smelt/Sprue) + # echo + # echo "๐Ÿ“œ Listing space info" + # guppy space info "$space" echo - echo "๐Ÿ“ค Uploading data from $dataDir to space $space" - ./guppy upload source add "$space" "$dataDir" - rootCID=$(./guppy upload "$space" | tee /dev/fd/3 | grep 'Upload completed successfully:' | awk '{print $4}') + echo "๐Ÿ“ค Uploading data from data/ to space $space" + guppy upload source add "$space" "data" + rootCID=$(guppy upload "$space" | tee /dev/fd/3 | grep 'Upload completed successfully:' | awk '{print $4}') echo echo "๐Ÿง Checking local upload state" - ./guppy upload check "$space" + guppy upload check "$space" echo "โœ… Upload state passed checks!" echo echo "๐Ÿ•ต Verifying uploaded data is accessible and consistent" - ./guppy verify "$rootCID" + guppy verify "$rootCID" echo "โœ… Uploaded data verified!" echo - echo "๐Ÿ“ฅ Retrieving data from space $space with root CID $rootCID to $outDir1" - ./guppy retrieve "$space" "$rootCID" "$outDir1" + echo "๐Ÿ“ฅ Retrieving data from space $space with root CID $rootCID to out1" + guppy retrieve "$space" "$rootCID" "out1" echo - echo "๐Ÿ“ฅ Retrieving data from only subdir with root CID $rootCID to $outDir2" - ./guppy retrieve "$space" "$rootCID/small-files" "$outDir2" + echo "๐Ÿ“ฅ Retrieving data from only subdir with root CID $rootCID to out2" + guppy retrieve "$space" "$rootCID/small-files" "out2" + # TODO: Re-login not working yet. - echo - echo "๐Ÿ”„ Resetting client" - ./guppy reset - echo "๐Ÿ” Logging in as $account again" + # echo + # echo "๐Ÿ”„ Resetting client" + # guppy reset + # echo "๐Ÿ” Logging in as $account again" - # Log in again - log_in "$account" + # # Log in again + # guppy login "$account" - # Remove from cleanup list once completed - cleanup_pids=("${cleanup_pids[@]/$login_pid}") - - echo - echo "๐Ÿ“ฅ Retrieving data from space $space with root CID $rootCID to $outDir3" - ./guppy retrieve "$space" "$rootCID" "$outDir3" + # echo + # echo "๐Ÿ“ฅ Retrieving data from space $space with root CID $rootCID to out3" + # guppy retrieve "$space" "$rootCID" "out3" echo "โ†”๏ธ Verifying retrieved data matches original" - diff -r "$dataDir" "$outDir1" - diff -r "$dataDir/small-files" "$outDir2" - diff -r "$dataDir" "$outDir3" + diff -r "data" "out1" + diff -r "data/small-files" "out2" + # diff -r "data" "out3" echo "โœ… Retrieval verified!" - jq -n \ - --arg account "$account" \ - --arg space "$space" \ - --arg rootCID "$rootCID" \ - --arg dataDir "$dataDir" \ - --arg subdir "subdir" \ - '{$account, $space, $rootCID, $dataDir, $subdir}' > "$sandbox/test-params.json" + # TODO: + # jq -n \ + # --arg account "$account" \ + # --arg space "$space" \ + # --arg rootCID "$rootCID" \ + # --arg dataDir "$dataDir" \ + # --arg subdir "subdir" \ + # '{$account, $space, $rootCID, $dataDir, $subdir}' > "$sandbox/test-params.json" } -log_in() { - local account="$1" - # Start login in background - ./guppy login "$account" >/dev/null & - login_pid=$! - cleanup_pids+=("$login_pid") +upServices () { + if ! command -v mprocs &> /dev/null; then + echo "To see logs while waiting, install \`mprocs\`." + docker compose up -d --wait "$@" + return + fi - # Verify email - verify_email "$account" + local mprocsPort + local config + + mprocsPort=$((5000 + RANDOM % 1000)) + config=$(mktemp) && mv "$config" "$config.json" && config="$config.json" + jq \ + '{ + server: "\($mprocsServer)", + procs: { + "Starting...": "docker compose up -d --wait \($ARGS.positional | join(" ")) && mprocs --server \"\($mprocsServer)\" --ctl \"{c: quit}\"" + } + ($ARGS.positional | INDEX(.[]; " " + .) | map_values({shell: "docker compose logs -f --no-log-prefix \(.)", stop: {"send-keys": [""]}})) + }' \ + -n \ + --arg mprocsServer "127.0.0.1:$mprocsPort" \ + --args "$@" \ + >"$config" + + mprocs --config "$config" + rm "$config" +} - # Wait for login to complete - wait $login_pid - # Remove from cleanup list once completed - cleanup_pids=("${cleanup_pids[@]/$login_pid}") -} +# Bootstrap this script into a `guppy` service container. +if [[ -z "$IN_CONTAINER" ]]; then + # Change to the directory of this script + cd "$(dirname "$0")" -seen_message_ids=() - -# Function to check maildrop inbox and click verification link -verify_email() { - local email="$1" - local inbox_name="${email%%@*}" - - echo "โณ Waiting for verification email..." - local max_attempts=30 - local attempt=0 - - while [ $attempt -lt $max_attempts ]; do - # Fetch inbox from Maildrop using GraphQL API - local inbox_query='{"query":"query { inbox(mailbox:\"'${inbox_name}'\") { id subject } }"}' - local inbox_response=$(curl -s -X POST \ - -H 'content-type: application/json' \ - --url https://api.maildrop.cc/graphql \ - --data "$inbox_query") - - # Find the first message that we haven't seen yet - local message_id="" - local all_message_ids=($(jq -r '.data.inbox[].id // empty' <<< "$inbox_response")) - - for id in "${all_message_ids[@]}"; do - # Check if this ID is in the seen list - if [[ ! " ${seen_message_ids[@]} " =~ " ${id} " ]]; then - message_id="$id" - seen_message_ids+=("$message_id") - break - fi - done - - if [ -n "$message_id" ]; then - echo "๐Ÿ“ฌ Found email (ID: $message_id), retrieving verification link..." - - # Fetch the message content (HTML) using GraphQL - local message_query='{"query":"query { message(mailbox:\"'${inbox_name}'\", id:\"'${message_id}'\") { html } }"}' - local message_response=$(curl -s -X POST \ - -H 'content-type: application/json' \ - --url https://api.maildrop.cc/graphql \ - --data "$message_query") - - # Extract the HTML content from the GraphQL response and decode it - local message_html=$(jq -r '.data.message.html' <<< "$message_response") - - # Extract all https links from the HTML - local verify_url=$(htmlq --attribute href a.button <<< "$message_html") - - echo "โœ… Submitting approval form to: $verify_url" - curl -sL -X POST "$verify_url" > /dev/null - - echo "โœ“ Email verified!" - return 0 - else - echo "โš ๏ธ No verification link found in email, retrying..." - fi - - attempt=$((attempt + 1)) - sleep 2 - done - - echo "โŒ Failed to receive verification email after ${max_attempts} attempts" - return 1 -} + runGuppyContainer "$@" +else + runTests +fi -main \ No newline at end of file diff --git a/test/email-clicker/click.sh b/test/email-clicker/click.sh new file mode 100755 index 00000000..47c6c89b --- /dev/null +++ b/test/email-clicker/click.sh @@ -0,0 +1,21 @@ +#!/bin/sh +set -eu + +: "${COMPOSE_PROJECT:?COMPOSE_PROJECT must be set}" + +until id=$(docker ps -q \ + --filter "label=com.docker.compose.project=$COMPOSE_PROJECT" \ + --filter "label=com.docker.compose.service=upload") && [ -n "$id" ]; do + sleep 0.2 +done + +echo "Watching upload container $id for login URLs..." + +docker logs -f --since 0s "$id" 2>&1 | while IFS= read -r line; do + url=$(printf '%s' "$line" | grep -oE '"url"[[:space:]]*:[[:space:]]*"[^"]+"' | sed -E 's/.*"url"[^"]*"([^"]+)".*/\1/' || true) + [ -n "$url" ] || continue + echo "clicking $url" + if ! wget -O /dev/null --post-data='' "$url"; then + echo "verification fetch failed: $url" >&2 + fi +done diff --git a/test/smelt-manifest.yml b/test/smelt-manifest.yml new file mode 100644 index 00000000..a5035d93 --- /dev/null +++ b/test/smelt-manifest.yml @@ -0,0 +1,12 @@ +version: 1 +piri: + nodes: + - storage: + db: sqlite + blob: filesystem + - storage: + db: sqlite + blob: filesystem + - storage: + db: sqlite + blob: filesystem diff --git a/test/upload-config/config.yaml b/test/upload-config/config.yaml new file mode 100644 index 00000000..4b957f04 --- /dev/null +++ b/test/upload-config/config.yaml @@ -0,0 +1,49 @@ +# Upload service configuration for Docker Compose environment + +deployment: + environment: "development" + allow_provision_without_payment_plan: true # allows provision without customer + max_replicas: 3 + +server: + host: "0.0.0.0" + port: 80 # Use port 80 for did:web resolution + public_url: http://upload:80 + +identity: + key_file: "/keys/upload.pem" + service_did: "did:web:upload" + +indexer: + endpoint: "http://indexer:80" + did: "did:web:indexer" + +mailer: + type: "nop" + +dynamodb: + endpoint: "http://dynamodb-local:8000" + region: "us-west-1" + agent_index_table: "agent-index" + blob_registry_table: "blob-registry" + consumer_table: "consumer" + customer_table: "customer" + delegation_table: "delegation" + space_metrics_table: "space-metrics" + admin_metrics_table: "admin-metrics" + replica_table: "replica" + revocation_table: "revocation" + storage_provider_table: "storage-provider" + subscription_table: "subscription" + space_diff_table: "space-diff" + upload_table: "upload" + +s3: + endpoint: "http://minio:9000" + region: "us-west-1" + agent_message_bucket: "agent-message" + delegation_bucket: "delegation" + upload_shards_bucket: "upload-shards" + +log: + level: "info"