fix: missing params fails decryption + multipart completion cleanup

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 |
 |---|---|

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

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

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)

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)
 	}
 }