G0 D3: catalogd Parquet manifests + ADR-020 idempotent register · 6 scrum fixes

Phase G0 Day 3 ships catalogd: Arrow Parquet manifest codec, in-memory
registry with the ADR-020 idempotency contract (same name+fingerprint
reuses dataset_id; different fingerprint → 409 Conflict), HTTP client
to storaged for persistence, and rehydration on startup. Acceptance
smoke 6/6 PASSES end-to-end including rehydrate-across-restart — the
load-bearing test that the catalog/storaged service split actually
preserves state.

dataset_id derivation diverges from Rust: UUIDv5(namespace, name)
instead of v4 surrogate. Same name on any box generates the same
dataset_id; rehydrate after disk loss converges to the same identity
rather than silently re-issuing. Namespace pinned at
a8f3c1d2-4e5b-5a6c-9d8e-7f0a1b2c3d4e — every dataset_id ever issued
depends on these bytes.

Cross-lineage scrum on shipped code:
  - Opus 4.7 (opencode):                       1 BLOCK + 5 WARN + 3 INFO
  - Kimi K2-0905 (openrouter, validated D2):   2 BLOCK + 2 WARN + 1 INFO
  - Qwen3-coder (openrouter):                  2 BLOCK + 2 WARN + 2 INFO

Fixed:
  C1 list-offsets BLOCK (3-way convergent) → ValueOffsets(0) + bounds
  C2 Rehydrate mutex held across I/O → swap-under-brief-lock pattern
  S1 split-brain on persist failure → candidate-then-swap
  S2 brittle string-match for 400 vs 500 → ErrEmptyName/ErrEmptyFingerprint sentinels
  S3 Get/List shallow-copy aliasing → cloneManifest deep copy
  S4 keep-alive socket leak on error paths → drainAndClose helper

Dismissed (false positives, all single-reviewer):
  Kimi BLOCK "Decode crashes on empty Parquet" — already handled
  Kimi INFO "safeKey double-escapes" — wrong, splitting before escape is required
  Qwen INFO "rb.NewRecord() error unchecked" — API returns no error

Deferred to G1+: name validation regex, per-call deadlines, Snappy
compression, list pagination continuation tokens (storaged caps at
10k with sentinel for now).

Build clean, vet clean, all tests pass, smoke 6/6 PASS after every
fix round. arrow-go/v18 + google/uuid added; Go 1.24 → 1.25 forced
by arrow-go's minimum.

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

internal/catalogd/manifest.go

