From 1e84ed7d088c73836abc020ede79ae199f1bf2af Mon Sep 17 00:00:00 2001 From: Code0987 <1825861+Code0987@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:46:06 +0000 Subject: [PATCH] refactor: extract ModeCounter into its own files Move store Counter off memory.go and split engine Counter into public API, owner write, and GetOrLoad. Peer CounterIncr unchanged. --- docs/design/2026-09-02-refactor-counter.md | 38 ++++++ docs/design/README.md | 1 + pkg/engine/counter.go | 63 +-------- pkg/engine/counter_apply.go | 32 +++++ pkg/engine/counter_cluster.go | 44 ++++++ pkg/store/counter.go | 147 +++++++++++++++++++++ pkg/store/memory.go | 135 ------------------- 7 files changed, 264 insertions(+), 196 deletions(-) create mode 100644 docs/design/2026-09-02-refactor-counter.md create mode 100644 pkg/engine/counter_apply.go create mode 100644 pkg/engine/counter_cluster.go create mode 100644 pkg/store/counter.go diff --git a/docs/design/2026-09-02-refactor-counter.md b/docs/design/2026-09-02-refactor-counter.md new file mode 100644 index 0000000..752e60e --- /dev/null +++ b/docs/design/2026-09-02-refactor-counter.md @@ -0,0 +1,38 @@ +# Refactor ModeCounter layout (no contract change) + +**Status:** approved (chat: continue) +**Branch:** `feat/refactor-counter` +**Date:** 2026-09-02 + +## Problem + +Counter store methods still live in `pkg/store/memory.go`. Engine Counter is one file mixing public verbs, owner write, and GetOrLoad. + +## Non-goals + +- No API / proto / Peer `CounterIncr` / version / fan-out change +- No `pkg/counter` algorithm change +- Not extracting other modes (HLL #40, Bitmap #41) + +## Contract + +Unchanged: `Incr` returns the new int64 (peer `CounterIncr` from non-owner). `CounterGet` present-bit. `FlagCounter` snapshot. Version under store mutex. Overflow → invalid argument, no wrap. + +## Approach + +Same file layout as HLL / Bitmap: + +| Before | After | +|--------|--------| +| `pkg/store/memory.go` C* | `pkg/store/counter.go` | +| `pkg/engine/counter.go` (all) | `counter.go` public verbs; `counter_apply.go` owner write + install; `counter_cluster.go` GetOrLoad / snapshot | + +Rejected: drop the Peer RPC (that is the return-*n* contract); split `pkg/counter` (39 lines). + +## Tests (already exist; keep them) + +Existing `pkg/counter`, `pkg/store` counter, `pkg/engine` counter unit + cluster. + +## Bench risk + +No new Get/Peek logic. Local Get-hit / StoreGetHit allocs/op must stay 15 / 2. diff --git a/docs/design/README.md b/docs/design/README.md index 7275235..487b4e8 100644 --- a/docs/design/README.md +++ b/docs/design/README.md @@ -22,6 +22,7 @@ Do not implement from a draft. | [2026-08-25-mode-hll.md](./2026-08-25-mode-hll.md) | `ModeHLL` | | [2026-08-31-mode-topk.md](./2026-08-31-mode-topk.md) | `ModeTopK` | | [2026-09-01-mode-cms.md](./2026-09-01-mode-cms.md) | `ModeCMS` | +| [2026-09-02-refactor-counter.md](./2026-09-02-refactor-counter.md) | ModeCounter file layout (no contract) | | [2026-08-25-list-counter-version.md](./2026-08-25-list-counter-version.md) | List/Counter snapshot version | | [2026-08-13-unify-grpc-error-map.md](./2026-08-13-unify-grpc-error-map.md) | grpcmap | diff --git a/pkg/engine/counter.go b/pkg/engine/counter.go index 3e675a3..f24bb04 100644 --- a/pkg/engine/counter.go +++ b/pkg/engine/counter.go @@ -6,10 +6,10 @@ import ( "github.com/Code0987/supercache/pkg/counter" "github.com/Code0987/supercache/pkg/keyspace" - "github.com/Code0987/supercache/pkg/store" ) // Incr adds delta to a ModeCounter and returns the new value. +// Non-owners use peer CounterIncr (the return payload needs a Peer RPC). func (e *Engine) Incr(ctx context.Context, keyspaceName, name string, delta int64) (int64, error) { if err := ctx.Err(); err != nil { return 0, err @@ -59,7 +59,7 @@ func (e *Engine) CounterGet(ctx context.Context, keyspaceName, name string) (int if err := e.validateKeyLen(ks, name); err != nil { return 0, false, err } - if e.hasCounterLocal(ks, name) { + if ks.store.HasCounter(name) { v, ok := ks.store.CGet(name) return v, ok, nil } @@ -73,62 +73,3 @@ func (e *Engine) CounterGet(ctx context.Context, keyspaceName, name string) (int } return v, true, nil } - -func (e *Engine) cIncrLocal(ks *ksRuntime, name string, delta int64, fanout bool) (int64, error) { - expire := e.expireAt(ks.cfg.TTL) - cur, _ := ks.store.PeekVersion(name) - gate := cur + 1 - n, applied, overflow := ks.store.CIncr(name, delta, gate, expire) - if overflow { - return 0, fmt.Errorf("%w: counter overflow", ErrInvalidArgument) - } - if !applied { - return 0, fmt.Errorf("%w: incr rejected", ErrInvalidArgument) - } - ver, _ := ks.store.PeekVersion(name) - ks.observeVersion(name, ver) - if fanout { - e.replicate(ks.cfg.Name, name, store.Entry{ - Value: counter.Encode(n), - Version: ver, - ExpireAt: expire, - Flags: store.FlagCounter, - }, false) - } - return n, nil -} - -func (e *Engine) hasCounterLocal(ks *ksRuntime, name string) bool { - return ks.store.HasCounter(name) -} - -func (e *Engine) cFetchOwner(ctx context.Context, ks *ksRuntime, name string) (store.Entry, bool, error) { - c := e.clusterSnapshot() - if c == nil || c.Ring == nil || c.Transport == nil { - return store.Entry{}, false, nil - } - owner, ok := c.Ring.Owner(name) - if !ok || owner.ID == "" || owner.ID == c.SelfID || owner.Addr == "" { - return store.Entry{}, false, nil - } - pctx, cancel := e.peerCtx(ctx, ks) - defer cancel() - res, err := c.Transport.GetOrLoad(pctx, owner.Addr, ks.cfg.Name, name) - if err != nil || !res.Found || !res.Entry.IsCounter() { - return store.Entry{}, false, nil - } - if _, decErr := counter.Decode(res.Entry.Value); decErr != nil { - return store.Entry{}, false, nil - } - if e.holdsReplica(c, ks, name) { - _ = ks.store.CInstall(name, res.Entry.Value, res.Entry.Version, res.Entry.ExpireAt) - } - return res.Entry, true, nil -} - -func (e *Engine) applyCounterInstall(ks *ksRuntime, name string, blob []byte, version uint64, expireAt int64) bool { - if expireAt == 0 { - expireAt = e.expireAt(ks.cfg.TTL) - } - return ks.store.CInstall(name, blob, version, expireAt) -} diff --git a/pkg/engine/counter_apply.go b/pkg/engine/counter_apply.go new file mode 100644 index 0000000..c892bc0 --- /dev/null +++ b/pkg/engine/counter_apply.go @@ -0,0 +1,32 @@ +package engine + +import "fmt" + +// cIncrLocal is the owner / single-node write path. +// The store assigns the stored version; we fan PeekVersion after the write. +func (e *Engine) cIncrLocal(ks *ksRuntime, name string, delta int64, fanout bool) (int64, error) { + expire := e.expireAt(ks.cfg.TTL) + cur, _ := ks.store.PeekVersion(name) + gate := cur + 1 + n, applied, overflow := ks.store.CIncr(name, delta, gate, expire) + if overflow { + return 0, fmt.Errorf("%w: counter overflow", ErrInvalidArgument) + } + if !applied { + return 0, fmt.Errorf("%w: incr rejected", ErrInvalidArgument) + } + ver, _ := ks.store.PeekVersion(name) + ks.observeVersion(name, ver) + if fanout { + e.cReplicateSnapshot(ks, name, n, ver, expire) + } + return n, nil +} + +// applyCounterInstall is replica / handoff ApplyPut of FlagCounter (LWW replace). +func (e *Engine) applyCounterInstall(ks *ksRuntime, name string, blob []byte, version uint64, expireAt int64) bool { + if expireAt == 0 { + expireAt = e.expireAt(ks.cfg.TTL) + } + return ks.store.CInstall(name, blob, version, expireAt) +} diff --git a/pkg/engine/counter_cluster.go b/pkg/engine/counter_cluster.go new file mode 100644 index 0000000..2748712 --- /dev/null +++ b/pkg/engine/counter_cluster.go @@ -0,0 +1,44 @@ +package engine + +import ( + "context" + + "github.com/Code0987/supercache/pkg/counter" + "github.com/Code0987/supercache/pkg/store" +) + +// cReplicateSnapshot fans the post-write FlagCounter blob to RF−1 replicas. +func (e *Engine) cReplicateSnapshot(ks *ksRuntime, name string, n int64, ver uint64, expire int64) { + e.replicate(ks.cfg.Name, name, store.Entry{ + Value: counter.Encode(n), + Version: ver, + ExpireAt: expire, + Flags: store.FlagCounter, + }, false) +} + +// cFetchOwner loads a missing local name from the owner (GetOrLoad). +// RPC / !Found / wrong type / bad blob → miss + nil error (do not return Unavailable). +func (e *Engine) cFetchOwner(ctx context.Context, ks *ksRuntime, name string) (store.Entry, bool, error) { + c := e.clusterSnapshot() + if c == nil || c.Ring == nil || c.Transport == nil { + return store.Entry{}, false, nil + } + owner, ok := c.Ring.Owner(name) + if !ok || owner.ID == "" || owner.ID == c.SelfID || owner.Addr == "" { + return store.Entry{}, false, nil + } + pctx, cancel := e.peerCtx(ctx, ks) + defer cancel() + res, err := c.Transport.GetOrLoad(pctx, owner.Addr, ks.cfg.Name, name) + if err != nil || !res.Found || !res.Entry.IsCounter() { + return store.Entry{}, false, nil + } + if _, decErr := counter.Decode(res.Entry.Value); decErr != nil { + return store.Entry{}, false, nil + } + if e.holdsReplica(c, ks, name) { + _ = ks.store.CInstall(name, res.Entry.Value, res.Entry.Version, res.Entry.ExpireAt) + } + return res.Entry, true, nil +} diff --git a/pkg/store/counter.go b/pkg/store/counter.go new file mode 100644 index 0000000..9421892 --- /dev/null +++ b/pkg/store/counter.go @@ -0,0 +1,147 @@ +package store + +import ( + "github.com/Code0987/supercache/pkg/counter" +) + +// CIncr adds delta to the named counter (creates if missing). +// +// version is only the tombstone-gate floor. The stored version is 1 on create +// or local+1 on a live/tombstone replace — never the inbound number. +// Returns (value, applied, overflow). +func (m *Memory) CIncr(key string, delta int64, version uint64, expireAt int64) (int64, bool, bool) { + m.mu.Lock() + defer m.mu.Unlock() + + if el, ok := m.items[key]; ok { + it := el.Value.(*lruItem) + if !it.entry.Expired(m.now()) { + if it.entry.IsTombstone() { + if version <= it.entry.Version { + m.staleSkip.Add(1) + return 0, false, false + } + stored := it.entry.Version + 1 + m.removeElement(el) + return m.cInsertLocked(key, delta, stored, expireAt) + } + if !it.entry.IsCounter() { + return 0, false, false + } + cur, err := counter.Decode(it.entry.Value) + if err != nil { + return 0, false, false + } + next, err := counter.Add(cur, delta) + if err != nil { + return cur, false, true + } + it.entry.Version = it.entry.Version + 1 + it.entry.Flags = FlagCounter + it.entry.Value = counter.Encode(next) + if expireAt != 0 { + it.entry.ExpireAt = expireAt + } + oldCost := it.cost + it.cost = entryCost(key, it.entry) + m.bytes += it.cost - oldCost + m.order.MoveToFront(el) + m.evictLocked() + return next, true, false + } + m.removeElement(el) + } + return m.cInsertLocked(key, delta, 1, expireAt) +} + +func (m *Memory) cInsertLocked(key string, val int64, version uint64, expireAt int64) (int64, bool, bool) { + ent := Entry{Value: counter.Encode(val), Version: version, ExpireAt: expireAt, Flags: FlagCounter} + if !m.insertCounterLocked(key, ent) { + return 0, false, false + } + return val, true, false +} + +// CGet returns the counter value. Missing → 0, ok=false. +func (m *Memory) CGet(key string) (int64, bool) { + m.mu.Lock() + defer m.mu.Unlock() + el, ok := m.items[key] + if !ok { + return 0, false + } + it := el.Value.(*lruItem) + if it.entry.Expired(m.now()) { + m.removeElement(el) + return 0, false + } + if it.entry.IsTombstone() || !it.entry.IsCounter() { + return 0, false + } + v, err := counter.Decode(it.entry.Value) + if err != nil { + return 0, false + } + return v, true +} + +// HasCounter is the present-bit: live, unexpired, FlagCounter, decodable. +func (m *Memory) HasCounter(key string) bool { + m.mu.Lock() + defer m.mu.Unlock() + el, ok := m.items[key] + if !ok { + return false + } + it := el.Value.(*lruItem) + if it.entry.Expired(m.now()) { + m.removeElement(el) + return false + } + if it.entry.IsTombstone() || !it.entry.IsCounter() { + return false + } + _, err := counter.Decode(it.entry.Value) + return err == nil +} + +// CInstall is LWW snapshot handoff: keep blob if version > local. +func (m *Memory) CInstall(key string, blob []byte, version uint64, expireAt int64) bool { + m.mu.Lock() + defer m.mu.Unlock() + if _, err := counter.Decode(blob); err != nil { + return false + } + if el, ok := m.items[key]; ok { + it := el.Value.(*lruItem) + if !it.entry.Expired(m.now()) { + if it.entry.IsTombstone() { + if version <= it.entry.Version { + m.staleSkip.Add(1) + return false + } + } else if it.entry.IsCounter() { + if version <= it.entry.Version { + m.staleSkip.Add(1) + return false + } + } else if version <= it.entry.Version { + return false + } + } + m.removeElement(el) + } + ent := Entry{Value: append([]byte(nil), blob...), Version: version, ExpireAt: expireAt, Flags: FlagCounter} + return m.insertCounterLocked(key, ent) +} + +func (m *Memory) insertCounterLocked(key string, ent Entry) bool { + cost := entryCost(key, ent) + it := &lruItem{key: key, entry: copyEntry(ent), cost: cost} + el := m.order.PushFront(it) + m.items[key] = el + m.bytes += cost + m.evictLocked() + _, ok := m.items[key] + return ok +} diff --git a/pkg/store/memory.go b/pkg/store/memory.go index 4647487..a94e263 100644 --- a/pkg/store/memory.go +++ b/pkg/store/memory.go @@ -8,7 +8,6 @@ import ( "github.com/Code0987/supercache/pkg/bitmapx" "github.com/Code0987/supercache/pkg/bloom" - "github.com/Code0987/supercache/pkg/counter" "github.com/Code0987/supercache/pkg/geo" "github.com/Code0987/supercache/pkg/hashx" "github.com/Code0987/supercache/pkg/hllx" @@ -1676,140 +1675,6 @@ func (m *Memory) insertHashLocked(key string, ent Entry, cache *hashx.Hash, dirt return ok } -func (m *Memory) CIncr(key string, delta int64, version uint64, expireAt int64) (int64, bool, bool) { - m.mu.Lock() - defer m.mu.Unlock() - - if el, ok := m.items[key]; ok { - it := el.Value.(*lruItem) - if !it.entry.Expired(m.now()) { - if it.entry.IsTombstone() { - if version <= it.entry.Version { - m.staleSkip.Add(1) - return 0, false, false - } - stored := it.entry.Version + 1 - m.removeElement(el) - return m.cInsertLocked(key, delta, stored, expireAt) - } - if !it.entry.IsCounter() { - return 0, false, false - } - cur, err := counter.Decode(it.entry.Value) - if err != nil { - return 0, false, false - } - next, err := counter.Add(cur, delta) - if err != nil { - return cur, false, true - } - it.entry.Version = it.entry.Version + 1 - it.entry.Flags = FlagCounter - it.entry.Value = counter.Encode(next) - if expireAt != 0 { - it.entry.ExpireAt = expireAt - } - oldCost := it.cost - it.cost = entryCost(key, it.entry) - m.bytes += it.cost - oldCost - m.order.MoveToFront(el) - m.evictLocked() - return next, true, false - } - m.removeElement(el) - } - return m.cInsertLocked(key, delta, 1, expireAt) -} - -func (m *Memory) cInsertLocked(key string, val int64, version uint64, expireAt int64) (int64, bool, bool) { - ent := Entry{Value: counter.Encode(val), Version: version, ExpireAt: expireAt, Flags: FlagCounter} - if !m.insertCounterLocked(key, ent) { - return 0, false, false - } - return val, true, false -} - -func (m *Memory) CGet(key string) (int64, bool) { - m.mu.Lock() - defer m.mu.Unlock() - el, ok := m.items[key] - if !ok { - return 0, false - } - it := el.Value.(*lruItem) - if it.entry.Expired(m.now()) { - m.removeElement(el) - return 0, false - } - if it.entry.IsTombstone() || !it.entry.IsCounter() { - return 0, false - } - v, err := counter.Decode(it.entry.Value) - if err != nil { - return 0, false - } - return v, true -} - -func (m *Memory) HasCounter(key string) bool { - m.mu.Lock() - defer m.mu.Unlock() - el, ok := m.items[key] - if !ok { - return false - } - it := el.Value.(*lruItem) - if it.entry.Expired(m.now()) { - m.removeElement(el) - return false - } - if it.entry.IsTombstone() || !it.entry.IsCounter() { - return false - } - _, err := counter.Decode(it.entry.Value) - return err == nil -} - -func (m *Memory) CInstall(key string, blob []byte, version uint64, expireAt int64) bool { - m.mu.Lock() - defer m.mu.Unlock() - if _, err := counter.Decode(blob); err != nil { - return false - } - if el, ok := m.items[key]; ok { - it := el.Value.(*lruItem) - if !it.entry.Expired(m.now()) { - if it.entry.IsTombstone() { - if version <= it.entry.Version { - m.staleSkip.Add(1) - return false - } - } else if it.entry.IsCounter() { - if version <= it.entry.Version { - m.staleSkip.Add(1) - return false - } - } else if version <= it.entry.Version { - return false - } - } - m.removeElement(el) - } - ent := Entry{Value: append([]byte(nil), blob...), Version: version, ExpireAt: expireAt, Flags: FlagCounter} - return m.insertCounterLocked(key, ent) -} - -func (m *Memory) insertCounterLocked(key string, ent Entry) bool { - cost := entryCost(key, ent) - it := &lruItem{key: key, entry: copyEntry(ent), cost: cost} - el := m.order.PushFront(it) - m.items[key] = el - m.bytes += cost - m.evictLocked() - _, ok := m.items[key] - return ok -} - func (m *Memory) JSet(key, path string, raw []byte, version uint64, expireAt int64, maxValue int) (applied, tooLarge bool) { m.mu.Lock() defer m.mu.Unlock()