feat: forge hardening + streaming blob I/O (arch phase 7)

- fix: skip re-uploading already-located blobs (forge dedup; was a 500 on re-PUT)
- registry.LocalLocator: local-table read tier for the appliance topology (no indexer)
- stream body blobs to disk with inline hashing — never held whole in RAM on read or write
- drop unused Spool methods (PutBlock/Has/Remove); trim to Path/WriteBlob/OpenBlob/GetBlock
- test: recover per-case names from the conformance suite

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,13 +48,21 @@ 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 {
-	// IndexerEndpoint is the indexing-service URL.
+	// Locator, when set, resolves blob locations instead of the indexing-service
+	// — the appliance read tier (e.g. registry.LocalLocator over blob_locations).
+	// When nil, an indexing-service-backed locator is built from
+	// IndexerEndpoint/IndexerDID. Either way the retrieval path is identical.
+	Locator locator.Locator
+	// IndexerEndpoint is the indexing-service URL. Required only when Locator is nil.
 	IndexerEndpoint string
-	// IndexerDID is the indexing-service principal.
+	// IndexerDID is the indexing-service principal. Required only when Locator is nil.
 	IndexerDID string
 	// Spaces scopes the locator queries; for ingot this is the single space it owns.
 	Spaces []did.DID
@@ -69,15 +78,10 @@ type ForgeConfig struct {
 	Logger *zap.Logger
 }
 
-// NewForge constructs a Forge block reader, building an indexing-service client
-// wrapped by the index locator.
+// NewForge constructs a Forge block reader. It uses cfg.Locator when set (the
+// appliance read tier); otherwise it builds an indexing-service-backed locator
+// from IndexerEndpoint/IndexerDID.
 func NewForge(cfg ForgeConfig) (*Forge, error) {
-	if cfg.IndexerEndpoint == "" {
-		return nil, errors.New("forge blockstore: indexer endpoint is required")
-	}
-	if cfg.IndexerDID == "" {
-		return nil, errors.New("forge blockstore: indexer DID is required")
-	}
 	if len(cfg.Spaces) == 0 {
 		return nil, errors.New("forge blockstore: at least one space is required")
 	}
@@ -88,15 +92,6 @@ func NewForge(cfg ForgeConfig) (*Forge, error) {
 		return nil, errors.New("forge blockstore: space signer is required")
 	}
 
-	endpointURL, err := url.Parse(cfg.IndexerEndpoint)
-	if err != nil {
-		return nil, fmt.Errorf("forge blockstore: parse indexer endpoint: %w", err)
-	}
-	indexerDID, err := did.Parse(cfg.IndexerDID)
-	if err != nil {
-		return nil, fmt.Errorf("forge blockstore: parse indexer DID: %w", err)
-	}
-
 	httpc := cfg.HTTPClient
 	if httpc == nil {
 		httpc = http.DefaultClient
@@ -106,14 +101,30 @@ func NewForge(cfg ForgeConfig) (*Forge, error) {
 		logger = zap.NewNop()
 	}
 
-	idxClient, err := indexclient.New(indexerDID, *endpointURL, indexclient.WithHTTPClient(httpc))
-	if err != nil {
-		return nil, fmt.Errorf("forge blockstore: build indexing-service client: %w", err)
+	loc := cfg.Locator
+	if loc == nil {
+		// No locator injected: resolve through the indexing-service.
+		if cfg.IndexerEndpoint == "" {
+			return nil, errors.New("forge blockstore: indexer endpoint is required (no locator injected)")
+		}
+		if cfg.IndexerDID == "" {
+			return nil, errors.New("forge blockstore: indexer DID is required (no locator injected)")
+		}
+		endpointURL, err := url.Parse(cfg.IndexerEndpoint)
+		if err != nil {
+			return nil, fmt.Errorf("forge blockstore: parse indexer endpoint: %w", err)
+		}
+		indexerDID, err := did.Parse(cfg.IndexerDID)
+		if err != nil {
+			return nil, fmt.Errorf("forge blockstore: parse indexer DID: %w", err)
+		}
+		idxClient, err := indexclient.New(indexerDID, *endpointURL, indexclient.WithHTTPClient(httpc))
+		if err != nil {
+			return nil, fmt.Errorf("forge blockstore: build indexing-service client: %w", err)
+		}
+		loc = locator.NewIndexLocator(idxClient, newAuthorizeRetrieval(cfg.SpaceSigner, indexerDID))
 	}
 
-	authFn := newAuthorizeRetrieval(cfg.SpaceSigner, indexerDID)
-	loc := locator.NewIndexLocator(idxClient, authFn)
-
 	return &Forge{
 		locator:     loc,
 		signer:      cfg.Signer,
@@ -124,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]
 
@@ -166,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(
@@ -180,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)
 	}
-	defer hcRes.Body.Close()
+	// Range is inclusive, so the expected length is End - Start + 1.
+	return hcRes.Body, loc.Range.End - loc.Range.Start + 1, nil
+}
 
-	body, err := io.ReadAll(hcRes.Body)
+// 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 rc.Close()
+
+	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)
+	}
+}