@@ -0,0 +1,200 @@
+// Package catalogd is the metadata authority — manifests for every
+// registered dataset. A Manifest is the catalog row Rust calls
+// `Manifest`: dataset_id (deterministic from name via UUIDv5),
+// schema_fingerprint (caller-supplied schema hash), the object keys
+// that physically back the dataset in storaged, plus timestamps and
+// optional row_count.
+//
+// G0 stores one Manifest per Parquet file at
+// primary://_catalog/manifests/<name>.parquet. One row per file —
+// catalog manifests are written rarely and read on startup, so the
+// per-file shape favors atomic register over storage density.
+package catalogd
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet"
+	"github.com/apache/arrow-go/v18/parquet/file"
+	"github.com/apache/arrow-go/v18/parquet/pqarrow"
+	"github.com/google/uuid"
+)
+
+// catalogNamespace is the v5 UUID namespace for dataset_id derivation.
+// Same name → same dataset_id across boxes / cold starts. Don't change
+// — every dataset_id ever issued depends on this byte sequence.
+var catalogNamespace = uuid.MustParse("a8f3c1d2-4e5b-5a6c-9d8e-7f0a1b2c3d4e")
+
+// Object is one storaged key contributing to a dataset.
+type Object struct {
+	Key  string `json:"key"`
+	Size int64  `json:"size"`
+}
+
+// Manifest is the catalog row for one dataset.
+type Manifest struct {
+	DatasetID         string    `json:"dataset_id"`
+	Name              string    `json:"name"`
+	SchemaFingerprint string    `json:"schema_fingerprint"`
+	Objects           []Object  `json:"objects"`
+	CreatedAt         time.Time `json:"created_at"`
+	UpdatedAt         time.Time `json:"updated_at"`
+	RowCount          *int64    `json:"row_count,omitempty"`
+}
+
+// DatasetIDForName returns the deterministic UUIDv5 dataset_id for a
+// logical dataset name. Idempotent on the same name across boxes.
+func DatasetIDForName(name string) string {
+	return uuid.NewSHA1(catalogNamespace, []byte(name)).String()
+}
+
+// manifestArrowSchema is the Arrow schema for the on-disk Parquet.
+// Field order matters — codec builders rely on it.
+var manifestArrowSchema = arrow.NewSchema([]arrow.Field{
+	{Name: "dataset_id", Type: arrow.BinaryTypes.String},
+	{Name: "name", Type: arrow.BinaryTypes.String},
+	{Name: "schema_fingerprint", Type: arrow.BinaryTypes.String},
+	{Name: "objects", Type: arrow.ListOf(arrow.StructOf(
+		arrow.Field{Name: "key", Type: arrow.BinaryTypes.String},
+		arrow.Field{Name: "size", Type: arrow.PrimitiveTypes.Int64},
+	))},
+	{Name: "created_at_unix_ns", Type: arrow.PrimitiveTypes.Int64},
+	{Name: "updated_at_unix_ns", Type: arrow.PrimitiveTypes.Int64},
+	{Name: "row_count", Type: arrow.PrimitiveTypes.Int64, Nullable: true},
+}, nil)
+
+// Encode writes a single Manifest to a Parquet byte slice. Memory
+// allocations are bounded — manifests have tens of objects, not
+// millions.
+func Encode(m *Manifest) ([]byte, error) {
+	mem := memory.NewGoAllocator()
+
+	rb := array.NewRecordBuilder(mem, manifestArrowSchema)
+	defer rb.Release()
+
+	rb.Field(0).(*array.StringBuilder).Append(m.DatasetID)
+	rb.Field(1).(*array.StringBuilder).Append(m.Name)
+	rb.Field(2).(*array.StringBuilder).Append(m.SchemaFingerprint)
+
+	listB := rb.Field(3).(*array.ListBuilder)
+	listB.Append(true)
+	structB := listB.ValueBuilder().(*array.StructBuilder)
+	keyB := structB.FieldBuilder(0).(*array.StringBuilder)
+	sizeB := structB.FieldBuilder(1).(*array.Int64Builder)
+	for _, o := range m.Objects {
+		structB.Append(true)
+		keyB.Append(o.Key)
+		sizeB.Append(o.Size)
+	}
+
+	rb.Field(4).(*array.Int64Builder).Append(m.CreatedAt.UnixNano())
+	rb.Field(5).(*array.Int64Builder).Append(m.UpdatedAt.UnixNano())
+	if m.RowCount != nil {
+		rb.Field(6).(*array.Int64Builder).Append(*m.RowCount)
+	} else {
+		rb.Field(6).(*array.Int64Builder).AppendNull()
+	}
+
+	rec := rb.NewRecord()
+	defer rec.Release()
+
+	var buf bytes.Buffer
+	props := parquet.NewWriterProperties()
+	arrowProps := pqarrow.NewArrowWriterProperties()
+	w, err := pqarrow.NewFileWriter(manifestArrowSchema, &buf, props, arrowProps)
+	if err != nil {
+		return nil, fmt.Errorf("pqarrow writer: %w", err)
+	}
+	if err := w.Write(rec); err != nil {
+		return nil, fmt.Errorf("pqarrow write: %w", err)
+	}
+	if err := w.Close(); err != nil {
+		return nil, fmt.Errorf("pqarrow close: %w", err)
+	}
+	return buf.Bytes(), nil
+}
+
+// Decode reads a single-row Parquet manifest back into a Manifest.
+func Decode(b []byte) (*Manifest, error) {
+	rdr, err := file.NewParquetReader(bytes.NewReader(b))
+	if err != nil {
+		return nil, fmt.Errorf("parquet reader: %w", err)
+	}
+	defer rdr.Close()
+
+	pr, err := pqarrow.NewFileReader(rdr, pqarrow.ArrowReadProperties{}, memory.NewGoAllocator())
+	if err != nil {
+		return nil, fmt.Errorf("pqarrow reader: %w", err)
+	}
+	tbl, err := pr.ReadTable(context.Background())
+	if err != nil {
+		return nil, fmt.Errorf("read table: %w", err)
+	}
+	defer tbl.Release()
+
+	if tbl.NumRows() != 1 {
+		return nil, fmt.Errorf("manifest parquet: expected 1 row, got %d", tbl.NumRows())
+	}
+
+	rr := array.NewTableReader(tbl, 1)
+	defer rr.Release()
+	if !rr.Next() {
+		return nil, errors.New("manifest parquet: no record batch")
+	}
+	rec := rr.Record()
+
+	m := &Manifest{
+		DatasetID:         rec.Column(0).(*array.String).Value(0),
+		Name:              rec.Column(1).(*array.String).Value(0),
+		SchemaFingerprint: rec.Column(2).(*array.String).Value(0),
+	}
+
+	// Per scrum C1 (3-way convergent): use ValueOffsets which accounts
+	// for the array's own offset (non-zero under slicing) and is bounds-
+	// safe by API contract. Direct Offsets()[0]/[1] indexing is fragile
+	// under multi-row reads and panics on malformed offset buffers.
+	listArr := rec.Column(3).(*array.List)
+	structArr := listArr.ListValues().(*array.Struct)
+	keyArr := structArr.Field(0).(*array.String)
+	sizeArr := structArr.Field(1).(*array.Int64)
+	start, end := listArr.ValueOffsets(0)
+	if start < 0 || end < start || end > int64(structArr.Len()) {
+		return nil, fmt.Errorf("manifest: bad list offsets [%d, %d] for struct len %d",
+			start, end, structArr.Len())
+	}
+	for i := start; i < end; i++ {
+		m.Objects = append(m.Objects, Object{
+			Key:  keyArr.Value(int(i)),
+			Size: sizeArr.Value(int(i)),
+		})
+	}
+
+	m.CreatedAt = time.Unix(0, rec.Column(4).(*array.Int64).Value(0))
+	m.UpdatedAt = time.Unix(0, rec.Column(5).(*array.Int64).Value(0))
+	rcArr := rec.Column(6).(*array.Int64)
+	if rcArr.IsValid(0) {
+		v := rcArr.Value(0)
+		m.RowCount = &v
+	}
+
+	return m, nil
+}
+
+// EncodeReader is a small convenience for callers that want an
+// io.Reader over the encoded bytes (matches the storaged HTTP PUT
+// signature).
+func EncodeReader(m *Manifest) (io.Reader, int64, error) {
+	b, err := Encode(m)
+	if err != nil {
+		return nil, 0, err
+	}
+	return bytes.NewReader(b), int64(len(b)), nil
+}

