diff --git a/forgeclient/blobadd.go b/forgeclient/blobadd.go index d27c76d..ad9c805 100644 --- a/forgeclient/blobadd.go +++ b/forgeclient/blobadd.go @@ -44,6 +44,7 @@ import ( "github.com/ipfs/go-cid" "github.com/multiformats/go-multihash" "go.uber.org/zap" + "golang.org/x/sync/errgroup" ) // BlobAddOption configures [Client.BlobAdd]. @@ -335,6 +336,141 @@ func (c *Client) BlobConclude(ctx context.Context, space did.DID, added AddedBlo }, nil } +// ConcludeResult is one blob's outcome from BlobConcludeBatch: the completed +// AddedBlob (Location set, PutInvocation dropped) or the error that kept it +// parked. Failed entries stay parked and are safe to resubmit. +type ConcludeResult struct { + Blob AddedBlob + Err error +} + +// BlobConcludeBatch finishes many parked BlobAdds in ONE round trip to the +// upload service: every parked blob's synthesized /http/put receipt and its +// /ucan/conclude invocation travel in a single UCAN container, which the +// service's ucantone server executes invocation-by-invocation, returning all +// the conclude receipts in one response container. Because the service runs +// each /blob/accept synchronously before answering, the follow-up accept +// receipt fetches (for the location commitments) succeed on their first +// attempt; they run bounded-parallel here. Outcomes are positional: one +// failed conclude does not fail the batch, and a transport-level error fails +// the whole call with no per-blob results (everything stays parked; conclude +// is idempotent, so resubmitting any subset is safe). +// +// Callers should chunk very large upload sessions (a put receipt + conclude +// invocation is a few KB, and the service concludes a container's invocations +// sequentially while the request hangs open) — see s3frontend's Complete for +// the chunking policy. +func (c *Client) BlobConcludeBatch(ctx context.Context, space did.DID, added []AddedBlob) ([]ConcludeResult, error) { + results := make([]ConcludeResult, len(added)) + var invs []ucan.Invocation + var rcpts []ucan.Receipt + idxByTask := make(map[cid.Cid]int) + for i, ab := range added { + if ab.Location != nil { + results[i] = ConcludeResult{Blob: ab} + continue + } + putInv := new(invocation.Invocation) + if err := putInv.UnmarshalCBOR(bytes.NewReader(ab.PutInvocation)); err != nil { + results[i] = ConcludeResult{Err: fmt.Errorf("decoding parked /http/put invocation: %w", err)} + continue + } + putRcpt, err := synthesizePutReceipt(putInv) + if err != nil { + results[i] = ConcludeResult{Err: fmt.Errorf("synthesizing put receipt: %w", err)} + continue + } + inv, err := ucancmds.Conclude.Invoke( + c.signer, c.signer.DID(), + &ucancmds.ConcludeArguments{Receipt: putRcpt.Link()}, + invocation.WithAudience(c.serviceID), + ) + if err != nil { + results[i] = ConcludeResult{Err: fmt.Errorf("generating conclude invocation: %w", err)} + continue + } + idxByTask[inv.Task().Link()] = i + invs = append(invs, inv) + rcpts = append(rcpts, putRcpt) + } + if len(invs) == 0 { + return results, nil + } + + start := time.Now() + // The stock client sends one primary invocation; the rest ride the same + // request container via WithInvocations and are executed all the same. + // The response's metadata is the full response container, so every + // conclude's receipt is recovered from it by task link. + resp, err := c.ucanClient.Execute(execution.NewRequest(ctx, invs[0], + execution.WithInvocations(invs[1:]...), + execution.WithReceipts(rcpts...), + )) + if err != nil { + c.logger.Error("blob conclude batch failed", + zap.Stringer("space", space), zap.Int("blobs", len(invs)), + zap.Error(err), zap.Duration("duration", time.Since(start))) + return nil, fmt.Errorf("executing conclude batch: %w", err) + } + container := resp.Metadata() + + // Match each conclude's receipt out of the response container, then fetch + // the accept receipts (location commitments) for the successes in bounded + // parallel. + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(concludeFetchConcurrency) + for _, inv := range invs { + i := idxByTask[inv.Task().Link()] + rcpt := maybeFindReceipt(inv.Task().Link(), container.Receipts()) + if rcpt == nil { + results[i] = ConcludeResult{Err: fmt.Errorf("conclude receipt missing for blob %s", added[i].Digest)} + continue + } + if rcpt.Out().IsErr() { + _, x := rcpt.Out().Unpack() + var model edm.ErrorModel + if err := model.UnmarshalCBOR(bytes.NewReader(x)); err != nil { + results[i] = ConcludeResult{Err: fmt.Errorf("conclude failed with undecodable error")} + continue + } + results[i] = ConcludeResult{Err: fmt.Errorf("conclude failed: %w", model)} + continue + } + g.Go(func() error { + location, aerr := c.awaitAccept(gctx, added[i].AcceptTask) + if aerr != nil { + results[i] = ConcludeResult{Err: aerr} + return nil + } + results[i] = ConcludeResult{Blob: AddedBlob{ + Digest: added[i].Digest, + Size: added[i].Size, + Location: location, + AddTask: added[i].AddTask, + AcceptTask: added[i].AcceptTask, + }} + return nil + }) + } + _ = g.Wait() + + failed := 0 + for _, r := range results { + if r.Err != nil { + failed++ + } + } + c.logger.Debug("blob conclude batch", + zap.Stringer("space", space), zap.Int("blobs", len(invs)), + zap.Int("failed", failed), zap.Duration("duration", time.Since(start))) + return results, nil +} + +// concludeFetchConcurrency bounds BlobConcludeBatch's parallel accept-receipt +// fetches. The receipts already exist by the time the batch returns, so these +// are quick single GETs — the bound just keeps a large chunk polite. +const concludeFetchConcurrency = 16 + // awaitAccept polls the /blob/accept receipt and extracts the // /assert/location commitment from its metadata. func (c *Client) awaitAccept(ctx context.Context, acceptTask cid.Cid) (ucan.Invocation, error) { @@ -389,35 +525,46 @@ func putBlob(ctx context.Context, client *http.Client, url *url.URL, headers map return nil } -func (c *Client) sendPutReceipt(ctx context.Context, putInv ucan.Invocation, opts ...execution.RequestOption) error { +// synthesizePutReceipt issues the /http/put success receipt for a parked put +// invocation, signing with the derived key embedded in the invocation's +// metadata (the same key piri expects the receipt from). +func synthesizePutReceipt(putInv ucan.Invocation) (ucan.Receipt, error) { var putMeta datamodel.Map if err := putMeta.UnmarshalCBOR(bytes.NewReader(putInv.MetadataBytes())); err != nil { - return fmt.Errorf("unmarshaling /http/put invocation metadata: %w", err) + return nil, fmt.Errorf("unmarshaling /http/put invocation metadata: %w", err) } keysMap, ok := putMeta["keys"].(ipld.Map) if !ok { - return fmt.Errorf("invalid put metadata, missing 'keys' field") + return nil, fmt.Errorf("invalid put metadata, missing 'keys' field") } id, ok := keysMap["id"].(string) if !ok { - return fmt.Errorf("invalid put metadata, missing 'id' field in 'keys'") + return nil, fmt.Errorf("invalid put metadata, missing 'id' field in 'keys'") } keysKeysMap, ok := keysMap["keys"].(ipld.Map) if !ok { - return fmt.Errorf("invalid put metadata, missing 'keys' field in 'keys'") + return nil, fmt.Errorf("invalid put metadata, missing 'keys' field in 'keys'") } keyBytes, ok := keysKeysMap[id].([]byte) if !ok { - return fmt.Errorf("invalid put metadata, missing key for %s", id) + return nil, fmt.Errorf("invalid put metadata, missing key for %s", id) } signer, err := ed25519.Decode(keyBytes) if err != nil { - return fmt.Errorf("decoding key for %q: %w", id, err) + return nil, fmt.Errorf("decoding key for %q: %w", id, err) } issuer := multikey.KeyIssuer(signer) putRcpt, err := receipt.IssueOK(issuer, putInv.Task().Link(), &httpcmds.PutOK{}, receipt.WithIssuedAt(ucan.Now())) if err != nil { - return fmt.Errorf("generating receipt: %w", err) + return nil, fmt.Errorf("generating receipt: %w", err) + } + return putRcpt, nil +} + +func (c *Client) sendPutReceipt(ctx context.Context, putInv ucan.Invocation, opts ...execution.RequestOption) error { + putRcpt, err := synthesizePutReceipt(putInv) + if err != nil { + return err } inv, err := ucancmds.Conclude.Invoke( diff --git a/forgeclient/blobadd_batch_test.go b/forgeclient/blobadd_batch_test.go new file mode 100644 index 0000000..ce30a7d --- /dev/null +++ b/forgeclient/blobadd_batch_test.go @@ -0,0 +1,194 @@ +package forgeclient + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "testing" + + assertcmds "github.com/fil-forge/libforge/commands/assert" + blobcmds "github.com/fil-forge/libforge/commands/blob" + "github.com/fil-forge/libforge/commands" + ucancmds "github.com/fil-forge/libforge/commands/ucan" + "github.com/fil-forge/ucantone/binding" + "github.com/fil-forge/ucantone/ipld/codec/dagcbor" + "github.com/fil-forge/ucantone/ipld/datamodel" + "github.com/fil-forge/ucantone/multikey" + "github.com/fil-forge/ucantone/multikey/ed25519" + "github.com/fil-forge/ucantone/server" + "github.com/fil-forge/ucantone/ucan/command" + "github.com/fil-forge/ucantone/ucan/container" + "github.com/fil-forge/ucantone/ucan/invocation" + "github.com/fil-forge/ucantone/ucan/promise" + "github.com/fil-forge/ucantone/ucan/receipt" + "github.com/ipfs/go-cid" + "github.com/multiformats/go-multihash" +) + +// TestBlobConcludeBatchSingleRequest pins BlobConcludeBatch's core assumption +// against the real ucantone stack: N parked blobs conclude through ONE POST +// to the service — every conclude invocation and put receipt in one UCAN +// container, all executed by the server's container loop, all conclude +// receipts recovered from the one response — followed only by the per-blob +// accept-receipt fetches. +func TestBlobConcludeBatchSingleRequest(t *testing.T) { + ctx := context.Background() + agent, err := ed25519.GenerateIssuer() + if err != nil { + t.Fatalf("generate agent: %v", err) + } + svc, err := ed25519.GenerateIssuer() + if err != nil { + t.Fatalf("generate service: %v", err) + } + spaceIss, err := ed25519.GenerateIssuer() + if err != nil { + t.Fatalf("generate space: %v", err) + } + space := spaceIss.DID() + + // The service: a stock ucantone UCAN server with a conclude route that + // records each executed invocation and checks its named put receipt is + // present in the request container. + var mu sync.Mutex + var concluded []cid.Cid + var missingReceipts int + ucanSrv := server.NewHTTP(svc) + route := ucancmds.Conclude.Route(func(req *binding.Request[*ucancmds.ConcludeArguments], res *binding.Response[*ucancmds.ConcludeOK]) error { + found := false + for _, r := range req.Metadata().Receipts() { + if r.Link() == req.Task().Arguments().Receipt { + found = true + break + } + } + mu.Lock() + concluded = append(concluded, req.Task().Arguments().Receipt) + if !found { + missingReceipts++ + } + mu.Unlock() + return res.SetSuccess(&ucancmds.ConcludeOK{}) + }) + ucanSrv.Handle(route.Command, route.Handler) + + // Three parked blobs: a dummy put invocation per blob carrying the + // derived signer key in its metadata (what synthesizePutReceipt reads), + // and a pre-staged accept receipt + location commitment served by the + // receipt endpoint (sprue stores these before answering the conclude, + // so the first fetch always succeeds). + receiptBodies := map[string][]byte{} + var added []AddedBlob + for _, name := range []string{"one", "two", "three"} { + digest, err := multihash.Sum([]byte("blob-"+name), multihash.SHA2_256, -1) + if err != nil { + t.Fatalf("digest: %v", err) + } + derived, err := ed25519.Generate() + if err != nil { + t.Fatalf("generate derived key: %v", err) + } + keyID := multikey.KeyIssuer(derived).DID().String() + putInv, err := invocation.Invoke(agent, agent.DID(), command.MustParse("/http/put"), nil, + invocation.WithMetadata(datamodel.Map{ + "keys": datamodel.Map{"id": keyID, "keys": datamodel.Map{keyID: ed25519.Encode(derived)}}, + })) + if err != nil { + t.Fatalf("build put invocation: %v", err) + } + addTask := cid.NewCidV1(cid.Raw, digest) + acceptDigest, err := multihash.Sum([]byte("accept-"+name), multihash.SHA2_256, -1) + if err != nil { + t.Fatalf("accept digest: %v", err) + } + acceptTask := cid.NewCidV1(cid.Raw, acceptDigest) + added = append(added, AddedBlob{ + Digest: digest, + Size: 64, + AddTask: addTask, + AcceptTask: acceptTask, + PutInvocation: putInv.Bytes(), + }) + + accRcpt, err := receipt.IssueOK(svc, acceptTask, &blobcmds.AcceptOK{ + Site: addTask, + PDP: promise.AwaitOK{Task: addTask}, + }) + if err != nil { + t.Fatalf("issue accept receipt: %v", err) + } + locURL, err := url.Parse("http://piri.test/blob/" + name) + if err != nil { + t.Fatalf("parse loc url: %v", err) + } + locInv, err := assertcmds.Location.Invoke(svc, space, &assertcmds.LocationArguments{ + Space: space, + Content: digest, + Location: []commands.CborURL{commands.CborURL(*locURL)}, + }) + if err != nil { + t.Fatalf("build location commitment: %v", err) + } + ct := container.New(container.WithReceipts(accRcpt), container.WithInvocations(locInv)) + var buf strings.Builder + if err := ct.MarshalCBOR(&buf); err != nil { + t.Fatalf("encode receipt container: %v", err) + } + receiptBodies[acceptTask.String()] = []byte(buf.String()) + } + + var posts int + mux := http.NewServeMux() + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + posts++ + mu.Unlock() + ucanSrv.ServeHTTP(w, r) + }) + mux.HandleFunc("/receipt/", func(w http.ResponseWriter, r *http.Request) { + task := strings.TrimPrefix(r.URL.Path, "/receipt/") + body, ok := receiptBodies[task] + if !ok { + w.WriteHeader(http.StatusNotFound) + return + } + w.Header().Set("Content-Type", dagcbor.ContentType) + _, _ = w.Write(body) + }) + ts := httptest.NewServer(mux) + t.Cleanup(ts.Close) + + tsURL, err := url.Parse(ts.URL) + if err != nil { + t.Fatalf("parse test server url: %v", err) + } + c, err := New(agent, svc.DID(), *tsURL) + if err != nil { + t.Fatalf("new client: %v", err) + } + + results, err := c.BlobConcludeBatch(ctx, space, added) + if err != nil { + t.Fatalf("BlobConcludeBatch: %v", err) + } + for i, r := range results { + if r.Err != nil { + t.Fatalf("blob %d: %v", i, r.Err) + } + if r.Blob.Location == nil || r.Blob.Location.Command() != assertcmds.Location.Command { + t.Fatalf("blob %d: location commitment missing from result", i) + } + } + if posts != 1 { + t.Fatalf("conclude POSTs = %d, want 1 (the whole batch in one request)", posts) + } + if len(concluded) != len(added) { + t.Fatalf("server executed %d concludes, want %d", len(concluded), len(added)) + } + if missingReceipts != 0 { + t.Fatalf("%d concludes could not find their put receipt in the request container", missingReceipts) + } +} diff --git a/go.mod b/go.mod index 9a05328..78a185c 100644 --- a/go.mod +++ b/go.mod @@ -33,6 +33,7 @@ require ( go.uber.org/fx v1.24.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.28.0 + golang.org/x/sync v0.22.0 golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da ) @@ -240,7 +241,6 @@ require ( golang.org/x/crypto v0.54.0 // indirect golang.org/x/exp v0.0.0-20260603202125-055de637280b // indirect golang.org/x/net v0.57.0 // indirect - golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/term v0.45.0 // indirect golang.org/x/text v0.40.0 // indirect diff --git a/inmem/store.go b/inmem/store.go index ade9c2c..3d21651 100644 --- a/inmem/store.go +++ b/inmem/store.go @@ -406,8 +406,12 @@ func (NopUploader) UploadBlob(_ context.Context, _ did.DID, digest multihash.Mul func (NopUploader) RemoveBlob(_ context.Context, _ did.DID, _ multihash.Multihash) error { return nil } -func (NopUploader) ConcludeBlob(_ context.Context, _ did.DID, parked uploader.UploadedBlob) (uploader.BlobLocation, error) { - return uploader.BlobLocation{Size: parked.Size}, nil +func (NopUploader) ConcludeBlobBatch(_ context.Context, _ did.DID, parked []uploader.UploadedBlob) ([]uploader.ConcludeBlobResult, error) { + results := make([]uploader.ConcludeBlobResult, len(parked)) + for i, p := range parked { + results[i] = uploader.ConcludeBlobResult{Location: uploader.BlobLocation{Size: p.Size}} + } + return results, nil } func (NopUploader) AbortBlob(_ context.Context, _ did.DID, _ multihash.Multihash, _ cid.Cid) error { diff --git a/itest/versity_multipart_test.go b/itest/versity_multipart_test.go index a6498df..4081e9d 100644 --- a/itest/versity_multipart_test.go +++ b/itest/versity_multipart_test.go @@ -166,9 +166,11 @@ var completeMultipartPass = []forgeCase{ {name: "racey_success", fn: integration.CompleteMultipartUpload_racey_success, skip: func() string { return "load-sensitive (10 concurrent 25MiB uploads under a 30s deadline); outcome depends on host load, not S3 semantics" }}, -} - -// racey_data_integrity leans on atomic concurrent overwrites under load. -var completeMultipartXFail = []forgeCase{ + // racey_data_integrity fires five concurrent Completes of ONE upload id + // and requires every one to succeed with the winner's ETag: it pins the + // #69 join — a Complete losing the open→completing latch awaits the + // in-flight peer and replays its result instead of 404ing. {name: "racey_data_integrity", fn: integration.CompleteMultipartUpload_racey_data_integrity}, } + +var completeMultipartXFail = []forgeCase{} diff --git a/s3frontend/multipart.go b/s3frontend/multipart.go index dcdd648..af55a30 100644 --- a/s3frontend/multipart.go +++ b/s3frontend/multipart.go @@ -11,6 +11,7 @@ import ( "io" "strconv" "strings" + "sync" "time" "github.com/aws/aws-sdk-go-v2/service/s3" @@ -24,6 +25,7 @@ import ( "github.com/ipfs/go-cid" mh "github.com/multiformats/go-multihash" "go.uber.org/zap" + "golang.org/x/sync/errgroup" msbucket "github.com/fil-forge/ingot/bucket" "github.com/fil-forge/ingot/internal/reqscope" @@ -35,6 +37,29 @@ import ( // defaultMaxListing is the S3 default and cap for max-parts / max-uploads. const defaultMaxListing = 1000 +// concludeBatchSize / concludeBatchInflight shape Complete's conclude phase +// (issue #69): parked blobs conclude in chunks, each chunk one request to +// the upload service carrying all its conclude invocations in a single UCAN +// container. The service executes a container's invocations sequentially +// while the request hangs open, so the chunk size bounds that hold-open +// time; chunks in flight restore cross-chunk parallelism, taking a 256-part +// upload's accept phase from minutes to seconds without swamping the +// service. +const ( + concludeBatchSize = 16 + concludeBatchInflight = 4 +) + +// completeJoinPoll/completeJoinWait pace a Complete that lost the latch to an +// in-flight Complete of the same upload: the loser polls the session until +// the winner resolves it (then reports the winner's outcome) and yields a +// retryable SlowDown once the budget is spent. Vars so tests can compress +// the schedule. +var ( + completeJoinPoll = 500 * time.Millisecond + completeJoinWait = 2 * time.Minute +) + // newUploadID returns a random 128-bit hex upload id. func newUploadID() (string, error) { var b [16]byte @@ -491,25 +516,66 @@ func (b *Backend) CompleteMultipartUpload(ctx context.Context, input *s3.Complet } } - // Idempotent re-Complete: the prior Complete committed the object; the - // validation above already proved the client's part list matches the - // retained parts, so return the same result without recommitting. - if sess.State == registry.SessionCompleted { + // completedResult answers a Complete whose object is already committed + // (idempotent re-Complete, or a joined peer's commit below): the part + // validation above proved the client's list matches the retained parts, + // so the recomputed ETag/checksum equal the committed ones. + completedResult := func() (s3response.CompleteMultipartUploadResult, string, error) { etagQ := `"` + etag + `"` res := s3response.CompleteMultipartUploadResult{Bucket: &bucket, Key: &key, ETag: &etagQ} setCompleteResultChecksum(&res, ckAlgo, ckValue, ckType) return res, "", nil } - - // Single-winner latch vs a racing Abort: only the writer that moves the - // session off 'open' proceeds (§7.3). - won, err := b.multipart.LatchSession(ctx, uploadID, registry.SessionOpen, registry.SessionCompleting) - if err != nil { - return s3response.CompleteMultipartUploadResult{}, "", fmt.Errorf("s3frontend: latch: %w", err) + if sess.State == registry.SessionCompleted { + return completedResult() } - if !won { - return s3response.CompleteMultipartUploadResult{}, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) + + // Single-winner latch vs a racing Abort or Complete: only the writer that + // moves the session off 'open' proceeds (§7.3). Losing to another + // Complete must NOT map to NoSuchUpload — the upload exists and is + // actively completing, and a client-timeout retry lands exactly here + // (issue #69: a terminal 404 for an upload that then commits) — so the + // loser joins: it awaits the in-flight completion and reports its + // outcome, taking over if the peer failed and reverted the session. + joinDeadline := time.Now().Add(completeJoinWait) + for { + won, err := b.multipart.LatchSession(ctx, uploadID, registry.SessionOpen, registry.SessionCompleting) + if err != nil { + return s3response.CompleteMultipartUploadResult{}, "", fmt.Errorf("s3frontend: latch: %w", err) + } + if won { + break + } + cur, gerr := b.multipart.GetSession(ctx, uploadID) + if gerr != nil { + if errors.Is(gerr, registry.ErrNotFound) { + return s3response.CompleteMultipartUploadResult{}, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) + } + return s3response.CompleteMultipartUploadResult{}, "", fmt.Errorf("s3frontend: session state: %w", gerr) + } + switch cur.State { + case registry.SessionCompleted: + return completedResult() + case registry.SessionAborting: + return s3response.CompleteMultipartUploadResult{}, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) + case registry.SessionOpen: + // The peer failed and reverted; contend for the latch again. + continue + } + // 'completing': the peer is mid-flight. Past the wait budget (the peer + // is wedged, or its process died leaving the row latched until the + // sweeper reaps it) answer with the retryable SlowDown instead of + // holding the connection indefinitely. + if time.Now().After(joinDeadline) { + return s3response.CompleteMultipartUploadResult{}, "", s3err.GetAPIError(s3err.ErrSlowDown) + } + select { + case <-ctx.Done(): + return s3response.CompleteMultipartUploadResult{}, "", fmt.Errorf("s3frontend: awaiting in-flight complete: %w", ctx.Err()) + case <-time.After(completeJoinPoll): + } } + // If anything below fails before the object is committed, revert the session // to 'open' so the upload stays abortable / retriable rather than zombied in // 'completing'. committed is set once the manifest is durable (the point of @@ -847,12 +913,29 @@ func (b *Backend) parkBlobs(ctx context.Context, space did.DID, blobs []msbucket // concludeBlobs is Complete's park-aware counterpart to uploadBlobs: located // blobs are already accepted (dedup); parked blobs conclude their deferred -// /http/put receipt — firing /blob/accept — and record their location; +// /http/put receipts — firing /blob/accept — and record their locations; // blobs that never parked (crash between spool and park) fall back to the // whole synchronous upload. +// +// Parked blobs conclude in chunked BATCHES: each chunk is one request to the +// upload service carrying all its conclude invocations in a single UCAN +// container (sequential per-blob concludes held Complete's connection open +// for O(parts) round trips — minutes at a few hundred parts, issue #69), +// with a few chunks in flight so chunks still overlap. Failures are +// per-blob: every blob that concluded keeps its recorded location, failed +// blobs stay parked, and the retry after the session reverts to 'open' +// resumes where this attempt stopped (conclude is idempotent). func (b *Backend) concludeBlobs(ctx context.Context, space did.DID, blobs []msbucket.BlobRef) error { + // Triage one entry per distinct digest: a content-addressed blob may back + // several parts, and its accept/location/park is per-digest state. + seen := make(map[string]struct{}, len(blobs)) + var parked []uploader.UploadedBlob + var stragglers []msbucket.BlobRef for _, blob := range blobs { - digest := mh.Multihash(blob.Digest) + if _, ok := seen[string(blob.Digest)]; ok { + continue + } + seen[string(blob.Digest)] = struct{}{} if existing, err := b.locations.GetLocation(ctx, space, blob.Digest); err == nil && existing != nil { if err := b.intents.SetIntentState(ctx, blob.Digest, registry.IntentAccepted); err != nil { return fmt.Errorf("mark accepted (dedup): %w", err) @@ -861,61 +944,107 @@ func (b *Backend) concludeBlobs(ctx context.Context, space did.DID, blobs []msbu } else if err != nil && !errors.Is(err, registry.ErrNotFound) { return fmt.Errorf("lookup location: %w", err) } - park, err := b.parks.GetPark(ctx, blob.Digest) if err != nil && !errors.Is(err, registry.ErrNotFound) { return fmt.Errorf("lookup park: %w", err) } - var loc uploader.BlobLocation - if park != nil { - addTask, err := cid.Cast(park.AddTask) - if err != nil { - return fmt.Errorf("decode park add task: %w", err) - } - acceptTask, err := cid.Cast(park.AcceptTask) + if park == nil { + stragglers = append(stragglers, blob) + continue + } + addTask, err := cid.Cast(park.AddTask) + if err != nil { + return fmt.Errorf("decode park add task: %w", err) + } + acceptTask, err := cid.Cast(park.AcceptTask) + if err != nil { + return fmt.Errorf("decode park accept task: %w", err) + } + parked = append(parked, uploader.UploadedBlob{ + Digest: mh.Multihash(blob.Digest), + Size: park.Size, + AddTask: addTask, + AcceptTask: acceptTask, + PutInvocation: park.PutInvocation, + }) + } + + // All chunks and stragglers run to completion regardless of individual + // failures — every additional recorded location is one less conclude for + // the retry — with the errors joined at the end. + var mu sync.Mutex + var errs []error + fail := func(err error) { + mu.Lock() + errs = append(errs, err) + mu.Unlock() + } + var g errgroup.Group + g.SetLimit(concludeBatchInflight) + for start := 0; start < len(parked); start += concludeBatchSize { + chunk := parked[start:min(start+concludeBatchSize, len(parked))] + g.Go(func() error { + results, err := b.deferred.ConcludeBlobBatch(ctx, space, chunk) if err != nil { - return fmt.Errorf("decode park accept task: %w", err) + fail(fmt.Errorf("conclude blobs: %w", err)) + return nil } - loc, err = b.deferred.ConcludeBlob(ctx, space, uploader.UploadedBlob{ - Digest: digest, - Size: park.Size, - AddTask: addTask, - AcceptTask: acceptTask, - PutInvocation: park.PutInvocation, - }) - if err != nil { - return fmt.Errorf("conclude blob: %w", err) + for i, r := range results { + digest := []byte(chunk[i].Digest) + if r.Err != nil { + fail(fmt.Errorf("conclude blob %x: %w", digest, r.Err)) + continue + } + if err := b.recordAccepted(ctx, space, digest, r.Location, true); err != nil { + fail(err) + } } - } else { + return nil + }) + } + for _, blob := range stragglers { + g.Go(func() error { // Never parked (crash between spool and park): the spooled copy // drives the whole synchronous upload. + digest := mh.Multihash(blob.Digest) res, uerr := b.uploader.UploadBlob(ctx, space, digest, blob.Length, b.spool.Path(digest)) if uerr != nil { - return fmt.Errorf("upload blob: %w", uerr) + fail(fmt.Errorf("upload blob: %w", uerr)) + return nil } if res.Location == nil { - return fmt.Errorf("upload blob %x: concluding upload returned no location", blob.Digest) + fail(fmt.Errorf("upload blob %x: concluding upload returned no location", blob.Digest)) + return nil } - loc = *res.Location - } - - if err := b.locations.PutLocation(ctx, registry.BlobLocation{ - Space: space, - Digest: blob.Digest, - Provider: loc.Provider, - URL: loc.URL, - Size: loc.Size, - }); err != nil { - return fmt.Errorf("record location: %w", err) - } - if err := b.intents.SetIntentState(ctx, blob.Digest, registry.IntentAccepted); err != nil { - return fmt.Errorf("mark accepted: %w", err) - } - if park != nil { - // The sealed put invocation is spent — drop it promptly. - if err := b.parks.DeletePark(ctx, blob.Digest); err != nil { - return fmt.Errorf("drop park: %w", err) + if err := b.recordAccepted(ctx, space, blob.Digest, *res.Location, false); err != nil { + fail(err) } + return nil + }) + } + _ = g.Wait() + return errors.Join(errs...) +} + +// recordAccepted persists one accepted blob's outcome: its location, the +// intent transition, and — for a parked blob — dropping the park row (the +// sealed put invocation is spent). +func (b *Backend) recordAccepted(ctx context.Context, space did.DID, digest []byte, loc uploader.BlobLocation, parked bool) error { + if err := b.locations.PutLocation(ctx, registry.BlobLocation{ + Space: space, + Digest: digest, + Provider: loc.Provider, + URL: loc.URL, + Size: loc.Size, + }); err != nil { + return fmt.Errorf("record location: %w", err) + } + if err := b.intents.SetIntentState(ctx, digest, registry.IntentAccepted); err != nil { + return fmt.Errorf("mark accepted: %w", err) + } + if parked { + if err := b.parks.DeletePark(ctx, digest); err != nil { + return fmt.Errorf("drop park: %w", err) } } return nil diff --git a/s3frontend/multipart_complete_test.go b/s3frontend/multipart_complete_test.go new file mode 100644 index 0000000..90e5ac2 --- /dev/null +++ b/s3frontend/multipart_complete_test.go @@ -0,0 +1,325 @@ +package s3frontend + +import ( + "bytes" + "context" + "errors" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/fil-forge/ucantone/did" + "github.com/fil-forge/versitygw/backend" + "github.com/fil-forge/versitygw/s3err" + "github.com/ipfs/go-cid" + "github.com/multiformats/go-multihash" + "go.uber.org/zap/zaptest" + + "github.com/fil-forge/ingot/blockstore" + "github.com/fil-forge/ingot/inmem" + "github.com/fil-forge/ingot/logstore" + "github.com/fil-forge/ingot/registry" + "github.com/fil-forge/ingot/uploader" +) + +// CompleteMultipartUpload race behavior (issue #69): a Complete that loses the +// open→completing latch to another Complete joins the in-flight completion +// instead of answering the terminal NoSuchUpload, and the conclude fan-out +// runs bounded-parallel so Complete stops holding the connection for O(parts) +// sequential round trips. + +// compressJoinSchedule shrinks the completing-join poll/budget for a test and +// restores the defaults on cleanup. +func compressJoinSchedule(t *testing.T, poll, wait time.Duration) { + t.Helper() + oldPoll, oldWait := completeJoinPoll, completeJoinWait + completeJoinPoll, completeJoinWait = poll, wait + t.Cleanup(func() { completeJoinPoll, completeJoinWait = oldPoll, oldWait }) +} + +// mpTwoParts uploads a two-part session (no declared checksum) and returns its +// upload id plus the Complete part list. +func mpTwoParts(t *testing.T, b *Backend, key string) (string, []types.CompletedPart) { + t.Helper() + uploadID := mpCreate(t, b, key, "", "") + var parts []types.CompletedPart + for i, data := range [][]byte{bytes.Repeat([]byte("a"), int(backend.MinPartSize)), []byte("tail")} { + out, err := mpUploadPart(t, b, key, uploadID, int32(i+1), data, nil) + if err != nil { + t.Fatalf("UploadPart %d: %v", i+1, err) + } + n := int32(i + 1) + parts = append(parts, types.CompletedPart{PartNumber: &n, ETag: out.ETag}) + } + return uploadID, parts +} + +// TestCompleteJoinsInflightPeer: a Complete arriving while another Complete +// holds the session in 'completing' (the client-timeout retry window) must +// await the peer and report its outcome — a NoSuchUpload here tells the +// client its upload is gone while that upload is actively committing. +func TestCompleteJoinsInflightPeer(t *testing.T) { + compressJoinSchedule(t, 5*time.Millisecond, 5*time.Second) + b, mem, _ := newRefTestBackend(t) + ctx := context.Background() + key := "join-inflight" + uploadID, parts := mpTwoParts(t, b, key) + + // A peer holds the completing latch... + if won, err := mem.LatchSession(ctx, uploadID, registry.SessionOpen, registry.SessionCompleting); err != nil || !won { + t.Fatalf("latch to completing: won=%v err=%v", won, err) + } + // ...and resolves it to 'completed' a beat after the retry arrives. + go func() { + time.Sleep(50 * time.Millisecond) + _, _ = mem.LatchSession(ctx, uploadID, registry.SessionCompleting, registry.SessionCompleted) + }() + + res, err := mpComplete(t, b, key, uploadID, parts, nil) + if err != nil { + t.Fatalf("Complete during completing window: %v (want joined success)", err) + } + if res.ETag == nil || !strings.HasSuffix(strings.Trim(*res.ETag, `"`), "-2") { + t.Fatalf("joined ETag = %v, want multipart ETag with -2 suffix", res.ETag) + } + // The peer owned the commit; the joiner must replay its outcome, not + // re-commit — the simulated peer wrote no object, so none may exist. + bucket := "bk" + if _, err := b.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: &key}); err == nil { + t.Fatal("joiner committed the object itself; want replay of the peer's outcome") + } +} + +// TestCompleteTakesOverAfterPeerReverts: when the in-flight peer fails and +// reverts the session to 'open', a joining Complete contends for the latch +// again and finishes the upload itself. +func TestCompleteTakesOverAfterPeerReverts(t *testing.T) { + compressJoinSchedule(t, 5*time.Millisecond, 5*time.Second) + b, mem, _ := newRefTestBackend(t) + ctx := context.Background() + key := "join-takeover" + uploadID, parts := mpTwoParts(t, b, key) + + if won, err := mem.LatchSession(ctx, uploadID, registry.SessionOpen, registry.SessionCompleting); err != nil || !won { + t.Fatalf("latch to completing: won=%v err=%v", won, err) + } + go func() { + time.Sleep(50 * time.Millisecond) + _, _ = mem.LatchSession(ctx, uploadID, registry.SessionCompleting, registry.SessionOpen) + }() + + if _, err := mpComplete(t, b, key, uploadID, parts, nil); err != nil { + t.Fatalf("Complete after peer revert: %v (want takeover success)", err) + } + bucket := "bk" + if _, err := b.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: &key}); err != nil { + t.Fatalf("HeadObject after takeover: %v (want committed object)", err) + } +} + +// TestCompleteJoinBudgetSlowDown: a session stuck in 'completing' (wedged +// peer, or a crash that stranded the row until the sweeper reaps it) answers +// a Complete retry with the retryable SlowDown once the join budget is spent +// — never the terminal NoSuchUpload. +func TestCompleteJoinBudgetSlowDown(t *testing.T) { + compressJoinSchedule(t, 5*time.Millisecond, 30*time.Millisecond) + b, mem, _ := newRefTestBackend(t) + ctx := context.Background() + key := "join-budget" + uploadID, parts := mpTwoParts(t, b, key) + + if won, err := mem.LatchSession(ctx, uploadID, registry.SessionOpen, registry.SessionCompleting); err != nil || !won { + t.Fatalf("latch to completing: won=%v err=%v", won, err) + } + _, err := mpComplete(t, b, key, uploadID, parts, nil) + if !errors.Is(err, s3err.GetAPIError(s3err.ErrSlowDown)) { + t.Fatalf("Complete against stuck completing session: %v, want SlowDown", err) + } +} + +// TestCompleteDuringAbortIsNoSuchUpload: losing the latch to an Abort still +// reports NoSuchUpload — only a Complete peer is joined. +func TestCompleteDuringAbortIsNoSuchUpload(t *testing.T) { + compressJoinSchedule(t, 5*time.Millisecond, 5*time.Second) + b, mem, _ := newRefTestBackend(t) + ctx := context.Background() + key := "abort-race" + uploadID, parts := mpTwoParts(t, b, key) + + if won, err := mem.LatchSession(ctx, uploadID, registry.SessionOpen, registry.SessionAborting); err != nil || !won { + t.Fatalf("latch to aborting: won=%v err=%v", won, err) + } + _, err := mpComplete(t, b, key, uploadID, parts, nil) + if !errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)) { + t.Fatalf("Complete against aborting session: %v, want NoSuchUpload", err) + } +} + +// parkingDeferred parks every blob at UploadPart and records each +// ConcludeBlobBatch call, so a test can assert Complete concludes its parked +// blobs together in batches rather than blob-by-blob. failDigests marks +// digests whose first conclude attempt reports a per-blob failure (cleared +// once used, so a retry succeeds). +type parkingDeferred struct { + inmem.NopUploader + mu sync.Mutex + calls [][]uploader.UploadedBlob + failDigests map[string]struct{} +} + +func (d *parkingDeferred) UploadBlob(_ context.Context, _ did.DID, digest multihash.Multihash, size int64, _ string, _ ...uploader.UploadOption) (uploader.UploadedBlob, error) { + task := cid.NewCidV1(cid.Raw, digest) + return uploader.UploadedBlob{ + Digest: digest, + Size: size, + AddTask: task, + AcceptTask: task, + PutInvocation: []byte("parked-put-invocation"), + }, nil +} + +func (d *parkingDeferred) ConcludeBlobBatch(_ context.Context, _ did.DID, parked []uploader.UploadedBlob) ([]uploader.ConcludeBlobResult, error) { + d.mu.Lock() + defer d.mu.Unlock() + d.calls = append(d.calls, append([]uploader.UploadedBlob(nil), parked...)) + results := make([]uploader.ConcludeBlobResult, len(parked)) + for i, p := range parked { + if _, ok := d.failDigests[string(p.Digest)]; ok { + delete(d.failDigests, string(p.Digest)) + results[i] = uploader.ConcludeBlobResult{Err: errors.New("injected conclude failure")} + continue + } + results[i] = uploader.ConcludeBlobResult{Location: uploader.BlobLocation{Provider: "did:test:piri", URL: "http://piri/blob", Size: p.Size}} + } + return results, nil +} + +func (d *parkingDeferred) batchCalls() [][]uploader.UploadedBlob { + d.mu.Lock() + defer d.mu.Unlock() + return append([][]uploader.UploadedBlob(nil), d.calls...) +} + +// newParkingTestBackend is newRefTestBackend with a parking deferred uploader +// wired in, so Complete exercises the real park → batch-conclude path. +func newParkingTestBackend(t *testing.T, def *parkingDeferred) (*Backend, *inmem.MemStore) { + t.Helper() + ctx := context.Background() + dir := t.TempDir() + mem := inmem.NewMemStore() + spool, err := blockstore.NewSpool(filepath.Join(dir, "spool")) + if err != nil { + t.Fatalf("spool: %v", err) + } + log, err := logstore.Open(ctx, logstore.Config{ + Dir: filepath.Join(dir, "segments"), + Meta: mem, + Catalog: logstore.PlaneConfig{Ship: false}, + Logger: zaptest.NewLogger(t), + }) + if err != nil { + t.Fatalf("logstore: %v", err) + } + t.Cleanup(func() { _ = log.Close(ctx) }) + + b := New(Deps{ + Authority: mem, + Registry: mem, + Intents: mem, + Locations: mem, + BlobRefs: mem, + GC: mem, + Multipart: mem, + Parks: mem, + Reads: blockstore.NewLayered(spool, log, inmem.NopBaseReader{}), + Log: log, + Spool: spool, + Uploader: inmem.NopUploader{}, + Deferred: def, + Remover: &recordingRemover{}, + }) + if err := mem.Create(ctx, "bk", did.Undef, registry.CreateState{}); err != nil { + t.Fatalf("create bucket: %v", err) + } + return b, mem +} + +// TestCompleteConcludesParkedBlobsInOneBatch: Complete's accept phase sends +// distinct parked digests to the deferred uploader as ONE batch — blob-by-blob +// concludes cost one round trip + receipt fetch per part, holding the +// connection past client read timeouts at a few hundred parts (issue #69). +func TestCompleteConcludesParkedBlobsInOneBatch(t *testing.T) { + def := &parkingDeferred{} + b, _ := newParkingTestBackend(t, def) + + // Two parts with distinct content → two parked digests → one batch call + // carrying both. + key := "batch-conclude" + uploadID := mpCreate(t, b, key, "", "") + var parts []types.CompletedPart + for i, data := range [][]byte{bytes.Repeat([]byte("a"), int(backend.MinPartSize)), bytes.Repeat([]byte("b"), 16)} { + out, uerr := mpUploadPart(t, b, key, uploadID, int32(i+1), data, nil) + if uerr != nil { + t.Fatalf("UploadPart %d: %v", i+1, uerr) + } + n := int32(i + 1) + parts = append(parts, types.CompletedPart{PartNumber: &n, ETag: out.ETag}) + } + if _, err := mpComplete(t, b, key, uploadID, parts, nil); err != nil { + t.Fatalf("Complete: %v", err) + } + calls := def.batchCalls() + if len(calls) != 1 || len(calls[0]) != 2 { + sizes := make([]int, len(calls)) + for i, c := range calls { + sizes[i] = len(c) + } + t.Fatalf("ConcludeBlobBatch calls = %v (want one call with both parked blobs)", sizes) + } +} + +// TestCompleteResumesAfterPartialConcludeFailure: a per-blob conclude failure +// fails the Complete but keeps every concluded blob's recorded location, so +// the retry re-concludes ONLY the failed blob (healing, issue #69). +func TestCompleteResumesAfterPartialConcludeFailure(t *testing.T) { + partA := bytes.Repeat([]byte("a"), int(backend.MinPartSize)) + partB := bytes.Repeat([]byte("b"), 16) + def := &parkingDeferred{failDigests: map[string]struct{}{string(digestOf(t, partB)): {}}} + b, mem := newParkingTestBackend(t, def) + ctx := context.Background() + + key := "partial-conclude" + uploadID := mpCreate(t, b, key, "", "") + var parts []types.CompletedPart + for i, data := range [][]byte{partA, partB} { + out, uerr := mpUploadPart(t, b, key, uploadID, int32(i+1), data, nil) + if uerr != nil { + t.Fatalf("UploadPart %d: %v", i+1, uerr) + } + n := int32(i + 1) + parts = append(parts, types.CompletedPart{PartNumber: &n, ETag: out.ETag}) + } + + if _, err := mpComplete(t, b, key, uploadID, parts, nil); err == nil { + t.Fatal("Complete with an injected conclude failure: want error, got nil") + } + // The successful blob's location was recorded despite the batch error... + if loc, err := mem.GetLocation(ctx, did.Undef, digestOf(t, partA)); err != nil || loc == nil { + t.Fatalf("location for concluded blob after failed Complete: %v, %v (want recorded)", loc, err) + } + // ...so the retry succeeds and re-concludes only the failed blob. + if _, err := mpComplete(t, b, key, uploadID, parts, nil); err != nil { + t.Fatalf("Complete retry: %v", err) + } + calls := def.batchCalls() + if len(calls) != 2 { + t.Fatalf("ConcludeBlobBatch calls = %d, want 2 (initial + retry)", len(calls)) + } + if len(calls[1]) != 1 || string(calls[1][0].Digest) != string(digestOf(t, partB)) { + t.Fatalf("retry batch = %d blobs, want only the failed blob", len(calls[1])) + } +} diff --git a/uploader/blob.go b/uploader/blob.go index ac49c5f..9ade9ad 100644 --- a/uploader/blob.go +++ b/uploader/blob.go @@ -168,33 +168,61 @@ func locationFromAdded(added forgeclient.AddedBlob) (BlobLocation, error) { var _ BodyUploader = (*Forge)(nil) +// ConcludeBlobResult is one parked blob's outcome from ConcludeBlobBatch: +// its published location, or the error that kept it parked (failed entries +// are safe to resubmit — conclude is idempotent). +type ConcludeBlobResult struct { + Location BlobLocation + Err error +} + // DeferredBodyUploader extends BodyUploader for multipart's deferred accept: // UploadBlob with WithConclude(false) makes the bytes durable (parked) at -// UploadPart; ConcludeBlob triggers accept at Complete; AbortBlob abandons a -// parked blob at Abort. +// UploadPart; ConcludeBlobBatch triggers the accepts at Complete; AbortBlob +// abandons a parked blob at Abort. type DeferredBodyUploader interface { BodyUploader - ConcludeBlob(ctx context.Context, space did.DID, parked UploadedBlob) (BlobLocation, error) + ConcludeBlobBatch(ctx context.Context, space did.DID, parked []UploadedBlob) ([]ConcludeBlobResult, error) AbortBlob(ctx context.Context, space did.DID, digest multihash.Multihash, cause cid.Cid) error } -// ConcludeBlob finishes a parked upload: it concludes the deferred /http/put -// receipt (triggering /blob/accept on the provider) and returns the published -// location. parked is the UploadedBlob a WithConclude(false) upload returned -// (rehydrated from its blob_parks row). The conclude carries no space proof -// (accept is owned by sprue), so no proof store is required. Safe to retry. -func (u *Forge) ConcludeBlob(ctx context.Context, space did.DID, parked UploadedBlob) (BlobLocation, error) { - added, err := u.client.BlobConclude(ctx, space, forgeclient.AddedBlob{ - Digest: parked.Digest, - Size: uint64(parked.Size), - AddTask: parked.AddTask, - AcceptTask: parked.AcceptTask, - PutInvocation: parked.PutInvocation, - }) +// ConcludeBlobBatch finishes many parked uploads in one request to the upload +// service: the deferred /http/put receipts conclude together (triggering +// /blob/accept per blob on the provider) and each blob's published location +// is returned positionally. parked entries are the UploadedBlobs that +// WithConclude(false) uploads returned (rehydrated from blob_parks rows). The +// conclude carries no space proof (accept is owned by sprue), so no proof +// store is required. A returned error means the whole batch went nowhere; +// otherwise inspect each ConcludeBlobResult. Safe to retry either way. +func (u *Forge) ConcludeBlobBatch(ctx context.Context, space did.DID, parked []UploadedBlob) ([]ConcludeBlobResult, error) { + added := make([]forgeclient.AddedBlob, len(parked)) + for i, p := range parked { + added[i] = forgeclient.AddedBlob{ + Digest: p.Digest, + Size: uint64(p.Size), + AddTask: p.AddTask, + AcceptTask: p.AcceptTask, + PutInvocation: p.PutInvocation, + } + } + concluded, err := u.client.BlobConcludeBatch(ctx, space, added) if err != nil { - return BlobLocation{}, fmt.Errorf("uploader: conclude blob: %w", err) + return nil, fmt.Errorf("uploader: conclude blobs: %w", err) + } + results := make([]ConcludeBlobResult, len(parked)) + for i, cr := range concluded { + if cr.Err != nil { + results[i] = ConcludeBlobResult{Err: cr.Err} + continue + } + loc, lerr := locationFromAdded(cr.Blob) + if lerr != nil { + results[i] = ConcludeBlobResult{Err: lerr} + continue + } + results[i] = ConcludeBlobResult{Location: loc} } - return locationFromAdded(added) + return results, nil } // AbortBlob abandons a parked blob via /blob/abort on the upload