Skip to content

fix: sync s3 metadata writes #382

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alecthomas merged 1 commit into from
Jul 3, 2026
Merged

fix: sync s3 metadata writes #382

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions internal/cache/tiered.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down Expand Up @@ -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
}
Expand Down
10 changes: 5 additions & 5 deletions internal/cache/tiered_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand Down
83 changes: 49 additions & 34 deletions internal/metadatadb/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 (
Expand All @@ -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
}

Expand Down Expand Up @@ -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")
}
Comment on lines +82 to 84

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Propagate cancellation into synchronous metadata writes

Now that S3 Apply performs the lock/load/store path synchronously, using context.Background() for every public metadata write discards the caller's request deadline/cancellation. Under an S3 outage or a stuck namespace lock, callers such as tiered cache create/delete can remain blocked in lockNamespace or MinIO calls even after the HTTP request context has been canceled, because this write path has no cancellable context to observe.

Useful? React with 👍 / 👎.


func (n *Namespace) query(q ReadOp, target any) {
Expand All @@ -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 {
Expand All @@ -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
Expand All @@ -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 {
Expand Down Expand Up @@ -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
Expand All @@ -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) {
Expand Down Expand Up @@ -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 {
Expand Down
37 changes: 24 additions & 13 deletions internal/metadatadb/metadatadbtest/soak.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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()
Expand Down
Loading