internal/catalogd/manifest_test.go

@@ -0,0 +1,103 @@
+package catalogd
+
+import (
+	"testing"
+	"time"
+)
+
+func TestEncodeDecode_RoundTrip(t *testing.T) {
+	rc := int64(500_000)
+	now := time.Unix(1777435000, 123456789)
+	want := &Manifest{
+		DatasetID:         DatasetIDForName("workers_500k"),
+		Name:              "workers_500k",
+		SchemaFingerprint: "sha256:abcdef",
+		Objects: []Object{
+			{Key: "datasets/workers_500k/part-001.parquet", Size: 75 * 1024 * 1024},
+			{Key: "datasets/workers_500k/part-002.parquet", Size: 12 * 1024 * 1024},
+		},
+		CreatedAt: now,
+		UpdatedAt: now.Add(time.Minute),
+		RowCount:  &rc,
+	}
+
+	b, err := Encode(want)
+	if err != nil {
+		t.Fatalf("Encode: %v", err)
+	}
+	if len(b) == 0 {
+		t.Fatal("Encode returned 0 bytes")
+	}
+
+	got, err := Decode(b)
+	if err != nil {
+		t.Fatalf("Decode: %v", err)
+	}
+
+	if got.DatasetID != want.DatasetID {
+		t.Errorf("DatasetID: got %q, want %q", got.DatasetID, want.DatasetID)
+	}
+	if got.Name != want.Name {
+		t.Errorf("Name: got %q, want %q", got.Name, want.Name)
+	}
+	if got.SchemaFingerprint != want.SchemaFingerprint {
+		t.Errorf("SchemaFingerprint: got %q, want %q", got.SchemaFingerprint, want.SchemaFingerprint)
+	}
+	if len(got.Objects) != 2 {
+		t.Fatalf("Objects: got %d, want 2", len(got.Objects))
+	}
+	for i, o := range got.Objects {
+		if o.Key != want.Objects[i].Key || o.Size != want.Objects[i].Size {
+			t.Errorf("Objects[%d]: got %+v, want %+v", i, o, want.Objects[i])
+		}
+	}
+	// Times round-trip via UnixNano so equality is on the nanosecond.
+	if !got.CreatedAt.Equal(want.CreatedAt) {
+		t.Errorf("CreatedAt: got %v, want %v", got.CreatedAt, want.CreatedAt)
+	}
+	if !got.UpdatedAt.Equal(want.UpdatedAt) {
+		t.Errorf("UpdatedAt: got %v, want %v", got.UpdatedAt, want.UpdatedAt)
+	}
+	if got.RowCount == nil || *got.RowCount != *want.RowCount {
+		t.Errorf("RowCount: got %v, want %v", got.RowCount, want.RowCount)
+	}
+}
+
+func TestEncodeDecode_NoRowCount(t *testing.T) {
+	want := &Manifest{
+		DatasetID:         DatasetIDForName("unknown_size"),
+		Name:              "unknown_size",
+		SchemaFingerprint: "sha256:zero",
+		Objects:           []Object{},
+		CreatedAt:         time.Unix(0, 0),
+		UpdatedAt:         time.Unix(0, 0),
+		RowCount:          nil,
+	}
+
+	b, err := Encode(want)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := Decode(b)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.RowCount != nil {
+		t.Errorf("RowCount: got %v, want nil", got.RowCount)
+	}
+	if len(got.Objects) != 0 {
+		t.Errorf("Objects: got %d, want 0", len(got.Objects))
+	}
+}
+
+func TestDatasetIDForName_Deterministic(t *testing.T) {
+	a := DatasetIDForName("workers_500k")
+	b := DatasetIDForName("workers_500k")
+	if a != b {
+		t.Errorf("DatasetIDForName not deterministic: %q vs %q", a, b)
+	}
+	c := DatasetIDForName("different")
+	if a == c {
+		t.Errorf("DatasetIDForName collided across distinct names")
+	}
+}

