refactor: replace oplog with public Op/ReadOp sum types on Backend interface

internal/metadatadb/memory.go

@@ -3,73 +3,60 @@ package metadatadb
 import (
 	"context"
 	"encoding/json"
-	"strconv"
 	"sync"
-	"sync/atomic"
 
 	"github.com/alecthomas/errors"
 )
 
 // MemoryBackend is an in-memory Backend for testing and single-instance
-// deployments. It is safe for concurrent use across multiple Store instances.
+// deployments. Ops are applied directly — there is no sync or persistence.
 type MemoryBackend struct {
-	mu      sync.Mutex
-	data    map[string]json.RawMessage
-	tokens  map[string]string
-	version atomic.Int64
-	locks   map[string]chan struct{}
+	mu    sync.RWMutex
+	state map[string]map[string]any // namespace -> state
 }
 
 func NewMemoryBackend() *MemoryBackend {
-	return &MemoryBackend{
-		data:   make(map[string]json.RawMessage),
-		tokens: make(map[string]string),
-		locks:  make(map[string]chan struct{}),
-	}
-}
-
-func (m *MemoryBackend) Load(_ context.Context, namespace string) (json.RawMessage, string, error) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	return m.data[namespace], m.tokens[namespace], nil
+	return &MemoryBackend{state: make(map[string]map[string]any)}
 }
 
