From 7b10869bdc2f14aeb47355426152311cc95e5df4 Mon Sep 17 00:00:00 2001 From: Alan Shaw Date: Tue, 1 Sep 2026 16:56:00 -0400 Subject: [PATCH] fix: missing params fails decryption and multipart completion cleanup --- docs/architecture.md | 2 +- itest/forge_encryption_test.go | 53 +++ migrations/sql/00014_blob_encryption.sql | 8 +- s3frontend/decrypt.go | 53 +-- s3frontend/decrypt_test.go | 109 +++---- s3frontend/multipart.go | 259 ++++++++++----- s3frontend/multipart_hygiene_test.go | 389 +++++++++++++++++++++++ 7 files changed, 706 insertions(+), 167 deletions(-) create mode 100644 s3frontend/multipart_hygiene_test.go diff --git a/docs/architecture.md b/docs/architecture.md index e5dd3a4c..26d525cc 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -455,7 +455,7 @@ a walk over sub-blocks. The only serialized work is the commit critical section, and under versioning two writes to the same key produce distinct versionIds — they contend only on the guarded root swap, which retries cheaply. Across processes, the swap (a Postgres conditional update) is the cross-instance guard; `blob_refs` -updates are transactional with the commit. +updates land after the commit, best-effort — not inside the transaction. | Case | Handling | |---|---| diff --git a/itest/forge_encryption_test.go b/itest/forge_encryption_test.go index 41808d01..9a721ee8 100644 --- a/itest/forge_encryption_test.go +++ b/itest/forge_encryption_test.go @@ -418,6 +418,59 @@ func TestForgeEncryption(t *testing.T) { t.Logf("supersede shredded %d key rows; winner round-trips", len(oldDigests)) }) + // CompleteShredsOrphanParts: a part uploaded but omitted from the + // Complete list is reaped at Complete — key row, upload intent, and park + // row gone — while the winner commits and round-trips. + t.Run("CompleteShredsOrphanParts", func(t *testing.T) { + const bucket, key = "mp-shred-orphan", "obj" + if _, err := cl.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(bucket)}); err != nil { + t.Fatalf("CreateBucket: %v", err) + } + create, err := cl.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{ + Bucket: aws.String(bucket), Key: aws.String(key), + }) + if err != nil { + t.Fatalf("CreateMultipartUpload: %v", err) + } + winner := tagged(patternBytes(5<<20), 0x2B) + up1, err := cl.UploadPart(ctx, &s3.UploadPartInput{ + Bucket: aws.String(bucket), Key: aws.String(key), UploadId: create.UploadId, + PartNumber: aws.Int32(1), Body: bytes.NewReader(winner), + }) + if err != nil { + t.Fatalf("UploadPart 1: %v", err) + } + if _, err := cl.UploadPart(ctx, &s3.UploadPartInput{ + Bucket: aws.String(bucket), Key: aws.String(key), UploadId: create.UploadId, + PartNumber: aws.Int32(2), Body: bytes.NewReader(tagged(patternBytes(5<<20), 0x9C)), + }); err != nil { + t.Fatalf("UploadPart 2: %v", err) + } + orphan := partBlobDigestsHex(t, ctx, s, aws.ToString(create.UploadId), 2) + if n := countRowsForDigests(t, ctx, s, "ingot.blob_encryption_params", "digest", orphan); n != len(orphan) { + t.Fatalf("pre-complete: %d enc-params rows for %d orphan blobs", n, len(orphan)) + } + + if _, err := cl.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: aws.String(bucket), Key: aws.String(key), UploadId: create.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{Parts: []types.CompletedPart{ + {PartNumber: aws.Int32(1), ETag: up1.ETag}, + }}, + }); err != nil { + t.Fatalf("CompleteMultipartUpload: %v", err) + } + + for _, tbl := range []string{"ingot.blob_encryption_params", "ingot.upload_intents", "ingot.blob_parks"} { + if n := countRowsForDigests(t, ctx, s, tbl, "digest", orphan); n != 0 { + t.Fatalf("%d %s rows survived for the orphaned part", n, tbl) + } + } + if got := getBody(t, ctx, cl, bucket, key, ""); !bytes.Equal(got, winner) { + t.Fatalf("winner mismatch after orphan reap (%d bytes)", len(got)) + } + t.Logf("Complete reaped %d orphan part blobs; winner round-trips", len(orphan)) + }) + t.Run("OverwriteRace", func(t *testing.T) { // When overwrite atomicity lands — deferred blob release (a grace // window so readers holding the prior catalog root finish before its diff --git a/migrations/sql/00014_blob_encryption.sql b/migrations/sql/00014_blob_encryption.sql index 09d2b1d7..5e037d03 100644 --- a/migrations/sql/00014_blob_encryption.sql +++ b/migrations/sql/00014_blob_encryption.sql @@ -7,9 +7,11 @@ -- unwraps the CEK (via the region's secrets manager, which custodies the region -- KEK) and goes straight to a body-range fetch — no envelope-header round-trip. -- --- The existence of a row is what marks a blob as encrypted, so every column is --- NOT NULL: there is no such thing as a half-populated parameter set the --- decrypt path could not use. An unencrypted blob simply has no row. +-- Every body blob is written encrypted, so every body blob has a row (written +-- before its manifest commits), and every column is NOT NULL: there is no such +-- thing as a half-populated parameter set the decrypt path could not use. A +-- referenced blob with a missing row is a read error (fail closed), never +-- "stored as plaintext". -- -- Deliberately a separate table from ingot.blob_locations, and deliberately -- WITHOUT a foreign key to it. blob_locations is a reconstructible cache of the diff --git a/s3frontend/decrypt.go b/s3frontend/decrypt.go index e32068ab..c68517af 100644 --- a/s3frontend/decrypt.go +++ b/s3frontend/decrypt.go @@ -33,11 +33,15 @@ import ( // decrypts that span as it streams (aesstream.SpanReader). Every range in // this file is HTTP-style inclusive [start, end]. -// bodyOpener returns the BlobRangeOpener for one request over body: the -// plain opener when nothing is encrypted, or a decrypting opener carrying -// the prefetched encryption state of every encrypted blob. Prefetching here -// (rather than at first Read) surfaces missing-row/missing-location problems -// as request errors, before any response headers are written. +// bodyOpener returns the BlobRangeOpener for one request over body: a +// decrypting opener carrying the prefetched encryption state of every blob +// (the plain opener alone only for a body with no blobs). Every body blob +// is written encrypted with its params row committed before the manifest, +// so a referenced blob without a row is an error — never "stored as +// plaintext": failing open there would stream raw envelope bytes under a +// 200. Prefetching here (rather than at first Read) surfaces missing-row/ +// missing-location problems as request errors, before any response headers +// are written. // // The encryption-params store and the region key provider are required // dependencies (validated at server construction): which implementation @@ -59,7 +63,10 @@ func (b *Backend) bodyOpener(ctx context.Context, space did.DID, body msbucket.B } params, err := b.encParams.GetEncryptionParams(ctx, space, ref.Digest) if errors.Is(err, registry.ErrNotFound) { - continue // stored as plaintext + // Fail closed: the row is the blob's decryption key material. A + // referenced blob without one is unreadable (a shredded or lost + // row), not plaintext. + return nil, fmt.Errorf("s3frontend: blob %x has no encryption-params row; refusing to serve ciphertext", ref.Digest) } if err != nil { return nil, fmt.Errorf("s3frontend: encryption params for blob %x: %w", ref.Digest, err) @@ -81,10 +88,9 @@ func (b *Backend) bodyOpener(ctx context.Context, space did.DID, body msbucket.B return plain, nil } return &decryptingOpener{ - plain: plain, - read: b.read, - keys: b.regionKeys, - enc: enc, + read: b.read, + keys: b.regionKeys, + enc: enc, }, nil } @@ -94,21 +100,20 @@ type encBlob struct { storedSize int64 // whole stored blob: envelope header + ciphertext } -// decryptingOpener serves each blob's plaintext range, dispatching per blob: -// blobs with no encryption row go through the plain opener untouched; for -// encrypted blobs it unwraps the CEK, fetches the covering ciphertext span, -// and decrypts it in a streaming pass. +// decryptingOpener serves each blob's plaintext range: it unwraps the CEK, +// fetches the covering ciphertext span, and decrypts it in a streaming pass. +// bodyOpener prefetched state for every blob of the body, so a digest absent +// from enc is a bug — fail closed rather than fall through to raw bytes. type decryptingOpener struct { - plain msbucket.BlobRangeOpener - read blockstore.BlobReader - keys regionkey.Provider - enc map[string]encBlob + read blockstore.BlobReader + keys regionkey.Provider + enc map[string]encBlob } func (o *decryptingOpener) OpenBlobRange(ctx context.Context, space did.DID, ref msbucket.BlobRef, start, end int64) (io.ReadCloser, error) { e, ok := o.enc[string(ref.Digest)] if !ok { - return o.plain.OpenBlobRange(ctx, space, ref, start, end) + return nil, fmt.Errorf("s3frontend: blob %x has no prefetched encryption state; refusing to serve ciphertext", ref.Digest) } // The CEK's wrap is context-bound to (space, digest): a row transplanted @@ -163,13 +168,15 @@ func (o *decryptingOpener) OpenBlobRange(ctx context.Context, space did.DID, ref // blobPlaintextLen reports how many plaintext bytes a stored blob decrypts // to: the FEE geometry derived from its encryption-params row and the stored -// (envelope) byte count. A blob with no row is stored as plaintext, so the -// stored size is the answer. Multipart Complete uses this to rebuild the -// manifest's plaintext spans from upload_intents' stored sizes. +// (envelope) byte count. Every body blob has a row; a missing one is an +// error — treating it as plaintext would record the envelope byte count as +// the manifest span, permanently corrupting the object's Size and every +// range GET. Multipart Complete uses this to rebuild the manifest's +// plaintext spans from upload_intents' stored sizes. func (b *Backend) blobPlaintextLen(ctx context.Context, space did.DID, digest multihash.Multihash, storedSize int64) (int64, error) { params, err := b.encParams.GetEncryptionParams(ctx, space, digest) if errors.Is(err, registry.ErrNotFound) { - return storedSize, nil + return 0, fmt.Errorf("blob %x has no encryption-params row", digest) } if err != nil { return 0, fmt.Errorf("encryption params for blob %x: %w", digest, err) diff --git a/s3frontend/decrypt_test.go b/s3frontend/decrypt_test.go index 80b07096..01abfd9d 100644 --- a/s3frontend/decrypt_test.go +++ b/s3frontend/decrypt_test.go @@ -32,6 +32,7 @@ type encFixture struct { body msbucket.Body plaintext []byte spool *blockstore.Spool + mem *inmem.MemStore } // encChunkSize is deliberately the FEE minimum so short test blobs still @@ -41,9 +42,8 @@ const encChunkSize = aesstream.MinChunkSize // newEncFixture splits plaintext into blobs of blobSize bytes, encrypts each // into a FEE envelope stored in a fresh spool under its ciphertext digest, // records the encryption params + location rows, and returns a Backend wired -// with just the pieces the read path uses. plainBlobs marks blob indexes to -// leave unencrypted, proving plain and encrypted blobs mix in one body. -func newEncFixture(t *testing.T, plaintext []byte, blobSize int, plainBlobs ...int) *encFixture { +// with just the pieces the read path uses. +func newEncFixture(t *testing.T, plaintext []byte, blobSize int) *encFixture { t.Helper() ctx := context.Background() @@ -58,59 +58,45 @@ func newEncFixture(t *testing.T, plaintext []byte, blobSize int, plainBlobs ...i } space := testutil.RandomDID(t) - plain := map[int]bool{} - for _, i := range plainBlobs { - plain[i] = true - } - var blobs []msbucket.BlobRef - for i, off := 0, 0; off < len(plaintext); i++ { + for off := 0; off < len(plaintext); { end := off + blobSize if end > len(plaintext) { end = len(plaintext) } piece := plaintext[off:end] - var digest multihash.Multihash - if plain[i] { - digest, _, err = spool.WriteBlob(ctx, bytes.NewReader(piece)) - if err != nil { - t.Fatalf("WriteBlob (plain): %v", err) - } - } else { - cek := randBytes(t, 32) - rc, desc, err := fee.EncryptWithCEK(bytes.NewReader(piece), cek, nil, fee.WithChunkSize(encChunkSize)) - if err != nil { - t.Fatalf("EncryptWithCEK: %v", err) - } - var n int64 - digest, n, err = spool.WriteBlob(ctx, rc) - if err != nil { - t.Fatalf("WriteBlob (envelope): %v", err) - } - _ = rc.Close() + cek := randBytes(t, 32) + rc, desc, err := fee.EncryptWithCEK(bytes.NewReader(piece), cek, nil, fee.WithChunkSize(encChunkSize)) + if err != nil { + t.Fatalf("EncryptWithCEK: %v", err) + } + digest, n, err := spool.WriteBlob(ctx, rc) + if err != nil { + t.Fatalf("WriteBlob (envelope): %v", err) + } + _ = rc.Close() - wrapped, err := provider.Wrap(ctx, regionkey.BindingContext{Space: space, Digest: digest}, cek) - if err != nil { - t.Fatalf("Wrap CEK: %v", err) - } - if err := mem.PutEncryptionParams(ctx, registry.BlobEncryptionParams{ - Space: space, - Digest: digest, - RegionWrappedCEK: wrapped.Ciphertext, - RegionKeyVersion: string(wrapped.Version), - HeaderLen: desc.HeaderLen, - BaseNonce: desc.BaseNonce, - ChunkSize: int64(desc.ChunkSize), - AAD: desc.AAD, - }); err != nil { - t.Fatalf("PutEncryptionParams: %v", err) - } - if err := mem.PutLocation(ctx, registry.BlobLocation{ - Space: space, Digest: digest, Provider: "did:web:piri.test", URL: "http://piri.test/blob", Size: n, - }); err != nil { - t.Fatalf("PutLocation: %v", err) - } + wrapped, err := provider.Wrap(ctx, regionkey.BindingContext{Space: space, Digest: digest}, cek) + if err != nil { + t.Fatalf("Wrap CEK: %v", err) + } + if err := mem.PutEncryptionParams(ctx, registry.BlobEncryptionParams{ + Space: space, + Digest: digest, + RegionWrappedCEK: wrapped.Ciphertext, + RegionKeyVersion: string(wrapped.Version), + HeaderLen: desc.HeaderLen, + BaseNonce: desc.BaseNonce, + ChunkSize: int64(desc.ChunkSize), + AAD: desc.AAD, + }); err != nil { + t.Fatalf("PutEncryptionParams: %v", err) + } + if err := mem.PutLocation(ctx, registry.BlobLocation{ + Space: space, Digest: digest, Provider: "did:web:piri.test", URL: "http://piri.test/blob", Size: n, + }); err != nil { + t.Fatalf("PutLocation: %v", err) } blobs = append(blobs, msbucket.BlobRef{Digest: digest, Start: int64(off), End: int64(end - 1)}) off = end @@ -128,6 +114,7 @@ func newEncFixture(t *testing.T, plaintext []byte, blobSize int, plainBlobs ...i body: msbucket.Body{Size: int64(len(plaintext)), Blobs: blobs}, plaintext: plaintext, spool: spool, + mem: mem, } } @@ -208,22 +195,22 @@ func TestDecryptingRead_Ranges(t *testing.T) { } } -// Plain and encrypted blobs mix in one body: the opener dispatches per blob -// on row existence. -func TestDecryptingRead_MixedPlainAndEncrypted(t *testing.T) { +// Fail closed: every body blob is written encrypted, so a referenced blob +// with no encryption-params row (a shredded or lost row) is a request error +// — never "stored as plaintext", which would stream the raw envelope bytes +// under a 200. +func TestDecryptingRead_MissingParamsRowFailsClosed(t *testing.T) { ctx := context.Background() - fx := newEncFixture(t, patterned(24000), 10000, 1) // blob index 1 stays plaintext + fx := newEncFixture(t, patterned(24000), 10000) - opener, err := fx.backend.bodyOpener(ctx, fx.space, fx.body) - if err != nil { - t.Fatalf("bodyOpener: %v", err) - } - got, err := io.ReadAll(msbucket.OpenBodyRange(ctx, opener, fx.space, fx.body, 9500, 20500)) - if err != nil { - t.Fatalf("range read: %v", err) + // Shred one blob's row out from under the body. + if err := fx.mem.DeleteEncryptionParams(ctx, fx.space, fx.body.Blobs[1].Digest); err != nil { + t.Fatalf("DeleteEncryptionParams: %v", err) } - if !bytes.Equal(got, fx.plaintext[9500:20501]) { - t.Fatal("mixed-body range mismatch") + if _, err := fx.backend.bodyOpener(ctx, fx.space, fx.body); err == nil { + t.Fatal("bodyOpener over a shredded row succeeded — raw ciphertext would have been served") + } else if !strings.Contains(err.Error(), "no encryption-params row") { + t.Fatalf("bodyOpener error = %v, want the missing-row rejection", err) } } diff --git a/s3frontend/multipart.go b/s3frontend/multipart.go index 1b567607..c40c21a3 100644 --- a/s3frontend/multipart.go +++ b/s3frontend/multipart.go @@ -235,14 +235,18 @@ func (b *Backend) UploadPart(ctx context.Context, input *s3.UploadPartInput) (*s } // Capture the superseded part's blobs (if any) before overwriting, so - // last-write-wins doesn't strand its spool files. + // last-write-wins doesn't strand its spool files. A listing failure + // fails the upload: proceeding would silently strand the replaced + // part's blobs and key rows. var superseded []mh.Multihash - if prior, err := b.multipart.ListParts(ctx, uploadID); err == nil { - for _, p := range prior { - if p.PartNumber == int(*input.PartNumber) { - superseded = p.BlobDigests - break - } + prior, err := b.multipart.ListParts(ctx, uploadID) + if err != nil { + return nil, fmt.Errorf("s3frontend: list parts before supersede: %w", err) + } + for _, p := range prior { + if p.PartNumber == int(*input.PartNumber) { + superseded = p.BlobDigests + break } } @@ -277,7 +281,7 @@ func (b *Backend) UploadPart(ctx context.Context, input *s3.UploadPartInput) (*s return nil, fmt.Errorf("s3frontend: park part blobs: %w", err) } if len(superseded) > 0 { - b.cleanupPartBlobs(ctx, space, uploadID, superseded) + b.cleanupPartBlobs(ctx, space, uploadID, superseded, nil) } etag := `"` + hex.EncodeToString(body.MD5) + `"` out := &s3.UploadPartOutput{ETag: &etag} @@ -621,12 +625,35 @@ func (b *Backend) CompleteMultipartUpload(ctx context.Context, input *s3.Complet // The object is durable. Retain the session (state 'completed') and its // parts so a duplicate Complete is idempotent; the sweeper reaps it later. // Best-effort: a failed latch leaves the row in 'completing', which the - // sweeper also treats as terminal after the TTL. + // sweeper reaps through its abort path after the TTL — harmless here, as + // the winners hold reference claims by now, so that cleanup skips them. if _, err := b.multipart.LatchSession(ctx, uploadID, registry.SessionCompleting, registry.SessionCompleted); err != nil { b.logger.Warn("latch session to completed failed; sweeper reaps the completing row after the TTL", zap.String("uploadID", uploadID), zap.Error(err)) } + // Reap parts uploaded but omitted from the winning list: no claim was + // ever added for them, and nothing else revisits their blobs (the part + // rows are retained for idempotency, then cascade away at the sweep). + // keep guards the winners explicitly — the retained part rows would + // otherwise mark every digest, orphans included, as live. Best-effort + // post-commit, like the reference-index reconcile. + winners := make(map[string]bool, len(blobs)) + for _, ref := range blobs { + winners[string(ref.Digest)] = true + } + var orphans []mh.Multihash + for _, sp := range stored { + for _, d := range sp.BlobDigests { + if !winners[string(d)] { + orphans = append(orphans, d) + } + } + } + if len(orphans) > 0 { + b.cleanupPartBlobs(ctx, bucketState.Space, uploadID, orphans, winners) + } + // The x-amz-version-id of the new version, omitted for unversioned buckets // (docs/s3-versioning.md §4.3). versionid := "" @@ -669,12 +696,17 @@ func (b *Backend) AbortMultipartUpload(ctx context.Context, input *s3.AbortMulti return s3err.GetAPIError(s3err.ErrNoSuchUpload) } // Snapshot the parts' blob digests before the cascade delete, then drop - // the session and clean the spool. + // the session and clean the spool. A listing failure fails the abort + // (the session stays latched 'aborting' for the sweeper): deleting the + // session first would cascade the part rows away — the only index to + // the blobs — stranding them unrecoverably. + parts, err := b.multipart.ListParts(ctx, uploadID) + if err != nil { + return fmt.Errorf("s3frontend: list parts before abort: %w", err) + } var digests []mh.Multihash - if parts, err := b.multipart.ListParts(ctx, uploadID); err == nil { - for _, p := range parts { - digests = append(digests, p.BlobDigests...) - } + for _, p := range parts { + digests = append(digests, p.BlobDigests...) } if err := b.multipart.DeleteSession(ctx, uploadID); err != nil { return fmt.Errorf("s3frontend: delete session: %w", err) @@ -683,7 +715,7 @@ func (b *Backend) AbortMultipartUpload(ctx context.Context, input *s3.AbortMulti if err != nil { return err } - b.cleanupPartBlobs(ctx, space, uploadID, digests) + b.cleanupPartBlobs(ctx, space, uploadID, digests, nil) return nil } @@ -703,33 +735,51 @@ func (b *Backend) abortOpenSession(ctx context.Context, space did.DID, sess regi if err != nil || !won { return } + parts, err := b.multipart.ListParts(ctx, sess.UploadID) + if err != nil { + // Leave the session latched 'aborting': deleting it now would + // cascade away the part rows — the only index to the blobs. The + // sweeper's aborting reap retries with a fresh listing. + b.logger.Warn("list parts before implicit abort failed; leaving the session for the sweeper", + zap.String("uploadID", sess.UploadID), zap.Error(err)) + return + } var digests []mh.Multihash - if parts, err := b.multipart.ListParts(ctx, sess.UploadID); err == nil { - for _, p := range parts { - digests = append(digests, p.BlobDigests...) - } + for _, p := range parts { + digests = append(digests, p.BlobDigests...) } if err := b.multipart.DeleteSession(ctx, sess.UploadID); err != nil { return } - b.cleanupPartBlobs(ctx, space, sess.UploadID, digests) + b.cleanupPartBlobs(ctx, space, sess.UploadID, digests, nil) } // cleanupPartBlobs removes spooled blobs that belonged to aborted, expired, or // superseded parts of uploadID — unless the blob is still referenced: by a // part of another in-flight session (content-addressed dedup), by a part still // live in THIS session (a re-uploaded part may share blobs with its -// replacement or a sibling part), or by a committed object (reference claims / -// non-spooled intent state). Best-effort: cleanup failure never fails the S3 -// operation; a stranded spool file is reapable later. -func (b *Backend) cleanupPartBlobs(ctx context.Context, space did.DID, uploadID string, digests []mh.Multihash) { +// replacement or a sibling part), or by a committed object (reference claims). +// keep, when non-nil, overrides the live-parts derivation: Complete passes +// the winning digests, whose part rows are retained for idempotency and +// would otherwise mark every digest live. Best-effort: cleanup failure never +// fails the S3 operation; a stranded spool file is reapable later. +func (b *Backend) cleanupPartBlobs(ctx context.Context, space did.DID, uploadID string, digests []mh.Multihash, keep map[string]bool) { if len(digests) == 0 { return } // Digests still referenced by this session's live parts (after the - // abort/supersede that triggered this cleanup). - live := map[string]bool{} - if parts, err := b.multipart.ListParts(ctx, uploadID); err == nil { + // abort/supersede that triggered this cleanup). A listing failure aborts + // the whole cleanup: an empty live set would delete blobs a sibling part + // still references. + live := keep + if live == nil { + parts, err := b.multipart.ListParts(ctx, uploadID) + if err != nil { + b.logger.Warn("cleanup: list live parts failed; skipping cleanup", + zap.String("uploadID", uploadID), zap.Error(err)) + return + } + live = map[string]bool{} for _, p := range parts { for _, d := range p.BlobDigests { live[string(d)] = true @@ -753,18 +803,18 @@ func (b *Backend) cleanupPartBlobs(ctx context.Context, space did.DID, uploadID if in, err := b.intents.GetIntent(ctx, d); err == nil { state = in.State } - if state != registry.IntentSpooled && state != registry.IntentParked { - // Accepted/published blobs are the reference index's to manage. - continue - } - // A parked blob is durable on its provider — release it there too - // (best-effort; the reject on piri is idempotent, a straggler is - // FIL-625's to reap). Cause is the /blob/add task link the - // upload service needs to locate the provider. A BlobAccepted - // refusal is benign — a concurrent session in this space accepted - // the same content, so the reference index owns the blob now — and - // the park row is obsolete either way. - if state == registry.IntentParked { + switch state { + case registry.IntentSpooled: + // Local only: the spool/intent/enc-params teardown below. + case registry.IntentParked: + // A parked blob is durable on its provider — release it there too + // (best-effort; the reject on piri is idempotent, a straggler is + // the provider's allocation-expiry GC's to reap). Cause is the + // /blob/add task link the upload service needs to locate the + // provider. A BlobAccepted refusal is benign — a concurrent + // session in this space accepted the same content, so the + // reference index owns the blob now — and the park row is + // obsolete either way. if park, err := b.parks.GetPark(ctx, d); err == nil { if cause, err := cid.Cast(park.AddTask); err == nil { if aerr := b.deferred.AbortBlob(ctx, space, d, cause); aerr != nil { @@ -777,6 +827,16 @@ func (b *Backend) cleanupPartBlobs(ctx context.Context, space did.DID, uploadID zap.String("digest", hex.EncodeToString(d)), zap.Error(derr)) } } + case registry.IntentAccepted: + // Accepted with zero claims and zero part refs: nothing will ever + // revisit it — an orphaned part whose Complete omitted it, or a + // Complete whose conclude ran and commit failed. Release through + // the same path a superseded committed blob takes: enc-params + + // location + network remove. + b.releaseBlobs(ctx, space, []mh.Multihash{d}) + default: + // Published blobs are the reference index's to manage. + continue } if rerr := b.spool.Remove(d); rerr != nil { b.logger.Warn("remove spooled blob failed", @@ -786,11 +846,14 @@ func (b *Backend) cleanupPartBlobs(ctx context.Context, space did.DID, uploadID b.logger.Warn("delete upload intent failed", zap.String("digest", hex.EncodeToString(d)), zap.Error(derr)) } - // Crypto-shred the abandoned blob's wrapped CEK: nothing references - // it any more, and without the row the region cannot decrypt it. - if derr := b.encParams.DeleteEncryptionParams(ctx, space, d); derr != nil { - b.logger.Warn("delete encryption params failed", - zap.String("digest", hex.EncodeToString(d)), zap.Error(derr)) + if state != registry.IntentAccepted { + // Crypto-shred the abandoned blob's wrapped CEK: nothing + // references it any more, and without the row the region cannot + // decrypt it. (The accepted arm shredded via releaseBlobs.) + if derr := b.encParams.DeleteEncryptionParams(ctx, space, d); derr != nil { + b.logger.Warn("delete encryption params failed", + zap.String("digest", hex.EncodeToString(d)), zap.Error(derr)) + } } } } @@ -1153,54 +1216,92 @@ func (b *Backend) ListMultipartUploads(ctx context.Context, input *s3.ListMultip // SweepStaleMultipartSessions aborts in-flight multipart sessions older than // ttl (dropping their spooled parts, exactly like a client Abort) and reaps -// completed/aborting leftovers past the same age. Returns how many sessions -// were cleaned. Called periodically by the daemon's sweeper loop. +// completed leftovers past the same age. Sessions a crash stranded +// mid-transition get the abort treatment too: a 'completing' row (Complete +// died before the commit) and an 'aborting' row (Abort died before dropping +// the session) still hold parts whose parked blobs must be released on their +// providers — deleting the row alone would leave those allocations to sit +// until expiry. Returns how many sessions were cleaned. Called periodically +// by the daemon's sweeper loop. func (b *Backend) SweepStaleMultipartSessions(ctx context.Context, ttl time.Duration) (int, error) { cutoff := time.Now().Add(-ttl) cleaned := 0 - // Stale open sessions: latch (losing gracefully to a concurrent - // Complete/Abort) and clean up like an abort. - stale, err := b.multipart.ListStaleSessions(ctx, registry.SessionOpen, cutoff) - if err != nil { - return 0, fmt.Errorf("s3frontend: sweep list: %w", err) - } - for _, s := range stale { - won, err := b.multipart.LatchSession(ctx, s.UploadID, registry.SessionOpen, registry.SessionAborting) - if err != nil || !won { - continue + // Stale open sessions and crash-stranded 'completing' rows: latch into + // 'aborting' (losing gracefully to a concurrent Complete/Abort) and clean + // up like an abort. A 'completing' row whose commit actually landed is + // safe here: its winners hold reference claims, which the cleanup skips. + for _, state := range []string{registry.SessionOpen, registry.SessionCompleting} { + stale, err := b.multipart.ListStaleSessions(ctx, state, cutoff) + if err != nil { + return cleaned, fmt.Errorf("s3frontend: sweep list: %w", err) } - var digests []mh.Multihash - if parts, err := b.multipart.ListParts(ctx, s.UploadID); err == nil { - for _, p := range parts { - digests = append(digests, p.BlobDigests...) + for _, s := range stale { + won, err := b.multipart.LatchSession(ctx, s.UploadID, state, registry.SessionAborting) + if err != nil || !won { + continue + } + if b.reapAbortingSession(ctx, s) { + cleaned++ } } - if err := b.multipart.DeleteSession(ctx, s.UploadID); err != nil { - continue - } - space, serr := b.bucketSpace(ctx, s.Bucket) - if serr != nil { - continue // bucket gone; spool rows are reapable later - } - b.cleanupPartBlobs(ctx, space, s.UploadID, digests) - cleaned++ } - // Terminal leftovers: completed sessions retained for Complete idempotency, - // and any 'completing'/'aborting' rows stranded by a crash mid-transition. - for _, state := range []string{registry.SessionCompleted, registry.SessionCompleting, registry.SessionAborting} { - leftovers, err := b.multipart.ListStaleSessions(ctx, state, cutoff) - if err != nil { - continue + // 'aborting' rows stranded by a crash between the latch and the session + // drop: the latch is already held, so just finish the cleanup. + stranded, err := b.multipart.ListStaleSessions(ctx, registry.SessionAborting, cutoff) + if err != nil { + return cleaned, fmt.Errorf("s3frontend: sweep list: %w", err) + } + for _, s := range stranded { + if b.reapAbortingSession(ctx, s) { + cleaned++ } - for _, s := range leftovers { - if err := b.multipart.DeleteSession(ctx, s.UploadID); err == nil { - cleaned++ - } + } + // Completed sessions, retained for Complete idempotency: Complete reaped + // their orphaned parts, and their winners belong to reference accounting — + // drop the row. + leftovers, err := b.multipart.ListStaleSessions(ctx, registry.SessionCompleted, cutoff) + if err != nil { + return cleaned, fmt.Errorf("s3frontend: sweep list: %w", err) + } + for _, s := range leftovers { + if err := b.multipart.DeleteSession(ctx, s.UploadID); err == nil { + cleaned++ } } return cleaned, nil } +// reapAbortingSession drops a session already latched into 'aborting' and +// releases its parts' now-unreferenced blobs — unallocating parked ones on +// their providers via /blob/abort and releasing accepted-but-unclaimed ones +// through the reference-release path (cleanupPartBlobs skips anything another +// session or a committed object still references). Reports whether the +// session row was removed; on a parts-listing failure the session stays +// latched for the next sweep rather than being deleted blind (the part rows +// are the only index to the blobs). +func (b *Backend) reapAbortingSession(ctx context.Context, s registry.MultipartSession) bool { + // Snapshot the parts' blob digests before the cascade delete. + parts, err := b.multipart.ListParts(ctx, s.UploadID) + if err != nil { + b.logger.Warn("sweep: list parts failed; retrying next sweep", + zap.String("uploadID", s.UploadID), zap.Error(err)) + return false + } + var digests []mh.Multihash + for _, p := range parts { + digests = append(digests, p.BlobDigests...) + } + if err := b.multipart.DeleteSession(ctx, s.UploadID); err != nil { + return false + } + space, err := b.bucketSpace(ctx, s.Bucket) + if err != nil { + return true // bucket gone; spool rows are reapable later + } + b.cleanupPartBlobs(ctx, space, s.UploadID, digests, nil) + return true +} + // etagsEqual compares two ETags ignoring surrounding quotes. func etagsEqual(a, b string) bool { return strings.Trim(a, `"`) == strings.Trim(b, `"`) diff --git a/s3frontend/multipart_hygiene_test.go b/s3frontend/multipart_hygiene_test.go new file mode 100644 index 00000000..6a70943b --- /dev/null +++ b/s3frontend/multipart_hygiene_test.go @@ -0,0 +1,389 @@ +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/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" +) + +// Multipart hygiene: every path that orphans a part blob — Complete omitting +// it, supersede, abort, and the sweeper — must release it fully: enc-params +// row (the crypto-shred), upload intent, park row, and the network copy. +// Under the NopUploader every part blob is IntentAccepted (there is no +// network to park on), so these tests exercise the accepted-at-zero-claims +// release arm; the parked arm is covered by the parkingUploader tests below +// and the Docker itests. + +// hygienePartDigests reads the blob digests recorded for uploadID's part n. +func hygienePartDigests(t *testing.T, mem *inmem.MemStore, uploadID string, n int) []multihash.Multihash { + t.Helper() + parts, err := mem.ListParts(context.Background(), uploadID) + if err != nil { + t.Fatalf("ListParts: %v", err) + } + for _, p := range parts { + if p.PartNumber == n { + if len(p.BlobDigests) == 0 { + t.Fatalf("part %d has no blob digests", n) + } + return p.BlobDigests + } + } + t.Fatalf("part %d not recorded for upload %s", n, uploadID) + return nil +} + +// assertReleased asserts a digest's enc-params row and upload intent are gone +// and (wantRemoved) its network release was invoked. +func assertReleased(t *testing.T, mem *inmem.MemStore, rm *recordingRemover, d multihash.Multihash, wantRemoved bool) { + t.Helper() + ctx := context.Background() + if _, err := mem.GetEncryptionParams(ctx, did.Undef, d); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("enc-params row for %x survived (err=%v) — crypto-shred missing", d, err) + } + if _, err := mem.GetIntent(ctx, d); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("upload intent for %x survived (err=%v)", d, err) + } + if got := rm.removedDigests()[string(d)] > 0; got != wantRemoved { + t.Fatalf("RemoveBlob(%x) invoked = %v, want %v", d, got, wantRemoved) + } +} + +// assertRetained asserts a digest's enc-params row survives and no network +// release was invoked for it. +func assertRetained(t *testing.T, mem *inmem.MemStore, rm *recordingRemover, d multihash.Multihash) { + t.Helper() + if _, err := mem.GetEncryptionParams(context.Background(), did.Undef, d); err != nil { + t.Fatalf("enc-params row for retained blob %x: %v", d, err) + } + if rm.removedDigests()[string(d)] > 0 { + t.Fatalf("retained blob %x was released on the network", d) + } +} + +// TestCompleteReapsOrphanParts: a part uploaded but omitted from the winning +// list is fully released at Complete, while the winners (guarded by the keep +// set — their part rows are retained for idempotency) stay untouched. +func TestCompleteReapsOrphanParts(t *testing.T) { + b, mem, rm := newRefTestBackend(t) + key := "orphan" + winner := testBody(int(backend.MinPartSize)) + + uploadID := mpCreate(t, b, key, "", "") + out1, err := mpUploadPart(t, b, key, uploadID, 1, winner, nil) + if err != nil { + t.Fatalf("UploadPart 1: %v", err) + } + if _, err := mpUploadPart(t, b, key, uploadID, 2, testBody(int(backend.MinPartSize)), nil); err != nil { + t.Fatalf("UploadPart 2: %v", err) + } + winners := hygienePartDigests(t, mem, uploadID, 1) + orphans := hygienePartDigests(t, mem, uploadID, 2) + + one := int32(1) + if _, err := mpComplete(t, b, key, uploadID, []types.CompletedPart{{PartNumber: &one, ETag: out1.ETag}}, nil); err != nil { + t.Fatalf("Complete: %v", err) + } + + for _, d := range orphans { + assertReleased(t, mem, rm, d, true) + } + for _, d := range winners { + assertRetained(t, mem, rm, d) + } + if got := getRange(t, b, key, ""); !bytes.Equal(got, winner) { + t.Fatalf("winner mismatch after orphan reap (%d bytes)", len(got)) + } +} + +// TestSupersededPartReleased: re-uploading a part number releases the +// replaced part's blobs inline (last write wins at Complete). +func TestSupersededPartReleased(t *testing.T) { + b, mem, rm := newRefTestBackend(t) + key := "supersede" + + uploadID := mpCreate(t, b, key, "", "") + if _, err := mpUploadPart(t, b, key, uploadID, 1, testBody(int(backend.MinPartSize)), nil); err != nil { + t.Fatalf("UploadPart (loser): %v", err) + } + old := hygienePartDigests(t, mem, uploadID, 1) + + winner := testBody(int(backend.MinPartSize) + 100) + out2, err := mpUploadPart(t, b, key, uploadID, 1, winner, nil) + if err != nil { + t.Fatalf("UploadPart (winner): %v", err) + } + for _, d := range old { + assertReleased(t, mem, rm, d, true) + } + for _, d := range hygienePartDigests(t, mem, uploadID, 1) { + assertRetained(t, mem, rm, d) + } + + one := int32(1) + if _, err := mpComplete(t, b, key, uploadID, []types.CompletedPart{{PartNumber: &one, ETag: out2.ETag}}, nil); err != nil { + t.Fatalf("Complete: %v", err) + } + if got := getRange(t, b, key, ""); !bytes.Equal(got, winner) { + t.Fatalf("winner mismatch after supersede (%d bytes)", len(got)) + } +} + +// TestAbortReleasesUnclaimedBlobs: a client abort releases every part blob — +// including accepted-with-zero-claims ones, the state every part blob has +// under the NopUploader. +func TestAbortReleasesUnclaimedBlobs(t *testing.T) { + b, mem, rm := newRefTestBackend(t) + bucket, key := "bk", "abort" + + uploadID := mpCreate(t, b, key, "", "") + if _, err := mpUploadPart(t, b, key, uploadID, 1, testBody(int(backend.MinPartSize)), nil); err != nil { + t.Fatalf("UploadPart: %v", err) + } + digests := hygienePartDigests(t, mem, uploadID, 1) + + if err := b.AbortMultipartUpload(context.Background(), &s3.AbortMultipartUploadInput{ + Bucket: &bucket, Key: &key, UploadId: &uploadID, + }); err != nil { + t.Fatalf("Abort: %v", err) + } + if _, err := mem.GetSession(context.Background(), uploadID); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("session survived the abort (err=%v)", err) + } + for _, d := range digests { + assertReleased(t, mem, rm, d, true) + } +} + +// TestSweepReapsCompletingSession: a session a crash stranded in +// 'completing' before the commit (no claims exist) is reaped like an abort — +// blobs released, session gone. +func TestSweepReapsCompletingSession(t *testing.T) { + b, mem, rm := newRefTestBackend(t) + ctx := context.Background() + + uploadID := mpCreate(t, b, "stranded", "", "") + if _, err := mpUploadPart(t, b, "stranded", uploadID, 1, testBody(int(backend.MinPartSize)), nil); err != nil { + t.Fatalf("UploadPart: %v", err) + } + digests := hygienePartDigests(t, mem, uploadID, 1) + 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) + } + + // Negative TTL = a future cutoff: everything is stale, no backdating. + if n, err := b.SweepStaleMultipartSessions(ctx, -time.Second); err != nil || n == 0 { + t.Fatalf("sweep: cleaned=%d err=%v", n, err) + } + if _, err := mem.GetSession(ctx, uploadID); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("completing session survived the sweep (err=%v)", err) + } + for _, d := range digests { + assertReleased(t, mem, rm, d, true) + } +} + +// TestSweepDropsCompletedSessionKeepingWinners: the sweeper drops a +// completed session's row without touching its winners — they hold reference +// claims and belong to the object. +func TestSweepDropsCompletedSessionKeepingWinners(t *testing.T) { + b, mem, rm := newRefTestBackend(t) + ctx := context.Background() + key := "done" + data := testBody(int(backend.MinPartSize)) + + uploadID := mpCreate(t, b, key, "", "") + out, err := mpUploadPart(t, b, key, uploadID, 1, data, nil) + if err != nil { + t.Fatalf("UploadPart: %v", err) + } + one := int32(1) + if _, err := mpComplete(t, b, key, uploadID, []types.CompletedPart{{PartNumber: &one, ETag: out.ETag}}, nil); err != nil { + t.Fatalf("Complete: %v", err) + } + winners := blobDigestsOf(t, b, key, "") + + if n, err := b.SweepStaleMultipartSessions(ctx, -time.Second); err != nil || n == 0 { + t.Fatalf("sweep: cleaned=%d err=%v", n, err) + } + if _, err := mem.GetSession(ctx, uploadID); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("completed session survived the sweep (err=%v)", err) + } + for _, d := range winners { + assertRetained(t, mem, rm, d) + } + if got := getRange(t, b, key, ""); !bytes.Equal(got, data) { + t.Fatalf("object unreadable after the sweep (%d bytes)", len(got)) + } +} + +// TestCompleteFailsClosedOnMissingParamsRow: a part blob whose enc-params +// row is gone fails Complete (blobPlaintextLen cannot derive the plaintext +// span) instead of committing a manifest with an envelope-sized span; the +// session reverts to open. +func TestCompleteFailsClosedOnMissingParamsRow(t *testing.T) { + b, mem, _ := newRefTestBackend(t) + ctx := context.Background() + key := "failclosed" + + uploadID := mpCreate(t, b, key, "", "") + out, err := mpUploadPart(t, b, key, uploadID, 1, testBody(int(backend.MinPartSize)), nil) + if err != nil { + t.Fatalf("UploadPart: %v", err) + } + d := hygienePartDigests(t, mem, uploadID, 1)[0] + if err := mem.DeleteEncryptionParams(ctx, did.Undef, d); err != nil { + t.Fatalf("DeleteEncryptionParams: %v", err) + } + + one := int32(1) + if _, err := mpComplete(t, b, key, uploadID, []types.CompletedPart{{PartNumber: &one, ETag: out.ETag}}, nil); err == nil { + t.Fatal("Complete over a missing enc-params row succeeded — a corrupt span would have committed") + } else if !strings.Contains(err.Error(), "no encryption-params row") { + t.Fatalf("Complete error = %v, want the missing-row rejection", err) + } + sess, err := mem.GetSession(ctx, uploadID) + if err != nil { + t.Fatalf("GetSession: %v", err) + } + if sess.State != registry.SessionOpen { + t.Fatalf("session state = %q after failed Complete, want open (abortable)", sess.State) + } +} + +// parkingUploader parks instead of accepting: UploadBlob returns no +// Location, so part blobs stay IntentParked — the provider shape the +// NopUploader cannot produce. AbortBlob calls are recorded. +type parkingUploader struct { + inmem.NopUploader + mu sync.Mutex + aborted []string +} + +func (p *parkingUploader) UploadBlob(_ context.Context, _ did.DID, digest multihash.Multihash, size int64, _ string, _ ...uploader.UploadOption) (uploader.UploadedBlob, error) { + c := cid.NewCidV1(cid.Raw, digest) + return uploader.UploadedBlob{Digest: digest, Size: size, AddTask: c, AcceptTask: c}, nil +} + +func (p *parkingUploader) AbortBlob(_ context.Context, _ did.DID, d multihash.Multihash, _ cid.Cid) error { + p.mu.Lock() + defer p.mu.Unlock() + p.aborted = append(p.aborted, string(d)) + return nil +} + +func (p *parkingUploader) abortedDigests() map[string]bool { + p.mu.Lock() + defer p.mu.Unlock() + m := map[string]bool{} + for _, d := range p.aborted { + m[d] = true + } + return m +} + +// newParkingBackend is newRefTestBackend with a parking deferred uploader, +// so part blobs reach IntentParked and abort exercises the /blob/abort arm. +func newParkingBackend(t *testing.T) (*Backend, *inmem.MemStore, *parkingUploader) { + 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) }) + + pu := &parkingUploader{} + 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: pu, + Deferred: pu, + Remover: &recordingRemover{}, + EncParams: mem, + RegionKeys: testRegionKeys(t), + TenantKeys: testTenantKeys(), + }) + if err := mem.Create(ctx, "bk", did.Undef, registry.CreateState{}); err != nil { + t.Fatalf("create bucket: %v", err) + } + return b, mem, pu +} + +// TestAbortUnparksParkedBlob: abort of a genuinely parked part blob releases +// it on the provider (/blob/abort), drops the park row, and shreds the key +// row. +func TestAbortUnparksParkedBlob(t *testing.T) { + b, mem, pu := newParkingBackend(t) + ctx := context.Background() + bucket, key := "bk", "parked" + + uploadID := mpCreate(t, b, key, "", "") + if _, err := mpUploadPart(t, b, key, uploadID, 1, testBody(int(backend.MinPartSize)), nil); err != nil { + t.Fatalf("UploadPart: %v", err) + } + digests := hygienePartDigests(t, mem, uploadID, 1) + for _, d := range digests { + in, err := mem.GetIntent(ctx, d) + if err != nil || in.State != registry.IntentParked { + t.Fatalf("part blob %x intent = %v/%v, want parked", d, in, err) + } + } + + if err := b.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{ + Bucket: &bucket, Key: &key, UploadId: &uploadID, + }); err != nil { + t.Fatalf("Abort: %v", err) + } + aborted := pu.abortedDigests() + for _, d := range digests { + if !aborted[string(d)] { + t.Fatalf("parked blob %x was not aborted on the provider", d) + } + if _, err := mem.GetPark(ctx, d); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("park row for %x survived (err=%v)", d, err) + } + if _, err := mem.GetEncryptionParams(ctx, did.Undef, d); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("enc-params row for %x survived (err=%v)", d, err) + } + } +}