internal/catalogd/registry.go

@@ -0,0 +1,219 @@
+// registry.go — the in-memory catalog index plus the ADR-020
+// idempotent register contract. The register path holds a single
+// mutex across (lookup → fingerprint check → storage write →
+// in-memory update) to close the check→insert TOCTOU window. The
+// historical Rust bug (308× duplicate manifests on re-register) is
+// the prior art — don't loosen this lock.
+package catalogd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+)
+
+// ManifestPrefix is the storaged key prefix that holds catalog
+// manifests. Objects under this prefix are NOT user data and never
+// surface through ingest/query paths.
+const ManifestPrefix = "_catalog/manifests/"
+
+// ErrFingerprintConflict is returned by Register when a manifest with
+// the same name already exists under a different schema fingerprint.
+// HTTP layer maps this to 409 Conflict; gRPC will map to FAILED_PRECONDITION.
+var ErrFingerprintConflict = errors.New("catalogd: schema fingerprint conflict")
+
+// ErrManifestNotFound is returned by Get when the requested name has
+// no manifest registered.
+var ErrManifestNotFound = errors.New("catalogd: manifest not found")
+
+// ErrEmptyName / ErrEmptyFingerprint are returned by Register on
+// missing required inputs. HTTP layer maps both to 400. Per scrum S2
+// (Opus): sentinel errors so the HTTP boundary uses errors.Is rather
+// than substring matching err.Error().
+var (
+	ErrEmptyName        = errors.New("catalogd: empty name")
+	ErrEmptyFingerprint = errors.New("catalogd: empty schema_fingerprint")
+)
+
+// Store abstracts the storaged HTTP wire so registry can be unit-
+// tested with an in-memory fake.
+type Store interface {
+	Put(ctx context.Context, key string, body []byte) error
+	Get(ctx context.Context, key string) ([]byte, error)
+	List(ctx context.Context, prefix string) ([]string, error)
+}
+
+// Registry is the in-memory authority. Persistence lives in storaged
+// at ManifestPrefix; Registry is rehydrated on startup.
+type Registry struct {
+	mu    sync.Mutex
+	byKey map[string]*Manifest // name → manifest
+	store Store
+	now   func() time.Time // injectable for tests
+}
+
+// NewRegistry builds an empty registry bound to a Store. Call
+// Rehydrate after construction to pick up persisted manifests.
+func NewRegistry(store Store) *Registry {
+	return &Registry{
+		byKey: make(map[string]*Manifest),
+		store: store,
+		now:   time.Now,
+	}
+}
+
+// Rehydrate lists ManifestPrefix in storaged and decodes every entry
+// into the in-memory map. Returns the count of manifests recovered.
+// On any per-file decode error, returns immediately so a corrupt
+// catalog doesn't half-load and silently lose state.
+//
+// Per scrum C2 (Opus + Kimi convergent): network I/O happens OUTSIDE
+// the registry mutex so a slow storaged doesn't block concurrent
+// Register/Get/List. The completed map is swapped in under the lock.
+func (r *Registry) Rehydrate(ctx context.Context) (int, error) {
+	keys, err := r.store.List(ctx, ManifestPrefix)
+	if err != nil {
+		return 0, fmt.Errorf("list manifests: %w", err)
+	}
+	loaded := make(map[string]*Manifest)
+	for _, k := range keys {
+		if !strings.HasPrefix(k, ManifestPrefix) || !strings.HasSuffix(k, ".parquet") {
+			continue
+		}
+		body, err := r.store.Get(ctx, k)
+		if err != nil {
+			return len(loaded), fmt.Errorf("get manifest %s: %w", k, err)
+		}
+		m, err := Decode(body)
+		if err != nil {
+			return len(loaded), fmt.Errorf("decode manifest %s: %w", k, err)
+		}
+		loaded[m.Name] = m
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.byKey = loaded
+	return len(loaded), nil
+}
+
+// Register applies the ADR-020 idempotency contract:
+//
+//   - No prior manifest for name           → create (returns existing=false)
+//   - Prior manifest, same fingerprint     → update objects + bump
+//     updated_at, reuse dataset_id (returns existing=true)
+//   - Prior manifest, different fingerprint → ErrFingerprintConflict
+//
+// The mutex is held across the storage write so concurrent calls for
+// the same name serialize through the persistence layer (low TPS,
+// correctness > throughput).
+func (r *Registry) Register(ctx context.Context, name, fingerprint string, objects []Object, rowCount *int64) (*Manifest, bool, error) {
+	if name == "" {
+		return nil, false, ErrEmptyName
+	}
+	if fingerprint == "" {
+		return nil, false, ErrEmptyFingerprint
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	now := r.now().UTC()
+	prior, exists := r.byKey[name]
+	if exists {
+		if prior.SchemaFingerprint != fingerprint {
+			return nil, true, fmt.Errorf("%w: name=%q have=%s got=%s",
+				ErrFingerprintConflict, name, prior.SchemaFingerprint, fingerprint)
+		}
+		// Same fingerprint — reuse dataset_id, replace objects, bump updated_at.
+		// Per scrum S1 (Opus): build candidate, persist, then swap in. Mutating
+		// `prior` before persist succeeds creates split-brain if storaged is
+		// down — in-memory advances, disk holds the old state, restart loses
+		// what callers were told didn't happen.
+		candidate := *prior
+		candidate.Objects = objects
+		candidate.UpdatedAt = now
+		candidate.RowCount = rowCount
+		if err := r.persist(ctx, &candidate); err != nil {
+			return nil, true, err
+		}
+		r.byKey[name] = &candidate
+		return &candidate, true, nil
+	}
+
+	m := &Manifest{
+		DatasetID:         DatasetIDForName(name),
+		Name:              name,
+		SchemaFingerprint: fingerprint,
+		Objects:           objects,
+		CreatedAt:         now,
+		UpdatedAt:         now,
+		RowCount:          rowCount,
+	}
+	if err := r.persist(ctx, m); err != nil {
+		return nil, false, err
+	}
+	r.byKey[name] = m
+	return m, false, nil
+}
+
+// Get returns a deep copy of the manifest for name, or ErrManifestNotFound.
+func (r *Registry) Get(name string) (*Manifest, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	m, ok := r.byKey[name]
+	if !ok {
+		return nil, ErrManifestNotFound
+	}
+	return cloneManifest(m), nil
+}
+
+// List returns deep copies of every manifest, sorted by name.
+// Callers may mutate the returned slice and the underlying Manifest
+// values without affecting registry state.
+func (r *Registry) List() []*Manifest {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make([]*Manifest, 0, len(r.byKey))
+	for _, m := range r.byKey {
+		out = append(out, cloneManifest(m))
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
+	return out
+}
+
+// cloneManifest deep-copies the Objects slice and dereferences
+// RowCount into a fresh pointer so a returned manifest cannot alias
+// registry state. Per scrum S3 (Opus): the prior `cp := *m` shape
+// shared the Objects backing array — caller-side index writes
+// corrupted registry state without holding the mutex.
+func cloneManifest(m *Manifest) *Manifest {
+	cp := *m
+	if m.Objects != nil {
+		cp.Objects = make([]Object, len(m.Objects))
+		copy(cp.Objects, m.Objects)
+	}
+	if m.RowCount != nil {
+		v := *m.RowCount
+		cp.RowCount = &v
+	}
+	return &cp
+}
+
+// persist encodes the manifest and writes it to storaged at the
+// canonical path. Caller MUST hold r.mu — this function does not
+// take the lock itself.
+func (r *Registry) persist(ctx context.Context, m *Manifest) error {
+	body, err := Encode(m)
+	if err != nil {
+		return fmt.Errorf("encode manifest %s: %w", m.Name, err)
+	}
+	key := ManifestPrefix + m.Name + ".parquet"
+	if err := r.store.Put(ctx, key, body); err != nil {
+		return fmt.Errorf("persist manifest %s: %w", m.Name, err)
+	}
+	return nil
+}

internal/catalogd/registry_test.go

@@ -0,0 +1,160 @@
+package catalogd
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+// memStore is an in-memory Store fake for unit tests.
+type memStore struct {
+	mu   sync.Mutex
+	data map[string][]byte
+}
+
+func newMemStore() *memStore { return &memStore{data: map[string][]byte{}} }
+
+func (m *memStore) Put(_ context.Context, key string, body []byte) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	cp := make([]byte, len(body))
+	copy(cp, body)
+	m.data[key] = cp
+	return nil
+}
+
+func (m *memStore) Get(_ context.Context, key string) ([]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	b, ok := m.data[key]
+	if !ok {
+		return nil, ErrKeyNotFound
+	}
+	return b, nil
+}
+
+func (m *memStore) List(_ context.Context, prefix string) ([]string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	out := []string{}
+	for k := range m.data {
+		if len(k) >= len(prefix) && k[:len(prefix)] == prefix {
+			out = append(out, k)
+		}
+	}
+	return out, nil
+}
+
+func mkRegistry(t *testing.T) (*Registry, *memStore) {
+	t.Helper()
+	s := newMemStore()
+	r := NewRegistry(s)
+	r.now = func() time.Time { return time.Unix(1777435000, 0).UTC() }
+	return r, s
+}
+
+func TestRegister_NewManifest(t *testing.T) {
+	r, _ := mkRegistry(t)
+	rc := int64(100)
+	m, existing, err := r.Register(context.Background(), "workers", "sha256:abc",
+		[]Object{{Key: "datasets/workers/p1.parquet", Size: 1024}}, &rc)
+	if err != nil {
+		t.Fatalf("Register: %v", err)
+	}
+	if existing {
+		t.Error("expected existing=false for new manifest")
+	}
+	if m.DatasetID != DatasetIDForName("workers") {
+		t.Errorf("DatasetID: got %q, want UUIDv5(workers)", m.DatasetID)
+	}
+}
+
+func TestRegister_SameFingerprint_Idempotent(t *testing.T) {
+	r, _ := mkRegistry(t)
+	rc := int64(100)
+	first, _, _ := r.Register(context.Background(), "workers", "sha256:abc",
+		[]Object{{Key: "p1.parquet", Size: 1024}}, &rc)
+
+	// Re-register same name + fingerprint with new objects.
+	rc2 := int64(200)
+	second, existing, err := r.Register(context.Background(), "workers", "sha256:abc",
+		[]Object{{Key: "p1.parquet", Size: 1024}, {Key: "p2.parquet", Size: 2048}}, &rc2)
+	if err != nil {
+		t.Fatalf("Register (idempotent): %v", err)
+	}
+	if !existing {
+		t.Error("expected existing=true on idempotent re-register")
+	}
+	if second.DatasetID != first.DatasetID {
+		t.Errorf("DatasetID changed: %q → %q", first.DatasetID, second.DatasetID)
+	}
+	if len(second.Objects) != 2 {
+		t.Errorf("Objects not replaced: got %d, want 2", len(second.Objects))
+	}
+	if second.RowCount == nil || *second.RowCount != 200 {
+		t.Errorf("RowCount not bumped: got %v, want 200", second.RowCount)
+	}
+}
+
+func TestRegister_DifferentFingerprint_Conflict(t *testing.T) {
+	r, _ := mkRegistry(t)
+	_, _, _ = r.Register(context.Background(), "workers", "sha256:abc",
+		[]Object{{Key: "p1.parquet", Size: 1024}}, nil)
+
+	_, _, err := r.Register(context.Background(), "workers", "sha256:DIFFERENT",
+		[]Object{{Key: "p1.parquet", Size: 1024}}, nil)
+	if !errors.Is(err, ErrFingerprintConflict) {
+		t.Fatalf("expected ErrFingerprintConflict, got %v", err)
+	}
+}
+
+func TestRehydrate_RecoversManifests(t *testing.T) {
+	// Build first registry, register 2 manifests.
+	r1, store := mkRegistry(t)
+	_, _, _ = r1.Register(context.Background(), "workers", "sha256:a", nil, nil)
+	_, _, _ = r1.Register(context.Background(), "candidates", "sha256:b", nil, nil)
+
+	// Build a second registry against the same store + rehydrate.
+	r2 := NewRegistry(store)
+	n, err := r2.Rehydrate(context.Background())
+	if err != nil {
+		t.Fatalf("Rehydrate: %v", err)
+	}
+	if n != 2 {
+		t.Errorf("recovered %d, want 2", n)
+	}
+	if _, err := r2.Get("workers"); err != nil {
+		t.Errorf("Get(workers): %v", err)
+	}
+	if _, err := r2.Get("candidates"); err != nil {
+		t.Errorf("Get(candidates): %v", err)
+	}
+}
+
+func TestList_Sorted(t *testing.T) {
+	r, _ := mkRegistry(t)
+	_, _, _ = r.Register(context.Background(), "zoo", "fp", nil, nil)
+	_, _, _ = r.Register(context.Background(), "alpha", "fp", nil, nil)
+	_, _, _ = r.Register(context.Background(), "midway", "fp", nil, nil)
+	got := r.List()
+	want := []string{"alpha", "midway", "zoo"}
+	for i, m := range got {
+		if m.Name != want[i] {
+			t.Errorf("List[%d]: got %q, want %q", i, m.Name, want[i])
+		}
+	}
+}
+
+func TestRegister_RejectsEmptyInputs(t *testing.T) {
+	r, _ := mkRegistry(t)
+	_, _, err := r.Register(context.Background(), "", "fp", nil, nil)
+	if err == nil {
+		t.Error("expected error on empty name")
+	}
+	_, _, err = r.Register(context.Background(), "x", "", nil, nil)
+	if err == nil {
+		t.Error("expected error on empty fingerprint")
+	}
+}

internal/catalogd/store_client.go

@@ -0,0 +1,140 @@
+// store_client.go — thin HTTP client to storaged. catalogd needs to
+// PUT manifest Parquets, GET them on startup, and LIST the manifests
+// directory. Staying inside an HTTP boundary (rather than reaching
+// into storaged's package directly) preserves the service-boundary
+// shape that gRPC will eventually formalize at G1+.
+package catalogd
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+)
+
+// StoreClient talks HTTP to the storaged service.
+type StoreClient struct {
+	baseURL string
+	hc      *http.Client
+}
+
+// listResponse mirrors storaged's GET /storage/list shape:
+//
+//	{"prefix":"_catalog/manifests/","objects":[{Key,Size,ETag,LastModified}, ...]}
+type listResponse struct {
+	Prefix  string `json:"prefix"`
+	Objects []struct {
+		Key  string `json:"Key"`
+		Size int64  `json:"Size"`
+	} `json:"objects"`
+}
+
+// NewStoreClient builds a client against the given storaged base URL
+// (e.g. "http://127.0.0.1:3211"). Timeout covers manifest read-write
+// only; rehydration of many manifests at startup runs sequentially
+// and each call gets its own timeout window.
+func NewStoreClient(baseURL string) *StoreClient {
+	return &StoreClient{
+		baseURL: strings.TrimRight(baseURL, "/"),
+		hc:      &http.Client{Timeout: 30 * time.Second},
+	}
+}
+
+// Put writes raw bytes at key. body is the encoded Parquet manifest.
+func (c *StoreClient) Put(ctx context.Context, key string, body []byte) error {
+	u := c.baseURL + "/storage/put/" + safeKey(key)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, u, bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("put req: %w", err)
+	}
+	req.ContentLength = int64(len(body))
+	resp, err := c.hc.Do(req)
+	if err != nil {
+		return fmt.Errorf("put do: %w", err)
+	}
+	defer drainAndClose(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		preview, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
+		return fmt.Errorf("put %s: status %d: %s", key, resp.StatusCode, string(preview))
+	}
+	return nil
+}
+
+// Get reads the bytes at key. ErrKeyNotFound on 404.
+func (c *StoreClient) Get(ctx context.Context, key string) ([]byte, error) {
+	u := c.baseURL + "/storage/get/" + safeKey(key)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return nil, fmt.Errorf("get req: %w", err)
+	}
+	resp, err := c.hc.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("get do: %w", err)
+	}
+	defer drainAndClose(resp.Body)
+	if resp.StatusCode == http.StatusNotFound {
+		return nil, ErrKeyNotFound
+	}
+	if resp.StatusCode != http.StatusOK {
+		preview, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
+		return nil, fmt.Errorf("get %s: status %d: %s", key, resp.StatusCode, string(preview))
+	}
+	return io.ReadAll(resp.Body)
+}
+
+// List returns the keys under prefix. Object metadata beyond Key is
+// ignored — catalogd only needs the keys to drive rehydration.
+func (c *StoreClient) List(ctx context.Context, prefix string) ([]string, error) {
+	u := c.baseURL + "/storage/list?prefix=" + url.QueryEscape(prefix)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return nil, fmt.Errorf("list req: %w", err)
+	}
+	resp, err := c.hc.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("list do: %w", err)
+	}
+	defer drainAndClose(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		preview, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
+		return nil, fmt.Errorf("list %s: status %d: %s", prefix, resp.StatusCode, string(preview))
+	}
+	var lr listResponse
+	if err := json.NewDecoder(resp.Body).Decode(&lr); err != nil {
+		return nil, fmt.Errorf("list decode: %w", err)
+	}
+	out := make([]string, 0, len(lr.Objects))
+	for _, o := range lr.Objects {
+		out = append(out, o.Key)
+	}
+	return out, nil
+}
+
+// ErrKeyNotFound mirrors storaged's not-found semantics on the catalogd
+// side without exposing storaged's package types.
+var ErrKeyNotFound = fmt.Errorf("catalogd store: key not found")
+
+// safeKey URL-escapes path segments while preserving "/". storaged's
+// chi `/storage/<verb>/*` routes accept literal slashes in the
+// wildcard match, so we only escape the segments, not the separators.
+func safeKey(key string) string {
+	parts := strings.Split(key, "/")
+	for i, p := range parts {
+		parts[i] = url.PathEscape(p)
+	}
+	return strings.Join(parts, "/")
+}
+
+// drainAndClose reads any remaining body bytes (capped at 64 KiB) and
+// closes the body. Per scrum S6 (Qwen): preview-then-close on error
+// paths leaves bytes in the kernel buffer, breaking HTTP/1.1 keep-
+// alive reuse and slowly leaking sockets.
+func drainAndClose(body io.ReadCloser) {
+	_, _ = io.Copy(io.Discard, io.LimitReader(body, 64<<10))
+	_ = body.Close()
+}

