Skip to content

Commit cb44752

Browse files
alecthomasalecthomas
authored
fix: sync s3 metadata writes (#382)
Persist metadata writes before updating local S3 backend state so storage errors are returned to callers instead of being hidden by deferred flushes.
1 parent d0785bd commit cb44752

9 files changed

Lines changed: 285 additions & 253 deletions

File tree

‎internal/cache/tiered.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -137,7 +137,7 @@ func (t Tiered) Delete(ctx context.Context, key Key) error {
137137
wg.Wait()
138138
err := errors.Join(errs...)
139139
if err == nil {
140-
t.etags.Delete(key)
140+
err = errors.Wrap(t.etags.Delete(key), "delete tiered etag")
141141
}
142142
return err
143143
}
@@ -518,7 +518,7 @@ func (t *tieredWriter) Close() error {
518518
wg.Wait()
519519
err := errors.Join(errs...)
520520
if err == nil && !t.aborted && t.replaceETag {
521-
t.etags.Set(t.key, t.etag)
521+
err = errors.Wrap(t.etags.Set(t.key, t.etag), "set tiered etag")
522522
}
523523
return err
524524
}

‎internal/cache/tiered_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -377,7 +377,7 @@ func TestTieredMetadataInvalidatesStaleLowerOnStat(t *testing.T) {
377377
key := cache.NewKey("tiered-stale-stat")
378378
seedTier(ctx, t, lower, key, []byte("stale"), "stale-etag")
379379
seedTier(ctx, t, upper, key, []byte("fresh"), "fresh-etag")
380-
tieredETags(store, "").Set(key, `"fresh-etag"`)
380+
assert.NoError(t, tieredETags(store, "").Set(key, `"fresh-etag"`))
381381

382382
headers, err := tiered.Stat(ctx, key)
383383
assert.NoError(t, err)
@@ -399,7 +399,7 @@ func TestTieredMetadataInvalidatesStaleLowerBeforeNotModified(t *testing.T) {
399399
key := cache.NewKey("tiered-stale-not-modified")
400400
seedTier(ctx, t, lower, key, []byte("stale"), "stale-etag")
401401
seedTier(ctx, t, upper, key, []byte("fresh"), "fresh-etag")
402-
tieredETags(store, "").Set(key, `"fresh-etag"`)
402+
assert.NoError(t, tieredETags(store, "").Set(key, `"fresh-etag"`))
403403

404404
r, headers, err := tiered.Open(ctx, key, cache.IfNoneMatch(`"stale-etag"`))
405405
assert.NoError(t, err)
@@ -425,7 +425,7 @@ func TestTieredMetadataReturnsHardStatErrorBeforeInvalidating(t *testing.T) {
425425
key := cache.NewKey("tiered-stale-hard-stat-error")
426426
seedTier(ctx, t, lower, key, []byte("lower"), "lower-etag")
427427
seedTier(ctx, t, upper, key, []byte("upper"), "upper-etag")
428-
tieredETags(store, "").Set(key, `"upper-etag"`)
428+
assert.NoError(t, tieredETags(store, "").Set(key, `"upper-etag"`))
429429

430430
_, err = tiered.Stat(ctx, key)
431431
assert.IsError(t, err, statErr)
@@ -447,7 +447,7 @@ func TestTieredMetadataClearsDiscardedConditionalError(t *testing.T) {
447447

448448
key := cache.NewKey("tiered-clears-stale-conditional")
449449
seedTier(ctx, t, lower, key, []byte("stale"), "stale-etag")
450-
tieredETags(store, "").Set(key, `"fresh-etag"`)
450+
assert.NoError(t, tieredETags(store, "").Set(key, `"fresh-etag"`))
451451

452452
_, _, err = tiered.Open(ctx, key, cache.IfNoneMatch(`"stale-etag"`))
453453
assert.IsError(t, err, os.ErrNotExist)
@@ -467,7 +467,7 @@ func TestTieredMetadataAllowsMatchingLower(t *testing.T) {
467467
key := cache.NewKey("tiered-matching-lower")
468468
seedTier(ctx, t, lower, key, []byte("lower"), "lower-etag")
469469
seedTier(ctx, t, upper, key, []byte("upper"), "upper-etag")
470-
tieredETags(store, "").Set(key, `"lower-etag"`)
470+
assert.NoError(t, tieredETags(store, "").Set(key, `"lower-etag"`))
471471

472472
r, headers, err := tiered.Open(ctx, key)
473473
assert.NoError(t, err)

‎internal/metadatadb/api.go‎

Lines changed: 49 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,5 @@
1-
// Package metadatadb provides an eventually consistent metadata store for
2-
// coordinating state across cachew replicas. Mutations are applied to local
3-
// state immediately and synced periodically to a shared backend. Last flush
4-
// wins — the lock serialises all writes.
1+
// Package metadatadb provides a metadata store for coordinating state across
2+
// cachew replicas.
53
package metadatadb
64

75
import (
@@ -18,16 +16,17 @@ import (
1816
)
1917

2018
// Backend is the pluggable storage layer for the metadata store. Implementations
21-
// handle write operations (Apply), read queries (Query), and persistence (Flush).
19+
// handle write operations (Apply), read queries (Query), and remote state
20+
// refreshes (Flush).
2221
type Backend interface {
2322
// Apply applies one or more write operations to the given namespace.
2423
Apply(ctx context.Context, namespace string, ops ...Op) error
2524
// Query executes a read query and unmarshals the result into target.
2625
// Target must be a pointer to the expected result type.
2726
Query(ctx context.Context, namespace string, q ReadOp, target any) error
28-
// Flush forces pending state to be persisted for the given namespace.
27+
// Flush forces the namespace to refresh from the backend.
2928
Flush(ctx context.Context, namespace string) error
30-
// Close shuts down the backend, flushing any pending state.
29+
// Close shuts down the backend.
3130
Close(ctx context.Context) error
3231
}
3332

@@ -75,13 +74,13 @@ type Namespace struct {
7574
name string
7675
}
7776

78-
// Flush forces an immediate sync with the backend.
77+
// Flush forces an immediate refresh from the backend.
7978
func (n *Namespace) Flush(ctx context.Context) error {
8079
return errors.Wrap(n.backend.Flush(ctx, n.name), "flush namespace")
8180
}
8281

83-
func (n *Namespace) apply(ops ...Op) {
84-
_ = n.backend.Apply(context.Background(), n.name, ops...) //nolint:errcheck // local backends never fail
82+
func (n *Namespace) apply(ops ...Op) error {
83+
return errors.Wrap(n.backend.Apply(context.Background(), n.name, ops...), "apply metadata op")
8584
}
8685

8786
func (n *Namespace) query(q ReadOp, target any) {
@@ -99,8 +98,12 @@ func NewScalar[V any](ns *Namespace, name string) *Scalar[V] {
9998
return &Scalar[V]{ns: ns, name: name}
10099
}
101100

102-
func (s *Scalar[V]) Set(value V) { s.ns.apply(ScalarSet{Key: s.name, Value: value}) }
103-
func (s *Scalar[V]) Delete() { s.ns.apply(ScalarDelete{Key: s.name}) }
101+
func (s *Scalar[V]) Set(value V) error {
102+
return s.ns.apply(ScalarSet{Key: s.name, Value: value})
103+
}
104+
func (s *Scalar[V]) Delete() error {
105+
return s.ns.apply(ScalarDelete{Key: s.name})
106+
}
104107

105108
func (s *Scalar[V]) Get() (V, bool) {
106109
var result struct {
@@ -122,10 +125,18 @@ func NewInt(ns *Namespace, name string) *Int {
122125
return &Int{ns: ns, name: name}
123126
}
124127

125-
func (i *Int) Set(value int64) { i.ns.apply(IntSet{Key: i.name, Value: value}) }
126-
func (i *Int) Add(delta int64) { i.ns.apply(IntAdd{Key: i.name, Delta: delta}) }
127-
func (i *Int) Mul(factor int64) { i.ns.apply(IntMul{Key: i.name, Factor: factor}) }
128-
func (i *Int) Div(divisor int64) { i.ns.apply(IntDiv{Key: i.name, Divisor: divisor}) }
128+
func (i *Int) Set(value int64) error {
129+
return i.ns.apply(IntSet{Key: i.name, Value: value})
130+
}
131+
func (i *Int) Add(delta int64) error {
132+
return i.ns.apply(IntAdd{Key: i.name, Delta: delta})
133+
}
134+
func (i *Int) Mul(factor int64) error {
135+
return i.ns.apply(IntMul{Key: i.name, Factor: factor})
136+
}
137+
func (i *Int) Div(divisor int64) error {
138+
return i.ns.apply(IntDiv{Key: i.name, Divisor: divisor})
139+
}
129140

130141
func (i *Int) Get() int64 {
131142
var v int64
@@ -144,20 +155,20 @@ func NewIntMap[K comparable](ns *Namespace, name string) *IntMap[K] {
144155
return &IntMap[K]{ns: ns, name: name}
145156
}
146157

147-
func (m *IntMap[K]) Set(key K, value int64) {
148-
m.ns.apply(IntMapSet{Key: m.name, MapKey: key, Value: value})
158+
func (m *IntMap[K]) Set(key K, value int64) error {
159+
return m.ns.apply(IntMapSet{Key: m.name, MapKey: key, Value: value})
149160
}
150-
func (m *IntMap[K]) Add(key K, delta int64) {
151-
m.ns.apply(IntMapAdd{Key: m.name, MapKey: key, Delta: delta})
161+
func (m *IntMap[K]) Add(key K, delta int64) error {
162+
return m.ns.apply(IntMapAdd{Key: m.name, MapKey: key, Delta: delta})
152163
}
153-
func (m *IntMap[K]) Mul(key K, factor int64) {
154-
m.ns.apply(IntMapMul{Key: m.name, MapKey: key, Factor: factor})
164+
func (m *IntMap[K]) Mul(key K, factor int64) error {
165+
return m.ns.apply(IntMapMul{Key: m.name, MapKey: key, Factor: factor})
155166
}
156-
func (m *IntMap[K]) Div(key K, divisor int64) {
157-
m.ns.apply(IntMapDiv{Key: m.name, MapKey: key, Divisor: divisor})
167+
func (m *IntMap[K]) Div(key K, divisor int64) error {
168+
return m.ns.apply(IntMapDiv{Key: m.name, MapKey: key, Divisor: divisor})
158169
}
159-
func (m *IntMap[K]) Delete(key K) {
160-
m.ns.apply(IntMapDelete{Key: m.name, MapKey: key})
170+
func (m *IntMap[K]) Delete(key K) error {
171+
return m.ns.apply(IntMapDelete{Key: m.name, MapKey: key})
161172
}
162173

163174
func (m *IntMap[K]) Get(key K) int64 {
@@ -190,8 +201,12 @@ func NewSet[V comparable](ns *Namespace, name string) *Set[V] {
190201
return &Set[V]{ns: ns, name: name}
191202
}
192203

193-
func (s *Set[V]) Add(member V) { s.ns.apply(SetAdd{Key: s.name, Member: member}) }
194-
func (s *Set[V]) Remove(member V) { s.ns.apply(SetRemove{Key: s.name, Member: member}) }
204+
func (s *Set[V]) Add(member V) error {
205+
return s.ns.apply(SetAdd{Key: s.name, Member: member})
206+
}
207+
func (s *Set[V]) Remove(member V) error {
208+
return s.ns.apply(SetRemove{Key: s.name, Member: member})
209+
}
195210

196211
func (s *Set[V]) Contains(member V) bool {
197212
var v bool
@@ -217,12 +232,12 @@ func NewMap[K comparable, V any](ns *Namespace, name string) *Map[K, V] {
217232
return &Map[K, V]{ns: ns, name: name}
218233
}
219234

220-
func (m *Map[K, V]) Set(key K, value V) {
221-
m.ns.apply(MapSet{Key: m.name, MapKey: key, Value: value})
235+
func (m *Map[K, V]) Set(key K, value V) error {
236+
return m.ns.apply(MapSet{Key: m.name, MapKey: key, Value: value})
222237
}
223238

224-
func (m *Map[K, V]) Delete(key K) {
225-
m.ns.apply(MapDelete{Key: m.name, MapKey: key})
239+
func (m *Map[K, V]) Delete(key K) error {
240+
return m.ns.apply(MapDelete{Key: m.name, MapKey: key})
226241
}
227242

228243
func (m *Map[K, V]) Get(key K) (V, bool) {
@@ -258,8 +273,8 @@ func NewList[V any](ns *Namespace, name string) *List[V] {
258273
return &List[V]{ns: ns, name: name}
259274
}
260275

261-
func (l *List[V]) Append(value V) {
262-
l.ns.apply(ListAppend{Key: l.name, Value: value})
276+
func (l *List[V]) Append(value V) error {
277+
return l.ns.apply(ListAppend{Key: l.name, Value: value})
263278
}
264279

265280
func (l *List[V]) Entries() []V {

‎internal/metadatadb/metadatadbtest/soak.go‎

Lines changed: 24 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -150,33 +150,36 @@ func soakWorker(
150150
switch {
151151
// Chaotic ops (not tracked for invariants).
152152
case op < 10:
153-
metadatadb.NewScalar[string](ns, "sc-"+key).Set(fmt.Sprintf("val-%d", rng.IntN(1000)))
153+
recordSoakWrite(result, metadatadb.NewScalar[string](ns, "sc-"+key).Set(fmt.Sprintf("val-%d", rng.IntN(1000))))
154154
case op < 15:
155-
metadatadb.NewScalar[string](ns, "sc-"+key).Delete()
155+
recordSoakWrite(result, metadatadb.NewScalar[string](ns, "sc-"+key).Delete())
156156
case op < 20:
157-
metadatadb.NewInt(ns, "int-"+key).Set(int64(rng.IntN(1000)))
157+
recordSoakWrite(result, metadatadb.NewInt(ns, "int-"+key).Set(int64(rng.IntN(1000))))
158158
case op < 25:
159-
metadatadb.NewSet[string](ns, "set-"+key).Add(fmt.Sprintf("m-%d", rng.IntN(20)))
159+
recordSoakWrite(result, metadatadb.NewSet[string](ns, "set-"+key).Add(fmt.Sprintf("m-%d", rng.IntN(20))))
160160
case op < 28:
161-
metadatadb.NewSet[string](ns, "set-"+key).Remove(fmt.Sprintf("m-%d", rng.IntN(20)))
161+
recordSoakWrite(result, metadatadb.NewSet[string](ns, "set-"+key).Remove(fmt.Sprintf("m-%d", rng.IntN(20))))
162162
case op < 35:
163-
metadatadb.NewMap[string, string](ns, "map-"+key).Set(
163+
recordSoakWrite(result, metadatadb.NewMap[string, string](ns, "map-"+key).Set(
164164
fmt.Sprintf("k-%d", rng.IntN(20)),
165165
fmt.Sprintf("v-%d", rng.IntN(1000)),
166-
)
166+
))
167167

168168
// Monotonic ops (tracked for invariant verification).
169169
case op < 50:
170170
delta := int64(rng.IntN(100) + 1) // always positive
171-
metadatadb.NewInt(ns, "mono-int-"+key).Add(delta)
172-
tr.addCounter("mono-int-"+key, delta)
171+
if recordSoakWrite(result, metadatadb.NewInt(ns, "mono-int-"+key).Add(delta)) {
172+
tr.addCounter("mono-int-"+key, delta)
173+
}
173174
case op < 65:
174175
member := fmt.Sprintf("m-%d", rng.IntN(50))
175-
metadatadb.NewSet[string](ns, "mono-set-"+key).Add(member)
176-
tr.addSetMember("mono-set-"+key, member)
176+
if recordSoakWrite(result, metadatadb.NewSet[string](ns, "mono-set-"+key).Add(member)) {
177+
tr.addSetMember("mono-set-"+key, member)
178+
}
177179
case op < 75:
178-
metadatadb.NewList[string](ns, "mono-list-"+key).Append(fmt.Sprintf("e-%d", rng.IntN(1000)))
179-
tr.appendList("mono-list-" + key)
180+
if recordSoakWrite(result, metadatadb.NewList[string](ns, "mono-list-"+key).Append(fmt.Sprintf("e-%d", rng.IntN(1000)))) {
181+
tr.appendList("mono-list-" + key)
182+
}
180183

181184
// Reads.
182185
case op < 80:
@@ -200,6 +203,14 @@ func soakWorker(
200203
}
201204
}
202205

206+
func recordSoakWrite(result *SoakResult, err error) bool {
207+
if err == nil {
208+
return true
209+
}
210+
atomic.AddInt64(&result.Errors, 1)
211+
return false
212+
}
213+
203214
func verifyMonotonicInvariants(t *testing.T, ns *metadatadb.Namespace, tr *tracker) {
204215
t.Helper()
205216
tr.mu.Lock()

0 commit comments

Comments
 (0)