feat: rewrite S3 metadata backend as op journal

cmd/cachewd/main.go

@@ -30,6 +30,7 @@ import (
 	"github.com/block/cachew/internal/jobscheduler"
 	"github.com/block/cachew/internal/logging"
 	"github.com/block/cachew/internal/metadatadb"
+	metadatas3 "github.com/block/cachew/internal/metadatadb/s3"
 	"github.com/block/cachew/internal/metrics"
 	"github.com/block/cachew/internal/opa"
 	"github.com/block/cachew/internal/reaper"
@@ -260,7 +261,7 @@ func newRegistries(
 
 	mr := metadatadb.NewRegistry()
 	metadatadb.RegisterMemory(mr)
-	metadatadb.RegisterS3(mr, s3ClientProvider)
+	metadatas3.Register(mr, s3ClientProvider)
 
 	sr := strategy.NewRegistry()
 	strategy.RegisterAPIV1(sr)

internal/metadatadb/memory.go

@@ -2,7 +2,6 @@ package metadatadb
 
 import (
 	"context"
-	"encoding/json"
 	"sync"
 
 	"github.com/alecthomas/errors"
@@ -39,16 +38,15 @@ func (m *MemoryBackend) Apply(_ context.Context, namespace string, ops ...Op) er
 	defer m.mu.Unlock()
 	ns := m.ns(namespace)
 	for _, o := range ops {
-		applyOp(ns, o)
+		ApplyOp(ns, o)
 	}
 	return nil
 }
 
 func (m *MemoryBackend) Query(_ context.Context, namespace string, q ReadOp, target any) error {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
-	result := queryState(m.ns(namespace), q)
-	return errors.Wrap(jsonUnmarshalInto(result, target), "memory query")
+	return errors.Wrap(QueryStateInto(m.ns(namespace), q, target), "memory query")
 }
 
 func (m *MemoryBackend) Flush(_ context.Context, _ string) error { return nil }
@@ -62,16 +60,3 @@ func (m *MemoryBackend) ns(namespace string) map[string]any {
 	}
 	return ns
 }
-
-// jsonUnmarshalInto marshals src to JSON then unmarshals into target,
-// bridging between the internal any-typed state and the caller's typed pointer.
-func jsonUnmarshalInto(src any, target any) error {
-	if src == nil {
-		return nil
-	}
-	data, err := json.Marshal(src)
-	if err != nil {
-		return errors.Wrap(err, "marshal")
-	}
-	return errors.Wrap(json.Unmarshal(data, target), "unmarshal")
-}

internal/metadatadb/metadatadbtest/soak.go

@@ -84,13 +84,20 @@ func (tr *tracker) appendList(key string) {
 	tr.listAppends[key]++
 }
 
+// SkipUnlessSoak skips the test unless SOAK_TEST is set. Call it before any
+// expensive setup; Soak and SoakReplicas also enforce it.
+func SkipUnlessSoak(t *testing.T) {
+	t.Helper()
+	if os.Getenv("SOAK_TEST") == "" {
+		t.Skip("Skipping soak test; set SOAK_TEST=1 to run")
+	}
+}
+
 // Soak runs a concurrent soak test against a backend, exercising all data
 // structure types with random operations and periodic flushes, then verifies
 // consistency by checking monotonic invariants and round-trip equality.
 func Soak(t *testing.T, backend metadatadb.Backend, config SoakConfig) SoakResult {
-	if os.Getenv("SOAK_TEST") == "" {
-		t.Skip("Skipping soak test; set SOAK_TEST=1 to run")
-	}
+	SkipUnlessSoak(t)
 	config.setDefaults()
 
 	ctx := logging.ContextWithLogger(context.Background(), slog.Default())
@@ -126,6 +133,88 @@ func Soak(t *testing.T, backend metadatadb.Backend, config SoakConfig) SoakResul
 	return result
 }
 
+// SoakReplicas runs the soak workload concurrently across multiple backends
+// sharing the same underlying storage, then verifies the monotonic invariants
+// on every replica and that all replicas converge to identical state.
+func SoakReplicas(t *testing.T, backends []metadatadb.Backend, config SoakConfig) SoakResult {
+	SkipUnlessSoak(t)
+	config.setDefaults()
+
+	ctx := logging.ContextWithLogger(context.Background(), slog.Default())
+	ctx, cancel := context.WithTimeout(ctx, config.Duration+time.Minute)
+	defer cancel()
+
+	replicas := make([]*metadatadb.Namespace, len(backends))
+	for i, backend := range backends {
+		store := metadatadb.New(ctx, backend)
+		t.Cleanup(func() { assert.NoError(t, store.Close(ctx)) })
+		replicas[i] = store.Namespace("soak-replicas")
+	}
+
+	tr := newTracker()
+	var result SoakResult
+	startTime := time.Now()
+	deadline := startTime.Add(config.Duration)
+
+	var wg sync.WaitGroup
+	for replica, ns := range replicas {
+		for worker := range config.Concurrency {
+			wg.Go(func() {
+				soakWorker(ctx, ns, &config, deadline, replica*config.Concurrency+worker, &result, tr)
+			})
+		}
+	}
+	wg.Wait()
+	result.Duration = time.Since(startTime)
+
+	// Writes are synchronous, so everything is durable once the workers
+	// stop; each replica's flush then observes every write.
+	for _, ns := range replicas {
+		assert.NoError(t, ns.Flush(ctx))
+	}
+	for _, ns := range replicas {
+		verifyMonotonicInvariants(t, ns, tr)
+	}
+	verifyReplicasConverge(t, replicas, config.NumKeys)
+	logSoakResult(t, &result)
+	return result
+}
+
+func verifyReplicasConverge(t *testing.T, replicas []*metadatadb.Namespace, numKeys int) {
+	t.Helper()
+	base := replicas[0]
+	for i, ns := range replicas[1:] {
+		for k := range numKeys {
+			key := fmt.Sprintf("key-%d", k)
+
+			baseScalar, baseOK := metadatadb.NewScalar[string](base, "sc-"+key).Get()
+			scalar, ok := metadatadb.NewScalar[string](ns, "sc-"+key).Get()
+			assert.Equal(t, baseOK, ok, "replica %d scalar sc-%s presence", i+1, key)
+			assert.Equal(t, baseScalar, scalar, "replica %d scalar sc-%s", i+1, key)
+
+			assert.Equal(t,
+				metadatadb.NewInt(base, "int-"+key).Get(),
+				metadatadb.NewInt(ns, "int-"+key).Get(), "replica %d int-%s", i+1, key)
+			assert.Equal(t,
+				metadatadb.NewSet[string](base, "set-"+key).Members(),
+				metadatadb.NewSet[string](ns, "set-"+key).Members(), "replica %d set-%s", i+1, key)
+			assert.Equal(t,
+				metadatadb.NewMap[string, string](base, "map-"+key).Entries(),
+				metadatadb.NewMap[string, string](ns, "map-"+key).Entries(), "replica %d map-%s", i+1, key)
+			assert.Equal(t,
+				metadatadb.NewInt(base, "mono-int-"+key).Get(),
+				metadatadb.NewInt(ns, "mono-int-"+key).Get(), "replica %d mono-int-%s", i+1, key)
+			assert.Equal(t,
+				metadatadb.NewSet[string](base, "mono-set-"+key).Members(),
+				metadatadb.NewSet[string](ns, "mono-set-"+key).Members(), "replica %d mono-set-%s", i+1, key)
+			// Element order must match too: canonical replay order is total.
+			assert.Equal(t,
+				metadatadb.NewList[string](base, "mono-list-"+key).Entries(),
+				metadatadb.NewList[string](ns, "mono-list-"+key).Entries(), "replica %d mono-list-%s", i+1, key)
+		}
+	}
+}
+
 func soakWorker(
 	ctx context.Context,
 	ns *metadatadb.Namespace,

internal/metadatadb/ops.go

@@ -1,16 +1,14 @@
 package metadatadb
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
+	"strings"
 
 	"github.com/alecthomas/errors"
 )
 
-// errInvalidToken is returned when an optimistic concurrency token does not
-// match the current version, indicating a concurrent write.
-var errInvalidToken = errors.New("invalid token")
-
 // Op is a write operation applied to the metadata store. Concrete types form a
 // closed set (sum type) — backends handle each variant via exhaustive type switch.
 //
@@ -238,8 +236,24 @@ type ListLen struct{ Key string }
 
 func (ListLen) readOp() {}
 
-// applyOp applies a single write Op to the in-memory state via exhaustive type switch.
-func applyOp(state map[string]any, o Op) { //nolint:funlen
+// QueryStateInto executes a read query against raw namespace state and
+// unmarshals the result into target, bridging the any-typed state and the
+// caller's typed pointer via a JSON round-trip.
+func QueryStateInto(state map[string]any, q ReadOp, target any) error {
+	result := queryState(state, q)
+	if result == nil {
+		return nil
+	}
+	data, err := json.Marshal(result)
+	if err != nil {
+		return errors.Wrap(err, "marshal result")
+	}
+	return errors.Wrap(json.Unmarshal(data, target), "unmarshal result")
+}
+
+// ApplyOp applies a single write Op to raw namespace state via exhaustive
+// type switch.
+func ApplyOp(state map[string]any, o Op) { //nolint:funlen
 	switch o := o.(type) {
 	case ScalarSet:
 		state[o.Key] = o.Value
@@ -320,9 +334,12 @@ func decodeMapKeys(m map[string]any) map[string]any {
 	for k, v := range m {
 		// k is a JSON-encoded key (e.g. "\"hello\"" for string "hello",
 		// "42" for int 42). We need to decode it so that when json.Marshal
-		// re-encodes the result map, the keys come out correctly.
+		// re-encodes the result map, the keys come out correctly. UseNumber
+		// keeps large integer keys exact through fmt.Sprint.
+		dec := json.NewDecoder(strings.NewReader(k))
+		dec.UseNumber()
 		var decoded any
-		if err := json.Unmarshal([]byte(k), &decoded); err != nil {
+		if err := dec.Decode(&decoded); err != nil {
 			result[k] = v
 			continue
 		}
@@ -339,6 +356,20 @@ func marshalKey(v any) string {
 	if err != nil {
 		panic(fmt.Sprintf("metadatadb: marshal key %T: %v", v, err))
 	}
+	// Canonicalize composite keys: a struct marshals in field order, but the
+	// same key decoded from a backend's wire format is a map, which
+	// re-marshals with sorted keys — the two encodings must agree or
+	// replayed entries land under different state keys than local ones.
+	if len(data) > 0 && (data[0] == '{' || data[0] == '[') {
+		dec := json.NewDecoder(bytes.NewReader(data))
+		dec.UseNumber()
+		var decoded any
+		if err := dec.Decode(&decoded); err == nil {
+			if canon, err := json.Marshal(decoded); err == nil {
+				return string(canon)
+			}
+		}
+	}
 	return string(data)
 }
 
@@ -348,6 +379,12 @@ func toInt64(v any) int64 {
 		return n
 	case float64:
 		return int64(n)
+	case json.Number:
+		if i, err := n.Int64(); err == nil {
+			return i
+		}
+		f, _ := n.Float64() //nolint:errcheck // malformed numbers become 0
+		return int64(f)
 	default:
 		return 0
 	}