-func (m *MemoryBackend) Store(_ context.Context, namespace string, data json.RawMessage, token string) error {
+func (m *MemoryBackend) Apply(_ context.Context, namespace string, ops ...Op) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	if m.tokens[namespace] != token {
-		return ErrInvalidToken
+	ns := m.ns(namespace)
+	for _, o := range ops {
+		applyOp(ns, o)
 	}
-	m.data[namespace] = data
-	m.tokens[namespace] = strconv.FormatInt(m.version.Add(1), 10)
 	return nil
 }
 
-func (m *MemoryBackend) Lock(ctx context.Context, namespace string) error {
-	for {
-		m.mu.Lock()
-		ch, locked := m.locks[namespace]
-		if !locked {
-			m.locks[namespace] = make(chan struct{})
-			m.mu.Unlock()
-			return nil
-		}
-		m.mu.Unlock()
+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")
+}
 
-		select {
-		case <-ch:
-		case <-ctx.Done():
-			return errors.WithStack(ctx.Err())
-		}
+func (m *MemoryBackend) Flush(_ context.Context, _ string) error { return nil }
+func (m *MemoryBackend) Close(_ context.Context) error           { return nil }
+
+func (m *MemoryBackend) ns(namespace string) map[string]any {
+	ns, ok := m.state[namespace]
+	if !ok {
+		ns = make(map[string]any)
+		m.state[namespace] = ns
 	}
+	return ns
 }
 
-func (m *MemoryBackend) Unlock(_ context.Context, namespace string) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if ch, ok := m.locks[namespace]; ok {
-		close(ch)
-		delete(m.locks, namespace)
+// 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
 	}
-	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/memory_test.go

@@ -2,23 +2,19 @@ package metadatadb_test
 
 import (
 	"testing"
-	"time"
 
 	"github.com/block/cachew/internal/metadatadb"
 	"github.com/block/cachew/internal/metadatadb/metadatadbtest"
 )
 
 func TestMemoryBackend(t *testing.T) {
-	metadatadbtest.Suite(t, func(t *testing.T) metadatadb.Backend {
+	metadatadbtest.Suite(t, func(t *testing.T, n int) []metadatadb.Backend {
 		t.Helper()
-		return metadatadb.NewMemoryBackend()
-	})
-}
-
-func TestMemoryBackendSoak(t *testing.T) {
-	metadatadbtest.Soak(t, metadatadb.NewMemoryBackend(), metadatadbtest.SoakConfig{
-		Duration:    5 * time.Second,
-		Concurrency: 4,
-		NumKeys:     20,
+		backend := metadatadb.NewMemoryBackend()
+		backends := make([]metadatadb.Backend, n)
+		for i := range backends {
+			backends[i] = backend
+		}
+		return backends
 	})
 }

internal/metadatadb/metadatadbtest/soak.go

@@ -97,9 +97,8 @@ func Soak(t *testing.T, backend metadatadb.Backend, config SoakConfig) SoakResul
 	ctx, cancel := context.WithTimeout(ctx, config.Duration+time.Minute)
 	defer cancel()
 
-	cfg := metadatadb.Config{SyncInterval: time.Hour, LockTTL: 5 * time.Second}
-	store := metadatadb.New(ctx, cfg, backend)
-	t.Cleanup(func() { assert.NoError(t, store.Close()) })
+	store := metadatadb.New(ctx, backend)
+	t.Cleanup(func() { assert.NoError(t, store.Close(ctx)) })
 
 	ns := store.Namespace("soak")
 	tr := newTracker()

internal/metadatadb/metadatadbtest/suite.go

@@ -2,10 +2,8 @@ package metadatadbtest
 
 import (
 	"context"
-	"encoding/json"
 	"log/slog"
 	"testing"
-	"time"
 
 	"github.com/alecthomas/assert/v2"
 
@@ -14,63 +12,67 @@ import (
 )
 
 // Suite runs a comprehensive test suite against a metadatadb.Backend implementation.
-func Suite(t *testing.T, newBackend func(t *testing.T) metadatadb.Backend) {
+// The factory must return n backends that share the same underlying storage.
+func Suite(t *testing.T, newBackends func(t *testing.T, n int) []metadatadb.Backend) {
+	one := func(t *testing.T) metadatadb.Backend { return newBackends(t, 1)[0] }
+
 	t.Run("Scalar", func(t *testing.T) {
-		testScalar(t, newBackend(t))
+		testScalar(t, one(t))
 	})
 	t.Run("ScalarDelete", func(t *testing.T) {
-		testScalarDelete(t, newBackend(t))
+		testScalarDelete(t, one(t))
 	})
 	t.Run("ScalarStruct", func(t *testing.T) {
-		testScalarStruct(t, newBackend(t))
+		testScalarStruct(t, one(t))
 	})
 	t.Run("Int", func(t *testing.T) {
-		testInt(t, newBackend(t))
+		testInt(t, one(t))
 	})
 	t.Run("IntArithmetic", func(t *testing.T) {
-		testIntArithmetic(t, newBackend(t))
+		testIntArithmetic(t, one(t))
 	})
 	t.Run("IntDivByZero", func(t *testing.T) {
-		testIntDivByZero(t, newBackend(t))
+		testIntDivByZero(t, one(t))
 	})
 	t.Run("Set", func(t *testing.T) {
-		testSet(t, newBackend(t))
+		testSet(t, one(t))
 	})
 	t.Run("SetRemove", func(t *testing.T) {
-		testSetRemove(t, newBackend(t))
+		testSetRemove(t, one(t))
 	})
 	t.Run("Map", func(t *testing.T) {
-		testMap(t, newBackend(t))
+		testMap(t, one(t))
 	})
 	t.Run("MapDelete", func(t *testing.T) {
-		testMapDelete(t, newBackend(t))
+		testMapDelete(t, one(t))
 	})
 	t.Run("MapStruct", func(t *testing.T) {
-		testMapStruct(t, newBackend(t))
+		testMapStruct(t, one(t))
 	})
 	t.Run("IntMap", func(t *testing.T) {
-		testIntMap(t, newBackend(t))
+		testIntMap(t, one(t))
 	})
 	t.Run("IntMapIncr", func(t *testing.T) {
-		testIntMapIncr(t, newBackend(t))
+		testIntMapIncr(t, one(t))
 	})
 	t.Run("List", func(t *testing.T) {
-		testList(t, newBackend(t))
+		testList(t, one(t))
 	})
 	t.Run("NamespaceIsolation", func(t *testing.T) {
-		testNamespaceIsolation(t, newBackend(t))
+		testNamespaceIsolation(t, one(t))
 	})
 	t.Run("FlushPersists", func(t *testing.T) {
-		testFlushPersists(t, newBackend(t))
+		testFlushPersists(t, one(t))
 	})
 	t.Run("FlushRoundTrip", func(t *testing.T) {
-		testFlushRoundTrip(t, newBackend(t))
+		testFlushRoundTrip(t, one(t))
 	})
 	t.Run("TwoStoresSync", func(t *testing.T) {
-		testTwoStoresSync(t, newBackend(t))
+		testTwoStoresSync(t, one(t))
 	})
-	t.Run("TokenMismatch", func(t *testing.T) {
-		testTokenMismatch(t, newBackend(t))
+	t.Run("TwoBackendsConcurrentOps", func(t *testing.T) {
+		backends := newBackends(t, 2)
+		testTwoBackendsConcurrentOps(t, backends[0], backends[1])
 	})
 }
 
@@ -80,9 +82,8 @@ func testContext() context.Context {
 
 func newStore(t *testing.T, backend metadatadb.Backend) *metadatadb.Store {
 	t.Helper()
-	cfg := metadatadb.Config{SyncInterval: time.Hour, LockTTL: 5 * time.Second}
-	s := metadatadb.New(testContext(), cfg, backend)
-	t.Cleanup(func() { assert.NoError(t, s.Close()) })
+	s := metadatadb.New(testContext(), backend)
+	t.Cleanup(func() { assert.NoError(t, s.Close(testContext())) })
 	return s
 }
 
@@ -406,30 +407,29 @@ func testTwoStoresSync(t *testing.T, backend metadatadb.Backend) {
 	assert.Equal(t, int64(15), metadatadb.NewInt(ns1, "counter").Get())
 }
 
-func testTokenMismatch(t *testing.T, backend metadatadb.Backend) {
+func testTwoBackendsConcurrentOps(t *testing.T, b1, b2 metadatadb.Backend) {
 	ctx := testContext()
+	s1 := metadatadb.New(ctx, b1)
+	t.Cleanup(func() { assert.NoError(t, s1.Close(ctx)) })
+	s2 := metadatadb.New(ctx, b2)
+	t.Cleanup(func() { assert.NoError(t, s2.Close(ctx)) })
+
+	ns1 := s1.Namespace("shared")
+	ns2 := s2.Namespace("shared")
+
+	// Both backends apply ops locally before either flushes.
+	metadatadb.NewInt(ns1, "val").Add(10)
+	metadatadb.NewInt(ns2, "val").Add(-5)
+
+	// Backend 1 flushes first — remote becomes 10.
+	assert.NoError(t, ns1.Flush(ctx))
+
+	// Backend 2 flushes — replays Add(-5) onto remote 10 = 5.
+	assert.NoError(t, ns2.Flush(ctx))
+
+	// Backend 1 reloads to pick up backend 2's contribution.
+	assert.NoError(t, ns1.Flush(ctx))
 
-	// Directly exercise the Backend token semantics.
-	_, token, err := backend.Load(ctx, "test")
-	assert.NoError(t, err)
-	assert.NoError(t, backend.Store(ctx, "test", json.RawMessage(`{"key":"v1"}`), token))
-
-	// Two readers load the same version.
-	_, token1, err := backend.Load(ctx, "test")
-	assert.NoError(t, err)
-	_, token2, err := backend.Load(ctx, "test")
-	assert.NoError(t, err)
-	assert.Equal(t, token1, token2)
-
-	// First writer succeeds.
-	assert.NoError(t, backend.Store(ctx, "test", json.RawMessage(`{"key":"v2"}`), token1))
-
-	// Second writer fails — token is stale.
-	err = backend.Store(ctx, "test", json.RawMessage(`{"key":"v3"}`), token2)
-	assert.IsError(t, err, metadatadb.ErrInvalidToken)
-
-	// Reload and retry succeeds.
-	_, token3, err := backend.Load(ctx, "test")
-	assert.NoError(t, err)
-	assert.NoError(t, backend.Store(ctx, "test", json.RawMessage(`{"key":"v3"}`), token3))
+	assert.Equal(t, int64(5), metadatadb.NewInt(ns1, "val").Get())
+	assert.Equal(t, int64(5), metadatadb.NewInt(ns2, "val").Get())
 }