Documentation
¶
Overview ¶
Package memory provides the OniWorks distributed in-memory database. It is the state layer of the framework: sessions, presence, rate limits, pub/sub events, and realtime state. It is NOT a replacement for PostgreSQL.
Index ¶
- type ClockValue
- type Options
- type PubSub
- type Store
- func (s *Store) CompareAndSwap(key string, expected, newValue any, ttl time.Duration) bool
- func (s *Store) Count(pattern string) int
- func (s *Store) Decr(key string) int64
- func (s *Store) Delete(key string)
- func (s *Store) Expire(key string, ttl time.Duration) bool
- func (s *Store) Flush()
- func (s *Store) Get(key string) (any, bool)
- func (s *Store) GetInt64(key string) (int64, bool)
- func (s *Store) GetString(key string) (string, bool)
- func (s *Store) Has(key string) bool
- func (s *Store) Incr(key string) int64
- func (s *Store) IncrBy(key string, n int64) int64
- func (s *Store) Keys(pattern string) []string
- func (s *Store) Publish(topic string, payload any)
- func (s *Store) Set(key string, value any, ttl time.Duration)
- func (s *Store) Shutdown() error
- func (s *Store) Subscribe(topic string, handler func(topic string, payload any)) func()
- type VectorClock
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ClockValue ¶
ClockValue is an immutable snapshot of a vector clock at a point in time.
Fields are EXPORTED so the value survives gob encoding across the gossip transport — with unexported fields the vector was silently dropped on the wire, collapsing last-write-wins to "always overwrite".
func (ClockValue) After ¶
func (cv ClockValue) After(other ClockValue) bool
After reports whether cv should win a last-write-wins conflict against other.
It first does a proper vector-dominance comparison across the union of all node IDs: if cv causally dominates other (cv[n] >= other[n] for every n and strictly greater for at least one), cv is newer. If neither dominates the writes are concurrent, and we break the tie deterministically (by TS, then NodeID) so every node converges on the same winner.
func (ClockValue) Equal ¶
func (cv ClockValue) Equal(other ClockValue) bool
Equal reports whether two clock values are concurrent (neither is after the other).
type Options ¶
type Options struct {
// NodeID uniquely identifies this node. Auto-generated if empty.
NodeID string
// BindAddr is the TCP address this node listens on for gossip (e.g. "0.0.0.0:7946").
// Leave empty to run in single-node mode (no gossip, no cross-node sync).
BindAddr string
// Peers is the list of known peer addresses for gossip bootstrapping.
// Example: []string{"10.0.0.2:7946", "10.0.0.3:7946"}
Peers []string
// GossipSecret is a pre-shared secret used to authenticate peer connections.
// Every node in a cluster must share the same value. If empty, gossip runs
// UNAUTHENTICATED (any host that can reach BindAddr can read and inject data)
// and a loud warning is logged — set this for any non-trusted network.
GossipSecret string
// Persist enables snapshot-to-disk on shutdown.
Persist bool
// SnapshotPath is the file path for the snapshot (default: "storage/memory.snap").
SnapshotPath string
// GracefulSave saves snapshot on SIGTERM/SIGINT.
GracefulSave bool
// RedisURL enables the Redis sync adapter instead of the built-in gossip.
// When set, gossip is disabled and Redis pub/sub is used for cross-node sync.
// Example: "redis://localhost:6379"
RedisURL string
// MaxKeys caps the number of stored keys (0 = unlimited).
MaxKeys int
// EvictInterval is how often the TTL eviction loop runs (default: 30s).
EvictInterval time.Duration
}
Options configures a Store instance.
type PubSub ¶
type PubSub struct {
// contains filtered or unexported fields
}
PubSub is a thread-safe, fan-out publish/subscribe engine. Topics support wildcard matching: "user.*" matches "user.login", "user.42.status".
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the OniWorks distributed in-memory database. It provides KV storage, pub/sub, TTL eviction, snapshot persistence, and cross-node sync via TCP gossip (or Redis as optional adapter).
func (*Store) CompareAndSwap ¶
CompareAndSwap atomically updates key to newValue only if it currently equals expected. Returns true if the swap occurred.
func (*Store) Delete ¶
Delete removes a key. The delete is stamped with a fresh clock and recorded as a tombstone so a reordered (causally older) remote set cannot resurrect the key here or on peers.
func (*Store) Flush ¶
func (s *Store) Flush()
Flush removes all keys. Use only in development/tests.
func (*Store) GetString ¶
GetString retrieves a string value. Returns ("", false) if absent or wrong type.
func (*Store) Incr ¶
Incr atomically increments a counter key by 1. Creates with value 1 if absent. Returns the new value.
func (*Store) Keys ¶
Keys returns all non-expired keys matching the given glob prefix pattern. Use "*" to match all keys. Use "session:*" to match all session keys.
func (*Store) Publish ¶
Publish broadcasts payload to all subscribers of topic. Cross-node: the gossip/Redis transport propagates this to peer nodes.
func (*Store) Set ¶
Set stores key with value and an optional TTL (0 = no expiry).
If Options.MaxKeys is set, a Set that would create a NEW key while the store is at capacity is rejected (updates to existing keys always succeed). The rejection is logged; callers that need a guaranteed write should provision a larger cap. A zero MaxKeys means unlimited.
func (*Store) Shutdown ¶
Shutdown stops the store, saves snapshot if configured, and closes gossip connections.
type VectorClock ¶
type VectorClock struct {
// contains filtered or unexported fields
}
VectorClock implements a simple Lamport-style vector clock for last-write-wins conflict resolution across nodes. Each node maintains its own counter and increments it on every write.
func NewVectorClock ¶
func NewVectorClock(nodeID string) *VectorClock
NewVectorClock creates a VectorClock for the given node ID.
func (*VectorClock) Merge ¶
func (vc *VectorClock) Merge(received ClockValue)
Merge updates this clock with a received clock value. Uses max(local[node], received[node]) for each node.
func (*VectorClock) Tick ¶
func (vc *VectorClock) Tick() ClockValue
Tick increments this node's local time and returns a new ClockValue. Call before every local write.
The counter increment happens UNDER the lock: if it were fetched outside, two concurrent Ticks could apply their map writes in the opposite order of their counter values, regressing clocks[nodeID] and snapshotting a vector older than an already-issued ClockValue.
Source Files