fix(s3): release shipped segment blobs on DeleteBucket

A shipped catalog segment registers TWO blobs in the bucket's space: the
sealed CAR and its sharded-dag-index (both blob/added by SubmitShard). Hilt
refuses to delete a space that still holds registrations, so any bucket
that lived past the catalog seal age with a successful ship could never be
deleted: DeleteBucket returned BucketNotEmpty with no objects left.
Surfaced by promoting CompleteMultipartUpload/should_verify_final_composite_
checksum — the first conformance case long enough (~11s of commits) to ship
a segment before its teardown.

The index blob's digest was recorded nowhere, so ship now persists it:
SubmitShard returns the index digest, the flush func hands it to
MarkSegmentShipped, and ingot.segments gains an index_digest column
(00010). DeleteBucket quiesces the bucket's log first — joining any
in-flight ship, so a segment can't register its blobs after the release
pass has read the rows (on faster hosts the teardown lands exactly inside
the ship window) — then releases every sealed segment's CAR and every
shipped segment's index from the space before the hilt delete. Releases
are idempotent (removing an unregistered blob is a no-op), so
sealed-but-unshipped CARs are released defensively and retries are safe.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: frrist

blockstore/staging_test.go

@@ -42,7 +42,7 @@ func (f *fakeMeta) MarkSegmentSealed(_ context.Context, plane blockstore.Plane,
 	f.roots = append(f.roots, opRoots...)
 	return nil
 }
