From de381755a3c144a3993cac0355561076aab5425e Mon Sep 17 00:00:00 2001 From: Alec Thomas Date: Fri, 3 Jul 2026 22:52:02 +1000 Subject: [PATCH] fix: sync s3 metadata writes Persist metadata writes before updating local S3 backend state so storage errors are returned to callers instead of being hidden by deferred flushes. --- internal/cache/tiered.go | 4 +- internal/cache/tiered_test.go | 10 +- internal/metadatadb/api.go | 83 +++++---- internal/metadatadb/metadatadbtest/soak.go | 37 ++-- internal/metadatadb/metadatadbtest/suite.go | 188 +++++++++++--------- internal/metadatadb/s3.go | 146 ++++++--------- internal/strategy/git/git.go | 10 +- internal/strategy/git/repocounts.go | 22 ++- internal/strategy/git/repocounts_test.go | 38 ++-- 9 files changed, 285 insertions(+), 253 deletions(-) diff --git a/internal/cache/tiered.go b/internal/cache/tiered.go index 939ccdad..ea3cff9f 100644 --- a/internal/cache/tiered.go +++ b/internal/cache/tiered.go @@ -137,7 +137,7 @@ func (t Tiered) Delete(ctx context.Context, key Key) error { wg.Wait() err := errors.Join(errs...) if err == nil { - t.etags.Delete(key) + err = errors.Wrap(t.etags.Delete(key), "delete tiered etag") } return err } @@ -518,7 +518,7 @@ func (t *tieredWriter) Close() error { wg.Wait() err := errors.Join(errs...) if err == nil && !t.aborted && t.replaceETag { - t.etags.Set(t.key, t.etag) + err = errors.Wrap(t.etags.Set(t.key, t.etag), "set tiered etag") } return err } diff --git a/internal/cache/tiered_test.go b/internal/cache/tiered_test.go index ae457177..4e7ff1a9 100644 --- a/internal/cache/tiered_test.go +++ b/internal/cache/tiered_test.go @@ -377,7 +377,7 @@ func TestTieredMetadataInvalidatesStaleLowerOnStat(t *testing.T) { key := cache.NewKey("tiered-stale-stat") seedTier(ctx, t, lower, key, []byte("stale"), "stale-etag") seedTier(ctx, t, upper, key, []byte("fresh"), "fresh-etag") - tieredETags(store, "").Set(key, `"fresh-etag"`) + assert.NoError(t, tieredETags(store, "").Set(key, `"fresh-etag"`)) headers, err := tiered.Stat(ctx, key) assert.NoError(t, err) @@ -399,7 +399,7 @@ func TestTieredMetadataInvalidatesStaleLowerBeforeNotModified(t *testing.T) { key := cache.NewKey("tiered-stale-not-modified") seedTier(ctx, t, lower, key, []byte("stale"), "stale-etag") seedTier(ctx, t, upper, key, []byte("fresh"), "fresh-etag") - tieredETags(store, "").Set(key, `"fresh-etag"`) + assert.NoError(t, tieredETags(store, "").Set(key, `"fresh-etag"`)) r, headers, err := tiered.Open(ctx, key, cache.IfNoneMatch(`"stale-etag"`)) assert.NoError(t, err) @@ -425,7 +425,7 @@ func TestTieredMetadataReturnsHardStatErrorBeforeInvalidating(t *testing.T) { key := cache.NewKey("tiered-stale-hard-stat-error") seedTier(ctx, t, lower, key, []byte("lower"), "lower-etag") seedTier(ctx, t, upper, key, []byte("upper"), "upper-etag") - tieredETags(store, "").Set(key, `"upper-etag"`) + assert.NoError(t, tieredETags(store, "").Set(key, `"upper-etag"`)) _, err = tiered.Stat(ctx, key) assert.IsError(t, err, statErr) @@ -447,7 +447,7 @@ func TestTieredMetadataClearsDiscardedConditionalError(t *testing.T) { key := cache.NewKey("tiered-clears-stale-conditional") seedTier(ctx, t, lower, key, []byte("stale"), "stale-etag") - tieredETags(store, "").Set(key, `"fresh-etag"`) + assert.NoError(t, tieredETags(store, "").Set(key, `"fresh-etag"`)) _, _, err = tiered.Open(ctx, key, cache.IfNoneMatch(`"stale-etag"`)) assert.IsError(t, err, os.ErrNotExist) @@ -467,7 +467,7 @@ func TestTieredMetadataAllowsMatchingLower(t *testing.T) { key := cache.NewKey("tiered-matching-lower") seedTier(ctx, t, lower, key, []byte("lower"), "lower-etag") seedTier(ctx, t, upper, key, []byte("upper"), "upper-etag") - tieredETags(store, "").Set(key, `"lower-etag"`) + assert.NoError(t, tieredETags(store, "").Set(key, `"lower-etag"`)) r, headers, err := tiered.Open(ctx, key) assert.NoError(t, err) diff --git a/internal/metadatadb/api.go b/internal/metadatadb/api.go index d4db49f4..44e5c0e4 100644 --- a/internal/metadatadb/api.go +++ b/internal/metadatadb/api.go @@ -1,7 +1,5 @@ -// Package metadatadb provides an eventually consistent metadata store for -// coordinating state across cachew replicas. Mutations are applied to local -// state immediately and synced periodically to a shared backend. Last flush -// wins — the lock serialises all writes. +// Package metadatadb provides a metadata store for coordinating state across +// cachew replicas. package metadatadb import ( @@ -18,16 +16,17 @@ import ( ) // Backend is the pluggable storage layer for the metadata store. Implementations -// handle write operations (Apply), read queries (Query), and persistence (Flush). +// handle write operations (Apply), read queries (Query), and remote state +// refreshes (Flush). type Backend interface { // Apply applies one or more write operations to the given namespace. Apply(ctx context.Context, namespace string, ops ...Op) error // Query executes a read query and unmarshals the result into target. // Target must be a pointer to the expected result type. Query(ctx context.Context, namespace string, q ReadOp, target any) error - // Flush forces pending state to be persisted for the given namespace. + // Flush forces the namespace to refresh from the backend. Flush(ctx context.Context, namespace string) error - // Close shuts down the backend, flushing any pending state. + // Close shuts down the backend. Close(ctx context.Context) error } @@ -75,13 +74,13 @@ type Namespace struct { name string } -// Flush forces an immediate sync with the backend. +// Flush forces an immediate refresh from the backend. func (n *Namespace) Flush(ctx context.Context) error { return errors.Wrap(n.backend.Flush(ctx, n.name), "flush namespace") } -func (n *Namespace) apply(ops ...Op) { - _ = n.backend.Apply(context.Background(), n.name, ops...) //nolint:errcheck // local backends never fail +func (n *Namespace) apply(ops ...Op) error { + return errors.Wrap(n.backend.Apply(context.Background(), n.name, ops...), "apply metadata op") } func (n *Namespace) query(q ReadOp, target any) { @@ -99,8 +98,12 @@ func NewScalar[V any](ns *Namespace, name string) *Scalar[V] { return &Scalar[V]{ns: ns, name: name} } -func (s *Scalar[V]) Set(value V) { s.ns.apply(ScalarSet{Key: s.name, Value: value}) } -func (s *Scalar[V]) Delete() { s.ns.apply(ScalarDelete{Key: s.name}) } +func (s *Scalar[V]) Set(value V) error { + return s.ns.apply(ScalarSet{Key: s.name, Value: value}) +} +func (s *Scalar[V]) Delete() error { + return s.ns.apply(ScalarDelete{Key: s.name}) +} func (s *Scalar[V]) Get() (V, bool) { var result struct { @@ -122,10 +125,18 @@ func NewInt(ns *Namespace, name string) *Int { return &Int{ns: ns, name: name} } -func (i *Int) Set(value int64) { i.ns.apply(IntSet{Key: i.name, Value: value}) } -func (i *Int) Add(delta int64) { i.ns.apply(IntAdd{Key: i.name, Delta: delta}) } -func (i *Int) Mul(factor int64) { i.ns.apply(IntMul{Key: i.name, Factor: factor}) } -func (i *Int) Div(divisor int64) { i.ns.apply(IntDiv{Key: i.name, Divisor: divisor}) } +func (i *Int) Set(value int64) error { + return i.ns.apply(IntSet{Key: i.name, Value: value}) +} +func (i *Int) Add(delta int64) error { + return i.ns.apply(IntAdd{Key: i.name, Delta: delta}) +} +func (i *Int) Mul(factor int64) error { + return i.ns.apply(IntMul{Key: i.name, Factor: factor}) +} +func (i *Int) Div(divisor int64) error { + return i.ns.apply(IntDiv{Key: i.name, Divisor: divisor}) +} func (i *Int) Get() int64 { var v int64 @@ -144,20 +155,20 @@ func NewIntMap[K comparable](ns *Namespace, name string) *IntMap[K] { return &IntMap[K]{ns: ns, name: name} } -func (m *IntMap[K]) Set(key K, value int64) { - m.ns.apply(IntMapSet{Key: m.name, MapKey: key, Value: value}) +func (m *IntMap[K]) Set(key K, value int64) error { + return m.ns.apply(IntMapSet{Key: m.name, MapKey: key, Value: value}) } -func (m *IntMap[K]) Add(key K, delta int64) { - m.ns.apply(IntMapAdd{Key: m.name, MapKey: key, Delta: delta}) +func (m *IntMap[K]) Add(key K, delta int64) error { + return m.ns.apply(IntMapAdd{Key: m.name, MapKey: key, Delta: delta}) } -func (m *IntMap[K]) Mul(key K, factor int64) { - m.ns.apply(IntMapMul{Key: m.name, MapKey: key, Factor: factor}) +func (m *IntMap[K]) Mul(key K, factor int64) error { + return m.ns.apply(IntMapMul{Key: m.name, MapKey: key, Factor: factor}) } -func (m *IntMap[K]) Div(key K, divisor int64) { - m.ns.apply(IntMapDiv{Key: m.name, MapKey: key, Divisor: divisor}) +func (m *IntMap[K]) Div(key K, divisor int64) error { + return m.ns.apply(IntMapDiv{Key: m.name, MapKey: key, Divisor: divisor}) } -func (m *IntMap[K]) Delete(key K) { - m.ns.apply(IntMapDelete{Key: m.name, MapKey: key}) +func (m *IntMap[K]) Delete(key K) error { + return m.ns.apply(IntMapDelete{Key: m.name, MapKey: key}) } func (m *IntMap[K]) Get(key K) int64 { @@ -190,8 +201,12 @@ func NewSet[V comparable](ns *Namespace, name string) *Set[V] { return &Set[V]{ns: ns, name: name} } -func (s *Set[V]) Add(member V) { s.ns.apply(SetAdd{Key: s.name, Member: member}) } -func (s *Set[V]) Remove(member V) { s.ns.apply(SetRemove{Key: s.name, Member: member}) } +func (s *Set[V]) Add(member V) error { + return s.ns.apply(SetAdd{Key: s.name, Member: member}) +} +func (s *Set[V]) Remove(member V) error { + return s.ns.apply(SetRemove{Key: s.name, Member: member}) +} func (s *Set[V]) Contains(member V) bool { var v bool @@ -217,12 +232,12 @@ func NewMap[K comparable, V any](ns *Namespace, name string) *Map[K, V] { return &Map[K, V]{ns: ns, name: name} } -func (m *Map[K, V]) Set(key K, value V) { - m.ns.apply(MapSet{Key: m.name, MapKey: key, Value: value}) +func (m *Map[K, V]) Set(key K, value V) error { + return m.ns.apply(MapSet{Key: m.name, MapKey: key, Value: value}) } -func (m *Map[K, V]) Delete(key K) { - m.ns.apply(MapDelete{Key: m.name, MapKey: key}) +func (m *Map[K, V]) Delete(key K) error { + return m.ns.apply(MapDelete{Key: m.name, MapKey: key}) } func (m *Map[K, V]) Get(key K) (V, bool) { @@ -258,8 +273,8 @@ func NewList[V any](ns *Namespace, name string) *List[V] { return &List[V]{ns: ns, name: name} } -func (l *List[V]) Append(value V) { - l.ns.apply(ListAppend{Key: l.name, Value: value}) +func (l *List[V]) Append(value V) error { + return l.ns.apply(ListAppend{Key: l.name, Value: value}) } func (l *List[V]) Entries() []V { diff --git a/internal/metadatadb/metadatadbtest/soak.go b/internal/metadatadb/metadatadbtest/soak.go index 7c36ff83..4f6aca8a 100644 --- a/internal/metadatadb/metadatadbtest/soak.go +++ b/internal/metadatadb/metadatadbtest/soak.go @@ -150,33 +150,36 @@ func soakWorker( switch { // Chaotic ops (not tracked for invariants). case op < 10: - metadatadb.NewScalar[string](ns, "sc-"+key).Set(fmt.Sprintf("val-%d", rng.IntN(1000))) + recordSoakWrite(result, metadatadb.NewScalar[string](ns, "sc-"+key).Set(fmt.Sprintf("val-%d", rng.IntN(1000)))) case op < 15: - metadatadb.NewScalar[string](ns, "sc-"+key).Delete() + recordSoakWrite(result, metadatadb.NewScalar[string](ns, "sc-"+key).Delete()) case op < 20: - metadatadb.NewInt(ns, "int-"+key).Set(int64(rng.IntN(1000))) + recordSoakWrite(result, metadatadb.NewInt(ns, "int-"+key).Set(int64(rng.IntN(1000)))) case op < 25: - metadatadb.NewSet[string](ns, "set-"+key).Add(fmt.Sprintf("m-%d", rng.IntN(20))) + recordSoakWrite(result, metadatadb.NewSet[string](ns, "set-"+key).Add(fmt.Sprintf("m-%d", rng.IntN(20)))) case op < 28: - metadatadb.NewSet[string](ns, "set-"+key).Remove(fmt.Sprintf("m-%d", rng.IntN(20))) + recordSoakWrite(result, metadatadb.NewSet[string](ns, "set-"+key).Remove(fmt.Sprintf("m-%d", rng.IntN(20)))) case op < 35: - metadatadb.NewMap[string, string](ns, "map-"+key).Set( + recordSoakWrite(result, metadatadb.NewMap[string, string](ns, "map-"+key).Set( fmt.Sprintf("k-%d", rng.IntN(20)), fmt.Sprintf("v-%d", rng.IntN(1000)), - ) + )) // Monotonic ops (tracked for invariant verification). case op < 50: delta := int64(rng.IntN(100) + 1) // always positive - metadatadb.NewInt(ns, "mono-int-"+key).Add(delta) - tr.addCounter("mono-int-"+key, delta) + if recordSoakWrite(result, metadatadb.NewInt(ns, "mono-int-"+key).Add(delta)) { + tr.addCounter("mono-int-"+key, delta) + } case op < 65: member := fmt.Sprintf("m-%d", rng.IntN(50)) - metadatadb.NewSet[string](ns, "mono-set-"+key).Add(member) - tr.addSetMember("mono-set-"+key, member) + if recordSoakWrite(result, metadatadb.NewSet[string](ns, "mono-set-"+key).Add(member)) { + tr.addSetMember("mono-set-"+key, member) + } case op < 75: - metadatadb.NewList[string](ns, "mono-list-"+key).Append(fmt.Sprintf("e-%d", rng.IntN(1000))) - tr.appendList("mono-list-" + key) + if recordSoakWrite(result, metadatadb.NewList[string](ns, "mono-list-"+key).Append(fmt.Sprintf("e-%d", rng.IntN(1000)))) { + tr.appendList("mono-list-" + key) + } // Reads. case op < 80: @@ -200,6 +203,14 @@ func soakWorker( } } +func recordSoakWrite(result *SoakResult, err error) bool { + if err == nil { + return true + } + atomic.AddInt64(&result.Errors, 1) + return false +} + func verifyMonotonicInvariants(t *testing.T, ns *metadatadb.Namespace, tr *tracker) { t.Helper() tr.mu.Lock() diff --git a/internal/metadatadb/metadatadbtest/suite.go b/internal/metadatadb/metadatadbtest/suite.go index 4698a8e2..344b09eb 100644 --- a/internal/metadatadb/metadatadbtest/suite.go +++ b/internal/metadatadb/metadatadbtest/suite.go @@ -3,6 +3,7 @@ package metadatadbtest import ( "context" "log/slog" + "strings" "testing" "github.com/alecthomas/assert/v2" @@ -64,6 +65,10 @@ func Suite(t *testing.T, newBackends func(t *testing.T, n int) []metadatadb.Back t.Run("FlushPersists", func(t *testing.T) { testFlushPersists(t, one(t)) }) + t.Run("WritePersists", func(t *testing.T) { + backends := newBackends(t, 2) + testWritePersists(t, backends[0], backends[1]) + }) t.Run("FlushRoundTrip", func(t *testing.T) { testFlushRoundTrip(t, one(t)) }) @@ -87,20 +92,26 @@ func newStore(t *testing.T, backend metadatadb.Backend) *metadatadb.Store { return s } +func namespace(t *testing.T, s *metadatadb.Store, name string) *metadatadb.Namespace { + t.Helper() + prefix := strings.NewReplacer("/", "-").Replace(t.Name()) + return s.Namespace(prefix + "-" + name) +} + func testScalar(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") sc := metadatadb.NewScalar[string](ns, "greeting") _, ok := sc.Get() assert.False(t, ok) - sc.Set("hello") + assert.NoError(t, sc.Set("hello")) v, ok := sc.Get() assert.True(t, ok) assert.Equal(t, "hello", v) - sc.Set("world") + assert.NoError(t, sc.Set("world")) v, ok = sc.Get() assert.True(t, ok) assert.Equal(t, "world", v) @@ -108,15 +119,15 @@ func testScalar(t *testing.T, backend metadatadb.Backend) { func testScalarDelete(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") sc := metadatadb.NewScalar[int](ns, "counter") - sc.Set(42) + assert.NoError(t, sc.Set(42)) v, ok := sc.Get() assert.True(t, ok) assert.Equal(t, 42, v) - sc.Delete() + assert.NoError(t, sc.Delete()) _, ok = sc.Get() assert.False(t, ok) } @@ -128,10 +139,10 @@ type testConfig struct { func testScalarStruct(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") sc := metadatadb.NewScalar[testConfig](ns, "config") - sc.Set(testConfig{Host: "localhost", Port: 8080}) + assert.NoError(t, sc.Set(testConfig{Host: "localhost", Port: 8080})) v, ok := sc.Get() assert.True(t, ok) assert.Equal(t, testConfig{Host: "localhost", Port: 8080}, v) @@ -139,55 +150,55 @@ func testScalarStruct(t *testing.T, backend metadatadb.Backend) { func testInt(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") i := metadatadb.NewInt(ns, "counter") assert.Equal(t, int64(0), i.Get()) - i.Set(10) + assert.NoError(t, i.Set(10)) assert.Equal(t, int64(10), i.Get()) - i.Add(5) + assert.NoError(t, i.Add(5)) assert.Equal(t, int64(15), i.Get()) } func testIntArithmetic(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") i := metadatadb.NewInt(ns, "val") - i.Set(100) - i.Mul(3) + assert.NoError(t, i.Set(100)) + assert.NoError(t, i.Mul(3)) assert.Equal(t, int64(300), i.Get()) - i.Div(4) + assert.NoError(t, i.Div(4)) assert.Equal(t, int64(75), i.Get()) - i.Add(-25) + assert.NoError(t, i.Add(-25)) assert.Equal(t, int64(50), i.Get()) } func testIntDivByZero(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") i := metadatadb.NewInt(ns, "val") - i.Set(42) - i.Div(0) + assert.NoError(t, i.Set(42)) + assert.NoError(t, i.Div(0)) assert.Equal(t, int64(42), i.Get()) } func testSet(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") set := metadatadb.NewSet[string](ns, "repos") assert.Equal(t, []string(nil), set.Members()) assert.False(t, set.Contains("a")) - set.Add("a") - set.Add("b") - set.Add("a") // duplicate + assert.NoError(t, set.Add("a")) + assert.NoError(t, set.Add("b")) + assert.NoError(t, set.Add("a")) assert.False(t, set.Contains("c")) assert.Equal(t, []string{"a", "b"}, set.Members()) @@ -195,50 +206,50 @@ func testSet(t *testing.T, backend metadatadb.Backend) { func testSetRemove(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") set := metadatadb.NewSet[string](ns, "tags") - set.Add("x") - set.Add("y") - set.Remove("x") - set.Remove("z") // non-existent is a no-op + assert.NoError(t, set.Add("x")) + assert.NoError(t, set.Add("y")) + assert.NoError(t, set.Remove("x")) + assert.NoError(t, set.Remove("z")) assert.Equal(t, []string{"y"}, set.Members()) } func testMap(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") m := metadatadb.NewMap[string, string](ns, "labels") _, ok := m.Get("a") assert.False(t, ok) assert.Equal(t, []string(nil), m.Keys()) - m.Set("a", "alpha") - m.Set("b", "beta") + assert.NoError(t, m.Set("a", "alpha")) + assert.NoError(t, m.Set("b", "beta")) assert.Equal(t, map[string]string{"a": "alpha", "b": "beta"}, m.Entries()) } func testMapDelete(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") m := metadatadb.NewMap[string, int](ns, "ports") - m.Set("http", 80) - m.Set("https", 443) - m.Delete("http") + assert.NoError(t, m.Set("http", 80)) + assert.NoError(t, m.Set("https", 443)) + assert.NoError(t, m.Delete("http")) assert.Equal(t, map[string]int{"https": 443}, m.Entries()) } func testMapStruct(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") m := metadatadb.NewMap[string, testConfig](ns, "services") - m.Set("web", testConfig{Host: "web.local", Port: 8080}) + assert.NoError(t, m.Set("web", testConfig{Host: "web.local", Port: 8080})) v, ok := m.Get("web") assert.True(t, ok) @@ -247,44 +258,44 @@ func testMapStruct(t *testing.T, backend metadatadb.Backend) { func testIntMap(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") m := metadatadb.NewIntMap[string](ns, "clones") assert.Equal(t, int64(0), m.Get("repo-a")) - m.Set("repo-a", 10) - m.Set("repo-b", 20) + assert.NoError(t, m.Set("repo-a", 10)) + assert.NoError(t, m.Set("repo-b", 20)) assert.Equal(t, map[string]int64{"repo-a": 10, "repo-b": 20}, m.Entries()) - m.Delete("repo-a") + assert.NoError(t, m.Delete("repo-a")) assert.Equal(t, map[string]int64{"repo-b": 20}, m.Entries()) } func testIntMapIncr(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") m := metadatadb.NewIntMap[string](ns, "histogram") - m.Add("repo-a", 1) - m.Add("repo-a", 1) - m.Add("repo-b", 5) - m.Add("repo-a", 3) + assert.NoError(t, m.Add("repo-a", 1)) + assert.NoError(t, m.Add("repo-a", 1)) + assert.NoError(t, m.Add("repo-b", 5)) + assert.NoError(t, m.Add("repo-a", 3)) assert.Equal(t, map[string]int64{"repo-a": 5, "repo-b": 5}, m.Entries()) } func testList(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") l := metadatadb.NewList[string](ns, "log") assert.Equal(t, 0, l.Len()) assert.Equal(t, []string(nil), l.Entries()) - l.Append("first") - l.Append("second") - l.Append("third") + assert.NoError(t, l.Append("first")) + assert.NoError(t, l.Append("second")) + assert.NoError(t, l.Append("third")) assert.Equal(t, 3, l.Len()) assert.Equal(t, []string{"first", "second", "third"}, l.Entries()) @@ -292,14 +303,14 @@ func testList(t *testing.T, backend metadatadb.Backend) { func testNamespaceIsolation(t *testing.T, backend metadatadb.Backend) { s := newStore(t, backend) - ns1 := s.Namespace("ns1") - ns2 := s.Namespace("ns2") + ns1 := namespace(t, s, "ns1") + ns2 := namespace(t, s, "ns2") sc1 := metadatadb.NewScalar[string](ns1, "key") sc2 := metadatadb.NewScalar[string](ns2, "key") - sc1.Set("from-ns1") - sc2.Set("from-ns2") + assert.NoError(t, sc1.Set("from-ns1")) + assert.NoError(t, sc2.Set("from-ns2")) v1, ok := sc1.Get() assert.True(t, ok) @@ -313,15 +324,15 @@ func testNamespaceIsolation(t *testing.T, backend metadatadb.Backend) { func testFlushPersists(t *testing.T, backend metadatadb.Backend) { ctx := testContext() s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") sc := metadatadb.NewScalar[string](ns, "key") - sc.Set("value") + assert.NoError(t, sc.Set("value")) assert.NoError(t, ns.Flush(ctx)) // Create a new store against the same backend — should see the value. s2 := newStore(t, backend) - ns2 := s2.Namespace("test") + ns2 := namespace(t, s2, "test") sc2 := metadatadb.NewScalar[string](ns2, "key") // Flush to load remote state. @@ -332,34 +343,48 @@ func testFlushPersists(t *testing.T, backend metadatadb.Backend) { assert.Equal(t, "value", v) } +func testWritePersists(t *testing.T, b1, b2 metadatadb.Backend) { + ctx := testContext() + s1 := newStore(t, b1) + ns1 := namespace(t, s1, "test") + assert.NoError(t, metadatadb.NewScalar[string](ns1, "key").Set("value")) + + s2 := newStore(t, b2) + ns2 := namespace(t, s2, "test") + assert.NoError(t, ns2.Flush(ctx)) + + v, ok := metadatadb.NewScalar[string](ns2, "key").Get() + assert.True(t, ok) + assert.Equal(t, "value", v) +} + func testFlushRoundTrip(t *testing.T, backend metadatadb.Backend) { ctx := testContext() s := newStore(t, backend) - ns := s.Namespace("test") + ns := namespace(t, s, "test") - // Write all data structure types, flush, then read back from a fresh store. - metadatadb.NewScalar[string](ns, "sc").Set("hello") - metadatadb.NewScalar[testConfig](ns, "cfg").Set(testConfig{Host: "h", Port: 1}) + assert.NoError(t, metadatadb.NewScalar[string](ns, "sc").Set("hello")) + assert.NoError(t, metadatadb.NewScalar[testConfig](ns, "cfg").Set(testConfig{Host: "h", Port: 1})) i := metadatadb.NewInt(ns, "counter") - i.Set(10) - i.Add(5) + assert.NoError(t, i.Set(10)) + assert.NoError(t, i.Add(5)) set := metadatadb.NewSet[string](ns, "tags") - set.Add("a") - set.Add("b") + assert.NoError(t, set.Add("a")) + assert.NoError(t, set.Add("b")) m := metadatadb.NewMap[string, int](ns, "ports") - m.Set("http", 80) + assert.NoError(t, m.Set("http", 80)) im := metadatadb.NewIntMap[string](ns, "clones") - im.Add("repo-a", 3) - im.Add("repo-b", 7) + assert.NoError(t, im.Add("repo-a", 3)) + assert.NoError(t, im.Add("repo-b", 7)) l := metadatadb.NewList[string](ns, "log") - l.Append("entry1") - l.Append("entry2") + assert.NoError(t, l.Append("entry1")) + assert.NoError(t, l.Append("entry2")) assert.NoError(t, ns.Flush(ctx)) // Fresh store, same backend. s2 := newStore(t, backend) - ns2 := s2.Namespace("test") + ns2 := namespace(t, s2, "test") assert.NoError(t, ns2.Flush(ctx)) sc, ok := metadatadb.NewScalar[string](ns2, "sc").Get() @@ -382,12 +407,11 @@ func testTwoStoresSync(t *testing.T, backend metadatadb.Backend) { s1 := newStore(t, backend) s2 := newStore(t, backend) - ns1 := s1.Namespace("shared") - ns2 := s2.Namespace("shared") + ns1 := namespace(t, s1, "shared") + ns2 := namespace(t, s2, "shared") - // Store 1 writes and flushes. - metadatadb.NewScalar[string](ns1, "owner").Set("store1") - metadatadb.NewInt(ns1, "counter").Add(10) + assert.NoError(t, metadatadb.NewScalar[string](ns1, "owner").Set("store1")) + assert.NoError(t, metadatadb.NewInt(ns1, "counter").Add(10)) assert.NoError(t, ns1.Flush(ctx)) // Store 2 loads. @@ -398,8 +422,7 @@ func testTwoStoresSync(t *testing.T, backend metadatadb.Backend) { assert.Equal(t, "store1", v) assert.Equal(t, int64(10), metadatadb.NewInt(ns2, "counter").Get()) - // Store 2 writes and flushes. - metadatadb.NewInt(ns2, "counter").Add(5) + assert.NoError(t, metadatadb.NewInt(ns2, "counter").Add(5)) assert.NoError(t, ns2.Flush(ctx)) // Store 1 loads. @@ -414,12 +437,11 @@ func testTwoBackendsConcurrentOps(t *testing.T, b1, b2 metadatadb.Backend) { s2 := metadatadb.New(ctx, b2) t.Cleanup(func() { assert.NoError(t, s2.Close(ctx)) }) - ns1 := s1.Namespace("shared") - ns2 := s2.Namespace("shared") + ns1 := namespace(t, s1, "shared") + ns2 := namespace(t, s2, "shared") - // Both backends apply ops locally before either flushes. - metadatadb.NewInt(ns1, "val").Add(10) - metadatadb.NewInt(ns2, "val").Add(-5) + assert.NoError(t, metadatadb.NewInt(ns1, "val").Add(10)) + assert.NoError(t, metadatadb.NewInt(ns2, "val").Add(-5)) // Backend 1 flushes first — remote becomes 10. assert.NoError(t, ns1.Flush(ctx)) diff --git a/internal/metadatadb/s3.go b/internal/metadatadb/s3.go index 19800114..f8fe6c76 100644 --- a/internal/metadatadb/s3.go +++ b/internal/metadatadb/s3.go @@ -18,33 +18,30 @@ import ( // RegisterS3 registers the S3 metadata backend. The clientProvider supplies the // shared minio client constructed from the global s3 config block. func RegisterS3(r *Registry, clientProvider s3client.ClientProvider) { - Register(r, "s3", "Stores metadata state in S3 with periodic sync", + Register(r, "s3", "Stores metadata state in S3 with synchronous writes", func(ctx context.Context, config S3BackendConfig) (*S3Backend, error) { return NewS3Backend(ctx, clientProvider, config) }, ) } -// S3Backend stores metadata state as JSON objects in S3 with periodic sync. -// Writes are applied to local state immediately and queued for the next flush. +// S3Backend stores metadata state as JSON objects in S3. +// Writes are applied to S3 before local state is updated. // Locking uses a separate lock object with TTL-based expiry for stale lock // recovery. The idempotence token maps to the S3 object ETag. type S3Backend struct { - client *minio.Client - bucket string - lockTTL time.Duration - syncInterval time.Duration - mu sync.Mutex - ns map[string]*s3Namespace - ctx context.Context - cancel context.CancelFunc + client *minio.Client + bucket string + lockTTL time.Duration + mu sync.Mutex + ns map[string]*s3Namespace } // S3BackendConfig configures the S3 metadata backend. type S3BackendConfig struct { Bucket string `hcl:"bucket" help:"S3 bucket name."` LockTTL time.Duration `hcl:"lock-ttl,optional" help:"TTL for namespace locks." default:"30s"` - SyncInterval time.Duration `hcl:"sync-interval,optional" help:"Interval between periodic syncs." default:"30s"` + SyncInterval time.Duration `hcl:"sync-interval,optional" help:"Deprecated; writes are synchronous."` } // s3MetadataPrefix is the fixed key prefix for all metadata objects in S3. @@ -56,9 +53,6 @@ func NewS3Backend(ctx context.Context, clientProvider s3client.ClientProvider, c if config.LockTTL == 0 { config.LockTTL = 30 * time.Second } - if config.SyncInterval == 0 { - config.SyncInterval = 30 * time.Second - } client, err := clientProvider() if err != nil { return nil, errors.Wrap(err, "create S3 client") @@ -72,17 +66,13 @@ func NewS3Backend(ctx context.Context, clientProvider s3client.ClientProvider, c } logging.FromContext(ctx).InfoContext(ctx, "Constructing S3 metadata backend", - "bucket", config.Bucket, "prefix", s3MetadataPrefix, "lock-ttl", config.LockTTL, "sync-interval", config.SyncInterval) + "bucket", config.Bucket, "prefix", s3MetadataPrefix, "lock-ttl", config.LockTTL) - ctx, cancel := context.WithCancel(ctx) return &S3Backend{ - client: client, - bucket: config.Bucket, - lockTTL: config.LockTTL, - syncInterval: config.SyncInterval, - ns: make(map[string]*s3Namespace), - ctx: ctx, - cancel: cancel, + client: client, + bucket: config.Bucket, + lockTTL: config.LockTTL, + ns: make(map[string]*s3Namespace), }, nil } @@ -96,22 +86,16 @@ func (s *S3Backend) namespace(name string) *s3Namespace { backend: s, name: name, state: make(map[string]any), - done: make(chan struct{}), } - go ns.syncLoop() s.ns[name] = ns return ns } -func (s *S3Backend) Apply(_ context.Context, namespace string, ops ...Op) error { - ns := s.namespace(namespace) - ns.mu.Lock() - defer ns.mu.Unlock() - for _, o := range ops { - applyOp(ns.state, o) +func (s *S3Backend) Apply(ctx context.Context, namespace string, ops ...Op) error { + if len(ops) == 0 { + return nil } - ns.pending = append(ns.pending, ops...) - return nil + return s.namespace(namespace).apply(ctx, ops) } func (s *S3Backend) Query(_ context.Context, namespace string, q ReadOp, target any) error { @@ -123,18 +107,10 @@ func (s *S3Backend) Query(_ context.Context, namespace string, q ReadOp, target } func (s *S3Backend) Flush(ctx context.Context, namespace string) error { - return s.namespace(namespace).doSync(ctx) + return s.namespace(namespace).reload(ctx) } -func (s *S3Backend) Close(_ context.Context) error { - s.cancel() - s.mu.Lock() - defer s.mu.Unlock() - for _, ns := range s.ns { - <-ns.done - } - return nil -} +func (s *S3Backend) Close(_ context.Context) error { return nil } // S3 object key helpers @@ -248,50 +224,58 @@ type s3Namespace struct { name string mu sync.RWMutex state map[string]any - pending []Op syncMu sync.Mutex - done chan struct{} } const maxTokenRetries = 3 -func (n *s3Namespace) doSync(ctx context.Context) error { +func (n *s3Namespace) reload(ctx context.Context) error { n.syncMu.Lock() defer n.syncMu.Unlock() + remote, err := n.load(ctx) + if err != nil { + return err + } + n.mu.Lock() - pending := n.pending - n.pending = nil + n.state = remote n.mu.Unlock() + return nil +} - if len(pending) > 0 { - if err := n.backend.lockNamespace(ctx, n.name); err != nil { - n.restorePending(pending) - return errors.Wrap(err, "lock namespace") - } - defer func() { - if err := n.backend.unlockNamespace(ctx, n.name); err != nil { - logging.FromContext(ctx).WarnContext(ctx, "unlock failed", "namespace", n.name, "error", err) - } - }() +func (n *s3Namespace) apply(ctx context.Context, ops []Op) error { + n.syncMu.Lock() + defer n.syncMu.Unlock() + + if err := n.backend.lockNamespace(ctx, n.name); err != nil { + return errors.Wrap(err, "lock namespace") } - remote, err := n.loadReplayStore(ctx, pending) + remote, err := n.loadReplayStore(ctx, ops) + unlockErr := n.backend.unlockNamespace(ctx, n.name) if err != nil { - n.restorePending(pending) return err } + if unlockErr != nil { + return errors.Wrap(unlockErr, "unlock namespace") + } n.mu.Lock() n.state = remote - for _, o := range n.pending { - applyOp(n.state, o) - } n.mu.Unlock() return nil } +func (n *s3Namespace) load(ctx context.Context) (map[string]any, error) { + data, _, err := n.backend.load(ctx, n.name) + if err != nil { + return nil, errors.Wrap(err, "load namespace") + } + return unmarshalState(data) +} + func (n *s3Namespace) loadReplayStore(ctx context.Context, pending []Op) (map[string]any, error) { for range maxTokenRetries { remote, err := n.tryLoadReplayStore(ctx, pending) @@ -309,11 +293,9 @@ func (n *s3Namespace) tryLoadReplayStore(ctx context.Context, pending []Op) (map return nil, errors.Wrap(err, "load namespace") } - remote := make(map[string]any) - if data != nil { - if err := json.Unmarshal(data, &remote); err != nil { - return nil, errors.Wrap(err, "unmarshal state") - } + remote, err := unmarshalState(data) + if err != nil { + return nil, err } for _, o := range pending { @@ -333,28 +315,12 @@ func (n *s3Namespace) tryLoadReplayStore(ctx context.Context, pending []Op) (map return remote, nil } -func (n *s3Namespace) restorePending(ops []Op) { - n.mu.Lock() - defer n.mu.Unlock() - n.pending = append(ops, n.pending...) -} - -func (n *s3Namespace) syncLoop() { - defer close(n.done) - ctx := n.backend.ctx - logger := logging.FromContext(ctx).With("namespace", n.name) - ticker := time.NewTicker(n.backend.syncInterval) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - if err := n.doSync(ctx); err != nil { - logger.WarnContext(ctx, "sync failed", "error", err) - } - } +func unmarshalState(data json.RawMessage) (map[string]any, error) { + state := make(map[string]any) + if data == nil { + return state, nil } + return state, errors.Wrap(json.Unmarshal(data, &state), "unmarshal state") } func isNotFound(err error) bool { diff --git a/internal/strategy/git/git.go b/internal/strategy/git/git.go index 986ea713..e6e941cc 100644 --- a/internal/strategy/git/git.go +++ b/internal/strategy/git/git.go @@ -207,7 +207,11 @@ func (s *Strategy) SetMetadataStore(store *metadatadb.Store) { logging.FromContext(s.ctx).InfoContext(s.ctx, "Per-repo clone histogram enabled", "retention_days", s.repoCounts.retentionDays) s.scheduler.SubmitPeriodicJob("repo-counts-reaper", "reap-repo-counts", defaultRepoCountsReapInterval, func(ctx context.Context) error { - if deleted := s.repoCounts.Reap(); deleted > 0 { + deleted, err := s.repoCounts.Reap() + if err != nil { + return err + } + if deleted > 0 { logging.FromContext(ctx).InfoContext(ctx, "Reaped stale repo clone counts", "deleted", deleted) } return nil @@ -337,7 +341,9 @@ func (s *Strategy) handleGitRequest(w http.ResponseWriter, r *http.Request, host if isClone, cerr := RequestIsClone(pathValue, r); cerr != nil { logger.WarnContext(ctx, "Failed to inspect upload-pack body for clone counting", "error", cerr) } else if isClone { - s.repoCounts.IncrementClone(upstreamURL) + if err := s.repoCounts.IncrementClone(upstreamURL); err != nil { + logger.WarnContext(ctx, "Failed to increment repo clone count", "error", err) + } } state := repo.State() diff --git a/internal/strategy/git/repocounts.go b/internal/strategy/git/repocounts.go index 24e7919f..f393b98a 100644 --- a/internal/strategy/git/repocounts.go +++ b/internal/strategy/git/repocounts.go @@ -44,11 +44,11 @@ func NewRepoCounts(ns *metadatadb.Namespace) *RepoCounts { } // IncrementClone bumps today's bucket (UTC) for upstreamURL. -func (r *RepoCounts) IncrementClone(upstreamURL string) { +func (r *RepoCounts) IncrementClone(upstreamURL string) error { if r == nil || upstreamURL == "" { - return + return nil } - r.counts.Add(repoCountsKey(upstreamURL, r.now()), 1) + return errors.Wrap(r.counts.Add(repoCountsKey(upstreamURL, r.now()), 1), "increment repo count") } // uploadPackBodyInspectLimit bounds CPU spend on hostile bodies; real bodies @@ -161,29 +161,33 @@ func (r *RepoCounts) TopRepos(windowDays, limit int) []RepoCount { // Reap deletes buckets older than the retention window and any malformed keys, // returning the number of entries deleted. -func (r *RepoCounts) Reap() int { +func (r *RepoCounts) Reap() (int, error) { if r == nil { - return 0 + return 0, nil } entries := r.counts.Entries() if len(entries) == 0 { - return 0 + return 0, nil } cutoff := r.now().UTC().AddDate(0, 0, -r.retentionDays).Truncate(24 * time.Hour) var deleted int for k := range entries { _, day, ok := splitRepoCountsKey(k) if !ok { - r.counts.Delete(k) + if err := r.counts.Delete(k); err != nil { + return deleted, errors.Wrap(err, "delete malformed repo count") + } deleted++ continue } if day.Before(cutoff) { - r.counts.Delete(k) + if err := r.counts.Delete(k); err != nil { + return deleted, errors.Wrap(err, "delete stale repo count") + } deleted++ } } - return deleted + return deleted, nil } func repoCountsKey(upstreamURL string, now time.Time) string { diff --git a/internal/strategy/git/repocounts_test.go b/internal/strategy/git/repocounts_test.go index 3ce51e7f..3bb5ee06 100644 --- a/internal/strategy/git/repocounts_test.go +++ b/internal/strategy/git/repocounts_test.go @@ -31,15 +31,19 @@ func newTestRepoCounts(t *testing.T, now func() time.Time) *RepoCounts { func TestRepoCountsNilSafe(t *testing.T) { var rc *RepoCounts - rc.IncrementClone("https://github.com/foo/bar") - assert.Equal(t, 0, rc.Reap()) + assert.NoError(t, rc.IncrementClone("https://github.com/foo/bar")) + deleted, err := rc.Reap() + assert.NoError(t, err) + assert.Equal(t, 0, deleted) assert.Zero(t, len(rc.TopRepos(0, 0))) assert.Zero(t, NewRepoCounts(nil)) } func TestRepoCountsReapEmpty(t *testing.T) { rc := newTestRepoCounts(t, nil) - assert.Equal(t, 0, rc.Reap(), "reap on an empty namespace should report zero deletions") + deleted, err := rc.Reap() + assert.NoError(t, err) + assert.Equal(t, 0, deleted, "reap on an empty namespace should report zero deletions") } func TestRepoCountsIncrementAndAggregate(t *testing.T) { @@ -47,15 +51,15 @@ func TestRepoCountsIncrementAndAggregate(t *testing.T) { rc := newTestRepoCounts(t, func() time.Time { return clock }) for range 5 { - rc.IncrementClone("https://github.com/foo/popular") + assert.NoError(t, rc.IncrementClone("https://github.com/foo/popular")) } for range 2 { - rc.IncrementClone("https://github.com/foo/quiet") + assert.NoError(t, rc.IncrementClone("https://github.com/foo/quiet")) } // Bump the popular repo on a previous day too. clock = clock.AddDate(0, 0, -1) for range 3 { - rc.IncrementClone("https://github.com/foo/popular") + assert.NoError(t, rc.IncrementClone("https://github.com/foo/popular")) } clock = clock.AddDate(0, 0, 1) @@ -73,11 +77,11 @@ func TestRepoCountsWindowFilter(t *testing.T) { // 10 days ago: only "old" repo gets hits. clock = time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC) for range 4 { - rc.IncrementClone("https://github.com/foo/old") + assert.NoError(t, rc.IncrementClone("https://github.com/foo/old")) } // Today: only "new" repo gets hits. clock = time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) - rc.IncrementClone("https://github.com/foo/new") + assert.NoError(t, rc.IncrementClone("https://github.com/foo/new")) all := rc.TopRepos(0, 0) assert.Equal(t, 2, len(all), "no window includes both repos") @@ -92,7 +96,7 @@ func TestRepoCountsLimit(t *testing.T) { for i, repo := range []string{"alpha", "bravo", "charlie", "delta"} { for j := 0; j <= i; j++ { - rc.IncrementClone("https://github.com/foo/" + repo) + assert.NoError(t, rc.IncrementClone("https://github.com/foo/"+repo)) } } @@ -110,13 +114,15 @@ func TestRepoCountsReap(t *testing.T) { // Old entry: 100 days ago. clock = time.Date(2026, 1, 25, 12, 0, 0, 0, time.UTC) - rc.IncrementClone("https://github.com/foo/ancient") + assert.NoError(t, rc.IncrementClone("https://github.com/foo/ancient")) // Recent entry: 5 days ago. clock = time.Date(2026, 4, 30, 12, 0, 0, 0, time.UTC) - rc.IncrementClone("https://github.com/foo/fresh") + assert.NoError(t, rc.IncrementClone("https://github.com/foo/fresh")) clock = time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) - assert.Equal(t, 1, rc.Reap(), "one stale bucket should be deleted") + deleted, err := rc.Reap() + assert.NoError(t, err) + assert.Equal(t, 1, deleted, "one stale bucket should be deleted") remaining := rc.TopRepos(0, 0) assert.Equal(t, []RepoCount{{Repo: "https://github.com/foo/fresh", Count: 1}}, remaining) @@ -127,10 +133,12 @@ func TestRepoCountsReapMalformedKeys(t *testing.T) { rc := newTestRepoCounts(t, func() time.Time { return clock }) // Inject a malformed key directly via Add. - rc.counts.Add("not-a-real-key", 7) - rc.IncrementClone("https://github.com/foo/valid") + assert.NoError(t, rc.counts.Add("not-a-real-key", 7)) + assert.NoError(t, rc.IncrementClone("https://github.com/foo/valid")) - assert.Equal(t, 1, rc.Reap(), "the malformed key should be deleted, the valid one preserved") + deleted, err := rc.Reap() + assert.NoError(t, err) + assert.Equal(t, 1, deleted, "the malformed key should be deleted, the valid one preserved") remaining := rc.TopRepos(0, 0) assert.Equal(t, []RepoCount{{Repo: "https://github.com/foo/valid", Count: 1}}, remaining) }