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 asks the log manager for every shipped segment's
(CAR, index) digests and releases both from the space before the hilt
delete; releases are idempotent, so 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,32 @@ func (m *Manager) Close(ctx context.Context) error {
 	return nil
 }
 
+// ShippedSegmentDigests returns the multihash of every blob the bucket's
+// catalog segments registered in its space: each shipped segment's CAR and
+// its sharded-dag-index blob. DeleteBucket must release them all before the
+// space itself can be deleted (the tenant service refuses to delete a space
+// that still holds registrations). Segments whose ship registered nothing —
+// unshipped, header-only, or shipped through a non-publishing uploader —
+// carry no IndexDigest and contribute nothing.
+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.ShippedAt == 0 || len(r.IndexDigest) == 0 || 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, 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,63 @@ func TestManager_RemoveBucketLog(t *testing.T) {
 	}
 }
 
+// TestManager_ShippedSegmentDigests: a shipped segment contributes its CAR
+// multihash and index-blob digest — the space registrations DeleteBucket
+// must release — and a bucket with no shipped segments contributes nothing.
+func TestManager_ShippedSegmentDigests(t *testing.T) {
+	dir := t.TempDir()
+	meta := newFakeMeta()
+	flush := newRecordingFlush()
+	m := openTestManager(t, dir, meta, flush)
+
+	appendFor(t, m, "alpha", "shipped-content")
+
+	// Wait for alpha's segment to seal + flush; the flush marks it shipped
+	// with the recorded index digest.
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if flush.count("alpha") > 0 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	var shipped bool
+	rows, err := meta.ListSegments(context.Background(), blockstore.PlaneCatalog, "alpha")
+	if err != nil {
+		t.Fatalf("ListSegments: %v", err)
+	}
+	for _, r := range rows {
+		if r.ShippedAt != 0 {
+			shipped = true
+		}
+	}
+	if !shipped {
+		t.Skip("segment did not ship within the deadline; nothing to assert")
+	}
+
+	digests, err := m.ShippedSegmentDigests(context.Background(), "alpha")
+	if err != nil {
+		t.Fatalf("ShippedSegmentDigests: %v", err)
+	}
+	if len(digests) == 0 || len(digests)%2 != 0 {
+		t.Fatalf("expected (CAR, index) digest pairs, got %d entries", len(digests))
+	}
+	for i := 0; i < len(digests); i += 2 {
+		// The CAR digest is the multihash-encoded sha256 (2-byte prefix +
+		// 32-byte digest); the index digest is whatever the flush reported.
+		if len(digests[i]) != 34 {
+			t.Fatalf("CAR digest %d: expected 34-byte sha2-256 multihash, got %d", i, len(digests[i]))
+		}
+		if string(digests[i+1]) != string(fakeIndexDigest) {
+			t.Fatalf("index digest %d: got %q", i+1, digests[i+1])
+		}
+	}
+	// A bucket with no shipped segments reports none.
+	if got, err := m.ShippedSegmentDigests(context.Background(), "gamma"); err != nil || len(got) != 0 {
+		t.Fatalf("gamma: got %d digests, err %v", len(got), err)
+	}
+}
+
 // 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;