internal/shared/config.go

@@ -21,13 +21,21 @@ import (
 type Config struct {
 	Gateway  ServiceConfig `toml:"gateway"`
 	Storaged ServiceConfig `toml:"storaged"`
-	Catalogd ServiceConfig `toml:"catalogd"`
+	Catalogd CatalogConfig `toml:"catalogd"`
 	Ingestd  ServiceConfig `toml:"ingestd"`
 	Queryd   ServiceConfig `toml:"queryd"`
 	S3       S3Config      `toml:"s3"`
 	Log      LogConfig     `toml:"log"`
 }
 
+// CatalogConfig adds catalogd-specific knobs on top of the standard
+// bind. StoragedURL points at the storaged service for manifest
+// persistence; G0 defaults to the localhost bind.
+type CatalogConfig struct {
+	Bind        string `toml:"bind"`
+	StoragedURL string `toml:"storaged_url"`
+}
+
 // ServiceConfig is the per-binary bind config. Default Bind ""
 // means "use the service's hardcoded G0 default" — see DefaultConfig.
 type ServiceConfig struct {
@@ -57,7 +65,7 @@ func DefaultConfig() Config {
 	return Config{
 		Gateway:  ServiceConfig{Bind: "127.0.0.1:3110"},
 		Storaged: ServiceConfig{Bind: "127.0.0.1:3211"},
-		Catalogd: ServiceConfig{Bind: "127.0.0.1:3212"},
+		Catalogd: CatalogConfig{Bind: "127.0.0.1:3212", StoragedURL: "http://127.0.0.1:3211"},
 		Ingestd:  ServiceConfig{Bind: "127.0.0.1:3213"},
 		Queryd:   ServiceConfig{Bind: "127.0.0.1:3214"},
 		S3: S3Config{