perf(blockstore,bucket): stream body blobs instead of buffering whole in RAM

Addresses PR review (alanshaw): the spool buffered each object-body blob
(up to max_blob_size = 256 MiB) whole in memory on both the write and read
paths, so peak RAM was blob_size × concurrency — wrong for a memory-bound
appliance, and counter to the architecture's "nothing held whole in RAM"
(§5/§7.1). Body blobs now stream; the small-catalog-block path (GetBlock /
PutBlock, CID-keyed, fine in RAM) is unchanged.

New BlobWriter/BlobReader seams (WriteBlob / OpenBlob); ReadStore gains
BlobReader.

Write: SplitBody streams each blob through Spool.WriteBlob, which hashes
(sha256) while copying straight to a temp file (atomic rename to the digest
path) — the blob never sits whole in RAM. Replaces the bytes.Buffer +
PutBlock(block.Block) per blob.

Read: blobBodyReader streams via OpenBlob(digest) (io.ReadCloser) instead
of GetBlock → os.ReadFile, holding at most one open blob at a time and
seeking into it for a ranged read, so peak RAM per concurrent GET is a copy
buffer, not a 256 MiB blob. OpenBlob is implemented by Spool (os.Open),
Forge (stream the /content/retrieve body, dropping io.ReadAll — GetBlock
and OpenBlob now share a `retrieve` helper), and Layered/Cached
(spool → base, no LRU for streamed blobs).

Validated in-process: full suite + new bucket/chunker tests (multi-blob
round trip, mid-blob/boundary-spanning ranged reads, empty body) and
blockstore OpenBlob tier-dispatch tests. The real-network Forge.OpenBlob
path is covered by smelt's read-after-eviction e2e, which currently can't
run — the published piri:main image fails to boot (fx: NewPieceAccepter
missing ucan.Issuer), unrelated to this change — so that leg is pending a
working piri image.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: frrist

blockstore/cache.go

@@ -3,10 +3,12 @@ package blockstore
 import (
 	"container/list"
 	"context"
+	"io"
 	"sync"
 
 	block "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
+	mh "github.com/multiformats/go-multihash"
 )
 
 // Cached wraps a BlockReader with a bounded, in-memory LRU keyed by CID. Blocks
@@ -50,7 +52,21 @@ func NewCached(base BlockReader, maxBytes int64) BlockReader {
 	}
 }
 
-var _ BlockReader = (*Cached)(nil)
+var (
+	_ BlockReader = (*Cached)(nil)
+	_ BlobReader  = (*Cached)(nil)
+)
+
+// OpenBlob streams a body blob straight from the base reader, bypassing the LRU.
+// Body blobs are large and streamed; caching one whole would defeat streaming
+// (and a single blob can exceed the whole budget). Small catalog blocks still
+// cache through GetBlock.
+func (c *Cached) OpenBlob(ctx context.Context, digest mh.Multihash) (io.ReadCloser, error) {
+	if br, ok := c.base.(BlobReader); ok {
+		return br.OpenBlob(ctx, digest)
+	}
+	return nil, ErrNotFound
+}
 
 // GetBlock returns the cached block if present, otherwise fetches it from the
 // base reader and caches it (subject to the byte budget).

blockstore/forge.go