-func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, _ []blockstore.OpRoot) error {
+func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, _ []byte, _ []blockstore.OpRoot) error {
 	if r, ok := f.rows[seq]; ok {
 		r.ShippedAt = shippedAt
 	}
@@ -86,7 +86,7 @@ func (f *fakeMeta) RehydrateSegment(_ context.Context, m logstore.SegmentMeta) e
 
 // nopFlush is the per-plane ship callback for these tests: the store
 // owns the ship-state transition, so the closure is a no-op.
-func nopFlush(_ context.Context, _ *logstore.Segment) error { return nil }
+func nopFlush(_ context.Context, _ *logstore.Segment) ([]byte, error) { return nil, nil }
 
 // noopBase satisfies blockstore.BlockReader but always returns
 // errUnknownBase so we can detect when a GetBlock falls through

inmem/segments_test.go

@@ -44,7 +44,7 @@ func TestMarkSegmentShipped_GuardsForgeRootOnRoot(t *testing.T) {
 
 	// Ship with the stale op-root LAST: unconditionally (the old behavior) it
 	// would win as the last write; the guard must skip it and keep `committed`.
-	if err := m.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq, 100, []blockstore.OpRoot{
+	if err := m.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq, 100, nil, []blockstore.OpRoot{
 		{Bucket: "bk", Root: committed},
 		{Bucket: "bk", Root: stale},
 	}); err != nil {
@@ -65,7 +65,7 @@ func TestMarkSegmentShipped_GuardsForgeRootOnRoot(t *testing.T) {
 	_ = m2.CASRoot(ctx, "bk2", cid.Undef, committed)
 	seq2, _ := m2.NextSegmentSeq(ctx)
 	_ = m2.InsertSegmentOpen(ctx, blockstore.PlaneCatalog, seq2, "bk2")
-	if err := m2.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq2, 100, []blockstore.OpRoot{
+	if err := m2.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq2, 100, nil, []blockstore.OpRoot{
 		{Bucket: "bk2", Root: stale},
 	}); err != nil {
 		t.Fatalf("MarkSegmentShipped(stale only): %v", err)

inmem/store.go

@@ -287,11 +287,12 @@ func (m *MemStore) MarkSegmentSealed(_ context.Context, plane blockstore.Plane,
 	return nil
 }
 
-func (m *MemStore) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, opRoots []blockstore.OpRoot) error {
+func (m *MemStore) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, indexDigest []byte, opRoots []blockstore.OpRoot) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if r, ok := m.segments[seq]; ok {
 		r.ShippedAt = shippedAt
+		r.IndexDigest = append([]byte(nil), indexDigest...)
 	}
 	if plane == blockstore.PlaneCatalog {
 		for _, opr := range opRoots {
@@ -375,8 +376,8 @@ func (NopBaseReader) OpenBlob(_ context.Context, _ did.DID, _ multihash.Multihas
 // network, so the spool's local copy serves all reads.
 type NopUploader struct{}
 
-func (NopUploader) SubmitShard(_ context.Context, _ blockstore.Plane, _ did.DID, _ uploader.CARShard) (uploader.BlobLocation, error) {
-	return uploader.BlobLocation{}, nil
+func (NopUploader) SubmitShard(_ context.Context, _ blockstore.Plane, _ did.DID, _ uploader.CARShard) (uploader.BlobLocation, multihash.Multihash, error) {
+	return uploader.BlobLocation{}, nil, nil
 }
 
 // UploadBlob accepts immediately, even with WithConclude(false) — there is

logstore/config.go

@@ -60,8 +60,12 @@ type PlaneConfig struct {
 }
 
 // FlushFunc is the contract for shipping one sealed segment's CAR to
-// Forge. The segment is single-plane, so no plane argument is needed.
-type FlushFunc func(ctx context.Context, seg *Segment) error
+// Forge. The segment is single-plane, so no plane argument is needed. On
+// success it returns the digest of the shipped sharded-dag-index blob (the
+// ship registers the CAR and its index in the bucket's space; DeleteBucket
+// releases both), or nil when nothing registered (header-only segment, or a
+// non-publishing uploader).
+type FlushFunc func(ctx context.Context, seg *Segment) ([]byte, error)
 
 func (c *Config) validate() error {
 	if c.Dir == "" {

logstore/manager.go

@@ -11,6 +11,7 @@ import (
 
 	block "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multihash"
 	"go.uber.org/zap"
 
 	"github.com/fil-forge/ingot/blockstore"
@@ -219,6 +220,63 @@ func (m *Manager) Close(ctx context.Context) error {
 	return nil
 }
 
+// QuiesceBucketLog stops bucket's flush pipeline and waits for any in-flight
+// ship to finish, so the segment rows ShippedSegmentDigests reads afterwards
+// are final: a ship that was mid-flight has either completed (registered its
+// blobs AND stamped shipped_at + index_digest) or aborted. Without this,
+// DeleteBucket races the flush goroutine — a segment whose CAR just
+// registered but wasn't yet marked shipped would be invisible to the release
+// pass, and the space delete would be refused. The closed store reopens
+// lazily on the bucket's next use, so a delete that fails downstream leaves
+// the bucket functional.
+func (m *Manager) QuiesceBucketLog(ctx context.Context, bucket string) error {
+	if err := validBucketDir(bucket); err != nil {
+		return err
+	}
+	m.mu.Lock()
+	s, ok := m.stores[bucket]
+	delete(m.stores, bucket)
+	m.mu.Unlock()
+	if !ok {
+		return nil
+	}
+	if err := s.Close(ctx); err != nil {
+		return fmt.Errorf("logstore: manager: quiesce log for %q: %w", bucket, err)
+	}
+	return nil
+}
+
+// ShippedSegmentDigests returns the multihash of every blob the bucket's
+// catalog segments may have registered in its space: each shipped segment's
+// CAR and its sharded-dag-index blob, plus the CAR of any sealed-but-
+// unshipped segment (a flush aborted between the CAR's blob/add and the
+// shipped stamp leaves that registration behind; releasing an unregistered
+// blob is a no-op, so over-listing is safe). DeleteBucket must release them
+// all before the space itself can be deleted — the tenant service refuses to
+// delete a space that still holds registrations. Call QuiesceBucketLog first
+// so no ship is in flight while this reads.
+func (m *Manager) ShippedSegmentDigests(ctx context.Context, bucket string) ([][]byte, error) {
+	rows, err := m.meta.ListSegments(ctx, blockstore.PlaneCatalog, bucket)
+	if err != nil {
+		return nil, fmt.Errorf("logstore: manager: list segments for %q: %w", bucket, err)
+	}
+	var out [][]byte
+	for _, r := range rows {
+		if r.State != StateSealed || len(r.SHA256) == 0 {
+			continue
+		}
+		carDigest, err := multihash.Encode(r.SHA256, multihash.SHA2_256)
+		if err != nil {
+			return nil, fmt.Errorf("logstore: manager: encode segment %d sha: %w", r.Seq, err)
+		}
+		out = append(out, carDigest)
+		if r.ShippedAt != 0 && len(r.IndexDigest) > 0 {
+			out = append(out, r.IndexDigest)
+		}
+	}
+	return out, nil
+}
+
 // RemoveBucketLog deletes bucket's log entirely: closes its store (dropping
 // queued-but-unshipped segments — a deleted bucket's history has nowhere to
 // ship), unlinks its directory, and removes its segment rows. Used by

logstore/manager_test.go

@@ -15,20 +15,23 @@ import (
 )
 
 // recordingFlush counts flushes per bucket so tests can assert each
-// bucket's segments ship through that bucket's own closure.
+// bucket's segments ship through that bucket's own closure. Every flush
+// reports fakeIndexDigest as the shipped index blob, as a real ship would.
 type recordingFlush struct {
 	mu    sync.Mutex
 	byBkt map[string]int
 }
 
+var fakeIndexDigest = []byte("fake-index-digest-multihash")
+
 func newRecordingFlush() *recordingFlush { return &recordingFlush{byBkt: map[string]int{}} }
 
 func (r *recordingFlush) forBucket(bucket string) FlushFunc {
-	return func(context.Context, *Segment) error {
+	return func(context.Context, *Segment) ([]byte, error) {
 		r.mu.Lock()
 		defer r.mu.Unlock()
 		r.byBkt[bucket]++
-		return nil
+		return fakeIndexDigest, nil
 	}
 }
 
@@ -181,6 +184,73 @@ func TestManager_RemoveBucketLog(t *testing.T) {
 	}
 }
 
+// TestManager_QuiesceAndShippedSegmentDigests: quiescing joins the flush
+// pipeline so the enumeration afterwards is final — every shipped segment
+// contributes its CAR multihash and index-blob digest, sealed-but-unshipped
+// segments contribute their CAR only (a flush aborted after the CAR's
+// blob/add may have registered it), and a bucket with no sealed segments
+// contributes nothing.
+func TestManager_QuiesceAndShippedSegmentDigests(t *testing.T) {
+	dir := t.TempDir()
+	meta := newFakeMeta()
+	flush := newRecordingFlush()
+	m := openTestManager(t, dir, meta, flush)
+
+	appendFor(t, m, "alpha", "shipped-content")
+
+	// Quiesce without waiting for the seal-age tick: it force-seals the open
+	// segment and joins the flush pipeline, so the rows are final on return
+	// regardless of whether the tail segment shipped or was dropped.
+	if err := m.QuiesceBucketLog(context.Background(), "alpha"); err != nil {
+		t.Fatalf("QuiesceBucketLog: %v", err)
+	}
+
+	rows, err := meta.ListSegments(context.Background(), blockstore.PlaneCatalog, "alpha")
+	if err != nil {
+		t.Fatalf("ListSegments: %v", err)
+	}
+	shipped := map[uint64]bool{}
+	sealed := 0
+	for _, r := range rows {
+		if r.State == StateSealed {
+			sealed++
+			shipped[r.Seq] = r.ShippedAt != 0
+		}
+	}
+	if sealed == 0 {
+		t.Fatalf("expected the quiesce to force-seal the open segment")
+	}
+
+	digests, err := m.ShippedSegmentDigests(context.Background(), "alpha")
+	if err != nil {
+		t.Fatalf("ShippedSegmentDigests: %v", err)
+	}
+	// Every sealed segment contributes a 34-byte sha2-256 CAR multihash;
+	// shipped ones additionally the flush's index digest.
+	wantLen := 0
+	for _, s := range shipped {
+		wantLen++
+		if s {
+			wantLen++
+		}
+	}
+	if len(digests) != wantLen {
+		t.Fatalf("got %d digests, want %d (sealed=%d shipped=%v)", len(digests), wantLen, sealed, shipped)
+	}
+	for _, d := range digests {
+		if len(d) != 34 && string(d) != string(fakeIndexDigest) {
+			t.Fatalf("unexpected digest %q (len %d)", d, len(d))
+		}
+	}
+
+	// A bucket with no sealed segments reports none, and the quiesced
+	// bucket reopens lazily for new appends.
+	if got, err := m.ShippedSegmentDigests(context.Background(), "gamma"); err != nil || len(got) != 0 {
+		t.Fatalf("gamma: got %d digests, err %v", len(got), err)
+	}
+	appendFor(t, m, "alpha", "post-quiesce-write")
+}
+
 // TestManager_RejectsUnsafeBucketNames: the bucket→directory mapping must
 // not be escapable.
 func TestManager_RejectsUnsafeBucketNames(t *testing.T) {

logstore/planelog.go

@@ -314,11 +314,11 @@ func (pl *PlaneLog) flushOne(seg *Segment) {
 	backoff := time.Second
 
 	for attempt := 1; attempt <= maxAttempts; attempt++ {
-		err := pl.pc.Flush(ctx, seg)
+		indexDigest, err := pl.pc.Flush(ctx, seg)
 		if err == nil {
 			now := time.Now().Unix()
 			seg.markShipped(now)
-			if merr := pl.meta.MarkSegmentShipped(ctx, pl.plane, seg.Seq(), now, seg.OpRoots()); merr != nil {
+			if merr := pl.meta.MarkSegmentShipped(ctx, pl.plane, seg.Seq(), now, indexDigest, seg.OpRoots()); merr != nil {
 				pl.logger.Error("logstore: mark shipped",
 					zap.Stringer("plane", pl.plane), zap.Uint64("seq", seg.Seq()), zap.Error(merr))
 			}

logstore/store_test.go

@@ -75,7 +75,7 @@ func (f *fakeMeta) MarkSegmentSealed(_ context.Context, plane blockstore.Plane,
 	return nil
 }
 
-func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, opRoots []blockstore.OpRoot) error {
+func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, indexDigest []byte, opRoots []blockstore.OpRoot) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	m, ok := f.segments[seq]
@@ -86,6 +86,7 @@ func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane,
 		return nil
 	}
 	m.ShippedAt = shippedAt
+	m.IndexDigest = append([]byte(nil), indexDigest...)
 	f.shipped = append(f.shipped, shipEvent{seq: seq, plane: plane})
 	return nil
 }
@@ -173,9 +174,9 @@ func newTestStore(t *testing.T, sealBytes int64, sealAge time.Duration, retain i
 	meta := newFakeMeta()
 	flushCalls := &atomicCounter{}
 	logger := zaptest.NewLogger(t)
-	flush := func(_ context.Context, _ *Segment) error {
+	flush := func(_ context.Context, _ *Segment) ([]byte, error) {
 		flushCalls.add(1)
-		return nil
+		return nil, nil
 	}
 	cfg := Config{
 		Dir:     dir,
@@ -335,7 +336,7 @@ func TestForceSealRecoveredOpenOnRestart(t *testing.T) {
 		cfg := Config{
 			Dir:     dir,
 			Meta:    meta,
-			Catalog: PlaneConfig{SealBytes: 1 << 30, SealAge: time.Hour, Ship: true, Flush: func(context.Context, *Segment) error { return nil }, Retain: 6},
+			Catalog: PlaneConfig{SealBytes: 1 << 30, SealAge: time.Hour, Ship: true, Flush: func(context.Context, *Segment) ([]byte, error) { return nil, nil }, Retain: 6},
 			Logger:  logger,
 		}
 		s, err := Open(context.Background(), cfg)
@@ -512,7 +513,7 @@ func TestCatalogNeverShips(t *testing.T) {
 			SealBytes: 1,
 			SealAge:   20 * time.Millisecond,
 			Ship:      false,
-			Flush:     func(context.Context, *Segment) error { ships.add(1); return nil },
+			Flush:     func(context.Context, *Segment) ([]byte, error) { ships.add(1); return nil, nil },
 		},
 		Logger: logger,
 	}

logstore/types.go

@@ -68,6 +68,11 @@ type SegmentMeta struct {
 	// plane stays 0 forever.
 	ShippedAt int64
 
+	// IndexDigest is the multihash of the shipped sharded-dag-index blob.
+	// Non-nil exactly when the ship registered blobs in the bucket's space
+	// (the CAR + this index) — the registrations DeleteBucket must release.
+	IndexDigest []byte
+
 	// OpRoots are the per-batch (bucket, root) records. Populated only
 	// for catalog-plane segments — op-roots are MST roots.
 	OpRoots []blockstore.OpRoot
@@ -100,7 +105,7 @@ type Meta interface {
 	// to Forge, stamping shipped_at, and advances forge_root_cid in
 	// ingot.buckets for every op-root recorded against this segment (catalog
 	// roots are the MST roots durable on Forge), all in one transaction.
-	MarkSegmentShipped(ctx context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, opRoots []blockstore.OpRoot) error
+	MarkSegmentShipped(ctx context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, indexDigest []byte, opRoots []blockstore.OpRoot) error
 
 	// DeleteSegment removes a segment row (cascades to op-root rows).
 	// Used by retention after the on-disk files are unlinked.

migrations/sql/00010_segment_index_digest.sql

@@ -0,0 +1,11 @@
+-- +goose Up
+-- A shipped segment registers TWO blobs in the bucket's space: the CAR and
+-- its sharded-dag-index. index_digest records the index blob's multihash at
+-- ship time so DeleteBucket can release both registrations (the tenant
+-- service refuses to delete a space that still holds any). NULL means the
+-- segment registered nothing (unshipped, header-only, or a non-publishing
+-- uploader).
+ALTER TABLE ingot.segments ADD COLUMN index_digest bytea;
+
+-- +goose Down
+ALTER TABLE ingot.segments DROP COLUMN index_digest;