@@ -18,6 +18,7 @@ import (
 	"github.com/fil-forge/ucantone/ucan/invocation"
 	block "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
+	mh "github.com/multiformats/go-multihash"
 	"go.uber.org/zap"
 
 	"github.com/fil-forge/ingot/blockstore/locator"
@@ -47,7 +48,10 @@ type Forge struct {
 	logger      *zap.Logger
 }
 
-var _ BlockReader = (*Forge)(nil)
+var (
+	_ BlockReader = (*Forge)(nil)
+	_ BlobReader  = (*Forge)(nil)
+)
 
 // ForgeConfig wires a read-only Forge block reader.
 type ForgeConfig struct {
@@ -131,26 +135,29 @@ func NewForge(cfg ForgeConfig) (*Forge, error) {
 	}, nil
 }
 
-// GetBlock resolves the CID through the indexer and retrieves the underlying
-// bytes from piri via a UCAN-authorized /content/retrieve invocation, scoped to
-// the inner block's byte range within the containing CAR shard.
-func (f *Forge) GetBlock(ctx context.Context, c cid.Cid) (block.Block, error) {
+// retrieve resolves c through the locator and issues a UCAN-authorized
+// /content/retrieve to the piri node that holds it, returning the response body
+// stream and the expected byte length (the location Range is inclusive, so
+// End-Start+1). The caller owns the returned reader and must Close it. Shared by
+// GetBlock (which buffers small catalog blocks) and OpenBlob (which streams large
+// body blobs straight through).
+func (f *Forge) retrieve(ctx context.Context, c cid.Cid) (io.ReadCloser, int64, error) {
 	locations, err := f.locator.Locate(ctx, f.spaces, c.Hash())
 	if err != nil {
 		var nf locator.NotFoundError
 		if errors.As(err, &nf) {
-			return nil, ErrNotFound
+			return nil, 0, ErrNotFound
 		}
-		return nil, fmt.Errorf("forge: locate %s: %w", c, err)
+		return nil, 0, fmt.Errorf("forge: locate %s: %w", c, err)
 	}
 	if len(locations) == 0 {
-		return nil, ErrNotFound
+		return nil, 0, ErrNotFound
 	}
 
 	loc := locations[0]
 	cm := loc.Commitment
 	if len(cm.Location) == 0 {
-		return nil, fmt.Errorf("forge: empty location URL set for %s", c)
+		return nil, 0, fmt.Errorf("forge: empty location URL set for %s", c)
 	}
 	target := cm.Location[0]
 
@@ -173,7 +180,7 @@ func (f *Forge) GetBlock(ctx context.Context, c cid.Cid) (block.Block, error) {
 		delegation.WithExpiration(ucan.Now()+retrievalAuthTTL),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("forge: build retrieval proof: %w", err)
+		return nil, 0, fmt.Errorf("forge: build retrieval proof: %w", err)
 	}
 
 	inv, err := contentcmds.Retrieve.Invoke(
@@ -187,41 +194,59 @@ func (f *Forge) GetBlock(ctx context.Context, c cid.Cid) (block.Block, error) {
 		invocation.WithProofs(retrievalProof.Link()),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("forge: build retrieve invocation: %w", err)
+		return nil, 0, fmt.Errorf("forge: build retrieve invocation: %w", err)
 	}
 
 	rclient, err := retrieval.NewClient(target.URL(), retrieval.WithHTTPClient(f.httpClient))
 	if err != nil {
-		return nil, fmt.Errorf("forge: build retrieval client: %w", err)
+		return nil, 0, fmt.Errorf("forge: build retrieval client: %w", err)
 	}
 
 	_, _, meta, err := ucanexec.Execute[*contentcmds.RetrieveOK](
 		ctx, rclient, inv,
 		execution.WithDelegations(retrievalProof),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("forge: retrieve %s: %w", c, err)
+		return nil, 0, fmt.Errorf("forge: retrieve %s: %w", c, err)
 	}
 
 	hcRes, ok := meta.(*retrieval.HTTPHeaderResponseContainer)
 	if !ok {
-		return nil, fmt.Errorf("forge: unexpected retrieval metadata type %T", meta)
+		return nil, 0, fmt.Errorf("forge: unexpected retrieval metadata type %T", meta)
+	}
+	// Range is inclusive, so the expected length is End - Start + 1.
+	return hcRes.Body, loc.Range.End - loc.Range.Start + 1, nil
+}
+
+// GetBlock resolves the CID through the locator and retrieves the bytes from piri
+// via a UCAN-authorized /content/retrieve. It buffers the whole block, so it is
+// for small catalog blocks; object-body blobs use the streaming OpenBlob.
+func (f *Forge) GetBlock(ctx context.Context, c cid.Cid) (block.Block, error) {
+	rc, wantLen, err := f.retrieve(ctx, c)
+	if err != nil {
+		return nil, err
 	}
-	defer hcRes.Body.Close()
+	defer rc.Close()
 
-	body, err := io.ReadAll(hcRes.Body)
+	body, err := io.ReadAll(rc)
 	if err != nil {
 		return nil, fmt.Errorf("forge: read retrieve body for %s: %w", c, err)
 	}
-	// Range is inclusive, so the expected length is End - Start + 1.
-	wantLen := loc.Range.End - loc.Range.Start + 1
 	if int64(len(body)) != wantLen {
 		return nil, fmt.Errorf("forge: %s short read: got %d bytes, want %d", c, len(body), wantLen)
 	}
-
 	return block.NewBlockWithCid(body, c)
 }
 
+// OpenBlob streams an object-body blob from piri by digest, without buffering it
+// in memory — the network counterpart of Spool.OpenBlob. Bytes are served
+// straight off the /content/retrieve response; the caller owns the reader and
+// must Close it.
+func (f *Forge) OpenBlob(ctx context.Context, digest mh.Multihash) (io.ReadCloser, error) {
+	rc, _, err := f.retrieve(ctx, cid.NewCidV1(cid.Raw, digest))
+	return rc, err
+}
+
 // newAuthorizeRetrieval returns the AuthorizeRetrievalFunc the IndexLocator
 // calls before each indexer query. The space signer (root authority) directly
 // authorizes the indexer to retrieve any blob in the space — the proof chain is

blockstore/layered.go

@@ -3,9 +3,11 @@ package blockstore
 import (
 	"context"
 	"errors"
+	"io"
 
 	block "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
+	mh "github.com/multiformats/go-multihash"
 )
 
 // Layered is the production ReadStore: a read-only seam that consults the local
@@ -76,6 +78,27 @@ func (l *Layered) GetBlock(ctx context.Context, c cid.Cid) (blk block.Block, ret
 	return l.base.GetBlock(ctx, c)
 }
 
+// OpenBlob streams an object-body blob by digest: the local spool first (the
+// read-after-write floor), then the network base (after eviction). The log is
+// skipped — it only ever holds catalog blocks, never body blobs. Tiers that
+// don't implement BlobReader are treated as a miss. Returns ErrNotFound if no
+// tier has the blob.
+func (l *Layered) OpenBlob(ctx context.Context, digest mh.Multihash) (io.ReadCloser, error) {
+	if br, ok := l.spool.(BlobReader); ok {
+		rc, err := br.OpenBlob(ctx, digest)
+		if err == nil {
+			return rc, nil
+		}
+		if !errors.Is(err, ErrNotFound) {
+			return nil, err
+		}
+	}
+	if br, ok := l.base.(BlobReader); ok {
+		return br.OpenBlob(ctx, digest)
+	}
+	return nil, ErrNotFound
+}
+
 // layeredAsBlockstore lifts Layered into a BaseStore for the
 // CborStore wrapper. Internal-only — exists so the CBOR decoder
 // reuses Layered's cache + fallthrough order rather than going

blockstore/openblob_test.go

@@ -0,0 +1,80 @@
+package blockstore
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"testing"
+
+	block "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	mh "github.com/multiformats/go-multihash"
+)
+
+// fakeBlobTier is a read tier that implements BlockReader (so it fits the
+// Layered/Cached constructors) and BlobReader. data == nil means a miss.
+type fakeBlobTier struct{ data []byte }
+
+func (f fakeBlobTier) GetBlock(context.Context, cid.Cid) (block.Block, error) {
+	return nil, ErrNotFound
+}
+
+func (f fakeBlobTier) OpenBlob(context.Context, mh.Multihash) (io.ReadCloser, error) {
+	if f.data == nil {
+		return nil, ErrNotFound
+	}
+	return io.NopCloser(bytes.NewReader(f.data)), nil
+}
+
+func readOpenBlob(t *testing.T, r BlobReader, digest mh.Multihash) (string, error) {
+	t.Helper()
+	rc, err := r.OpenBlob(context.Background(), digest)
+	if err != nil {
+		return "", err
+	}
+	defer rc.Close()
+	b, err := io.ReadAll(rc)
+	return string(b), err
+}
+
+// TestLayered_OpenBlob_Tiering covers the body-blob read fallthrough: spool first
+// (read-after-write), then the network base after eviction, then ErrNotFound. The
+// log is never consulted (it holds no body blobs).
+func TestLayered_OpenBlob_Tiering(t *testing.T) {
+	digest, _ := mh.Sum([]byte("digest"), mh.SHA2_256, -1)
+
+	// spool hit — base is not consulted.
+	l := NewLayered(fakeBlobTier{data: []byte("from-spool")}, nil, fakeBlobTier{data: []byte("from-base")})
+	if got, err := readOpenBlob(t, l, digest); err != nil || got != "from-spool" {
+		t.Fatalf("spool hit: got %q err %v, want from-spool", got, err)
+	}
+
+	// spool miss → base hit (the after-eviction path).
+	l = NewLayered(fakeBlobTier{data: nil}, nil, fakeBlobTier{data: []byte("from-base")})
+	if got, err := readOpenBlob(t, l, digest); err != nil || got != "from-base" {
+		t.Fatalf("spool miss → base: got %q err %v, want from-base", got, err)
+	}
+
+	// both miss → ErrNotFound.
+	l = NewLayered(fakeBlobTier{data: nil}, nil, fakeBlobTier{data: nil})
+	if _, err := l.OpenBlob(context.Background(), digest); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("both miss: err = %v, want ErrNotFound", err)
+	}
+
+	// nil spool is skipped, not panicked.
+	l = NewLayered(nil, nil, fakeBlobTier{data: []byte("from-base")})
+	if got, err := readOpenBlob(t, l, digest); err != nil || got != "from-base" {
+		t.Fatalf("nil spool → base: got %q err %v, want from-base", got, err)
+	}
+}
+
+// TestCached_OpenBlob_BypassesCache confirms OpenBlob streams straight from the
+// base reader (no LRU) — a large body blob must not be buffered into the cache.
+func TestCached_OpenBlob_BypassesCache(t *testing.T) {
+	digest, _ := mh.Sum([]byte("digest"), mh.SHA2_256, -1)
+	c := NewCached(fakeBlobTier{data: []byte("streamed")}, 1<<20).(*Cached)
+	if got, err := readOpenBlob(t, c, digest); err != nil || got != "streamed" {
+		t.Fatalf("cached OpenBlob: got %q err %v, want streamed", got, err)
+	}
+}

blockstore/spool.go

@@ -2,9 +2,11 @@ package blockstore
 
 import (
 	"context"
+	"crypto/sha256"
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 
@@ -72,6 +74,60 @@ func (s *Spool) PutBlock(_ context.Context, blk block.Block) error {
 	return nil
 }
 
+// WriteBlob streams r to the spool, computing its sha256 digest as it writes so
+// the blob is never held whole in memory (object-body blobs run up to
+// max_blob_size = 256 MiB; buffering them would put that × concurrency in RAM).
+// The write is atomic (temp file → rename to the digest path), so a crash leaves
+// no partial blob readable under its digest. An empty r writes nothing and
+// returns a nil digest with n == 0 (a zero-byte object has no blob). Re-writing
+// an identical blob is idempotent (same digest, rename overwrites in place).
+func (s *Spool) WriteBlob(_ context.Context, r io.Reader) (mh.Multihash, int64, error) {
+	tmp, err := os.CreateTemp(s.dir, ".tmp-*")
+	if err != nil {
+		return nil, 0, fmt.Errorf("blockstore: spool tempfile: %w", err)
+	}
+	tmpName := tmp.Name()
+	hasher := sha256.New()
+	n, copyErr := io.Copy(io.MultiWriter(tmp, hasher), r)
+	if closeErr := tmp.Close(); closeErr != nil && copyErr == nil {
+		copyErr = closeErr
+	}
+	if copyErr != nil {
+		_ = os.Remove(tmpName)
+		return nil, n, fmt.Errorf("blockstore: spool write: %w", copyErr)
+	}
+	if n == 0 {
+		_ = os.Remove(tmpName)
+		return nil, 0, nil
+	}
+	digest, err := mh.Encode(hasher.Sum(nil), mh.SHA2_256)
+	if err != nil {
+		_ = os.Remove(tmpName)
+		return nil, n, fmt.Errorf("blockstore: spool digest: %w", err)
+	}
+	if err := os.Rename(tmpName, s.Path(digest)); err != nil {
+		_ = os.Remove(tmpName)
+		return nil, n, fmt.Errorf("blockstore: spool rename: %w", err)
+	}
+	return digest, n, nil
+}
+
+// OpenBlob returns a streaming reader over the spooled blob with the given
+// digest, or ErrNotFound. Unlike GetBlock it does not read the blob into memory —
+// the body read path serves bytes straight off disk. The caller owns the reader
+// and must Close it. The returned *os.File is seekable, which the body reader
+// uses to start a ranged read mid-blob without reading-and-discarding.
+func (s *Spool) OpenBlob(_ context.Context, digest mh.Multihash) (io.ReadCloser, error) {
+	f, err := os.Open(s.Path(digest))
+	if errors.Is(err, os.ErrNotExist) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("blockstore: spool open %s: %w", digest.B58String(), err)
+	}
+	return f, nil
+}
+
 // GetBlock returns the blob stored under c's multihash, or ErrNotFound. A miss
 // is expected and cheap: it lets the layered read path fall through to the log
 // (for catalog blocks, which are never spooled) or the network tier (for a body
@@ -102,8 +158,11 @@ func (s *Spool) Remove(digest mh.Multihash) error {
 	return nil
 }
 
-// Compile-time assertions: Spool is a read+write block tier.
+// Compile-time assertions: Spool is a read+write block tier, and the streaming
+// blob tier for object bodies.
 var (
 	_ BlockReader = (*Spool)(nil)
 	_ BlockWriter = (*Spool)(nil)
+	_ BlobReader  = (*Spool)(nil)
+	_ BlobWriter  = (*Spool)(nil)
 )