capacitor

package module
v0.26.8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 9, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

README

Capacitor

High-voltage synchronization for local-first apps.

Capacitor Mascot

Capacitor is a distributed, persistent, and sub-millisecond caching layer for the Cuprite Flux engine. It provides high availability and zero-latency local reads/writes by utilizing a local-first architecture synchronized via a background replication log and gossip-based discovery.

✨ Key Features

  • Local-First Performance: All operations (Get, Set, Increment) are performed against a sharded in-memory cache backed by local BadgerDB persistence. The network is never on the hot path.
  • Hybrid Logical Clocks (HLC): Ensures causality-preserving order for distributed updates without requiring perfect clock synchronization across nodes.
  • Eventual Consistency via CRDTs (see the Conflict Resolution Guide):
    • Registers: Uses Last-Write-Wins (LWW) based on HLC timestamps.
    • Counters: Implements state-based PN-Counters for idempotent increments.
    • Sliding Windows: Distributed windowed counters with automated pruning.
  • Asynchronous Replication: A binary circular Delta Log ensures high-throughput, low-latency propagation of updates between peers.
  • Secure by Default: Supports mTLS for replication streams and shared-secret authentication for the gossip layer.
  • Observability: Built-in metrics tracking for end-to-end replication latency and operation performance.

🏗️ Architecture

Capacitor is designed for high-throughput environments where read/write latency is critical. For a complete deep-dive into internal modules, data flows, and subsystem diagrams, see the Architecture Guide:

  1. Write Path: When a write occurs, it is committed to the sharded in-memory cache, appended to an in-memory binary Delta Log, and asynchronously flushed to the local BadgerDB database.
  2. Discovery: Nodes use the SWIM protocol (via HashiCorp Memberlist) to discover peers and maintain cluster membership.
  3. Sync Path: Background replicators stream entries from the Delta Log to peers over TCP/TLS.
  4. Conflict Resolution: Received updates are merged into the local store using HLC-based conflict resolution, ensuring all nodes eventually converge to the same state.

📊 Performance

Operation Latency (ns/op) Latency (ms)
Set ~226 0.00022 ms
Get ~21 0.00002 ms
Get (Scan) ~29 0.00003 ms
Increment ~409 0.00041 ms
Exists ~19 0.00002 ms
Delete ~179 0.00018 ms
SetAdd ~211 0.00021 ms
SetRemove ~545 0.00055 ms
SetIsMember ~220 0.00022 ms
SetMIsMember ~812 0.00081 ms
SetCard ~6536 (1k set size) 0.00654 ms
SetMembers ~293947 (1k set size) 0.29395 ms
SetPop ~11447 (1k set size) 0.01145 ms
SetRandMember ~10294 (1k set size) 0.01029 ms
SetMove ~238 0.00024 ms
SortedSetAdd ~765 0.00077 ms
SortedSetRemove ~480 0.00048 ms
SortedSetScore ~227 0.00023 ms
SortedSetCard ~16 0.00002 ms
SortedSetRank ~280 (1k set size) 0.00028 ms
SortedSetCount ~1000 (1k set size) 0.00100 ms
SortedSetIncrementBy ~780 0.00078 ms
SortedSetRangeScan ~285000 (1k set size) 0.28500 ms
SortedSetRangeByScoreScan ~110000 (1k set size) 0.11000 ms
SortedSetRevRangeScan ~305000 (1k set size) 0.30500 ms
SortedSetRevRangeByScoreScan ~111000 (1k set size) 0.11100 ms
SortedSetPopScan ~930 (1k set size) 0.00093 ms
MapSet ~720 0.00072 ms
MapGetScan ~70 0.00007 ms
MapGetMScan ~142 0.00014 ms
MapGetAllScan ~538 0.00054 ms
MapIncrementBy ~580 0.00058 ms
MapExists ~35 0.00003 ms
MapLen ~165 0.00016 ms
MapKeysScan ~780 0.00078 ms
MapValuesScan ~598 0.00059 ms
NMapSet ~1057 0.00105 ms
NMapGetScan ~448 0.00044 ms
NMapGetMScan ~843 0.00084 ms
NMapGetAllScan ~868 0.00086 ms
NMapIncrementBy ~811 0.00081 ms
NMapExists ~48 0.00004 ms
NMapLen ~193 0.00019 ms
NMapKeysScan ~1005 0.00100 ms
NMapValuesScan ~700 0.00070 ms
ListLeftPush ~2144 0.00214 ms
ListRightPush ~1622492 1.62249 ms
ListLeftPopScan ~1552419 1.55241 ms
ListRightPopScan ~1514050 1.51405 ms
ListRangeScan ~5039 0.00503 ms
ListLen ~369829 0.36983 ms
HLLAdd ~27 0.00003 ms
HLLCount ~260 0.00026 ms
BloomAdd ~32 0.00003 ms
BloomExists ~7 0.000007 ms
CMSIncrement ~111 0.00011 ms
CMSQuery ~7 0.000007 ms
PubSub Propagation ~8002 0.00800 ms

💻 Usage

import "github.com/cuprite-io/capacitor"

cfg := capacitor.Config{
    NodeID:     "node-1",
    BindPort:   7946,           // Gossip port
    StreamPort: 7947,           // Replication port
    Peers:      []string{"10.0.0.1:7946"},
    DataPath:   "/var/lib/capacitor",
    AuthToken:  "your-secure-token",
}

cp, _ := capacitor.New(cfg)
defer cp.Close()

// Standard Cache Operations
cp.Set(ctx, "session:123", "data", 10 * time.Minute)
val, _ := cp.Get(ctx, "session:123")
exists, _ := cp.Exists(ctx, "session:123")
_ = cp.Delete(ctx, "session:123")

// Structured Scanning (similar to json.Unmarshal)
var session UserSession
_ = cp.GetScan(ctx, "session:123", &session)

// Distributed Counters
count, _ := cp.Increment(ctx, "page_views")

// Sliding Windows
windowCount, _ := cp.IncrementSlidingWindow(ctx, "rate_limit:user_1", 1 * time.Minute)

// Replicated Conflict-Free Sets
_, _ = cp.SetAdd(ctx, "user:1:tags", "golang")
_, _ = cp.SetAdd(ctx, "user:1:tags", "crdt")
tags, _ := cp.SetMembers(ctx, "user:1:tags")
isMem, _ := cp.SetIsMember(ctx, "user:1:tags", "golang")
card, _ := cp.SetCard(ctx, "user:1:tags")
_, _ = cp.SetRemove(ctx, "user:1:tags", "golang")

// Storing and scanning non-string types in Sets (e.g. integers, custom structs)
type UserProfile struct {
    Age  int    `msgpack:"age"`
    Name string `msgpack:"name"`
}
_, _ = cp.SetAdd(ctx, "users", UserProfile{Age: 30, Name: "Alice"})
_, _ = cp.SetAdd(ctx, "users", UserProfile{Age: 25, Name: "Bob"})

var users []UserProfile
_ = cp.SetMembersScan(ctx, "users", &users)

var poppedUser UserProfile
_, _ = cp.SetPopScan(ctx, "users", &poppedUser)

🔒 Security

For production deployments, always configure a non-empty AuthToken. Setting an AuthToken automatically encrypts the gossip memberlist layer (via AES-256 using Memberlist's SecretKey derived via SHA-256), securing stream addresses and node metadata against plaintext eavesdropping. If AuthToken is left empty, metadata and replication ports are broadcast in plaintext.

🧪 Testing

Run the comprehensive test suite, including chaos and convergence tests:

go test -v .
go test -bench=. .

Documentation

Index

Constants

View Source 
const DefaultMaxOffset = 500 * time.Millisecond
View Source 
const NoExpiry = time.Duration(-1)
View Source 
const Version = "0.26.8"

Version is the current version of the capacitor library.

Variables

View Source 
var ErrClockSmash = errors.New("HLC clock smash detected")
View Source 
var ErrKeyNotFound = errors.New("capacitor: key not found")

Functions

This section is empty.

Types

type Batch 

type Batch struct {
	FromNode  string     `json:"f" msg:"f"`
	Entries   [][]byte   `json:"e" msg:"e"`
	Handshake *Handshake `json:"h,omitempty" msg:"h,omitempty"`
}

func (*Batch) DecodeMsg 

func (z *Batch) DecodeMsg(dc *msgp.Reader) (err error)

DecodeMsg implements msgp.Decodable

func (*Batch) EncodeMsg 

func (z *Batch) EncodeMsg(en *msgp.Writer) (err error)

EncodeMsg implements msgp.Encodable

func (*Batch) MarshalMsg 

func (z *Batch) MarshalMsg(b []byte) (o []byte, err error)

MarshalMsg implements msgp.Marshaler

func (*Batch) Msgsize 

func (z *Batch) Msgsize() (s int)

Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message

func (*Batch) UnmarshalMsg 

func (z *Batch) UnmarshalMsg(bts []byte) (o []byte, err error)

UnmarshalMsg implements msgp.Unmarshaler

type BloomData added in v0.25.1 

type BloomData struct {
	Filter []byte `msgpack:"f"`
}

type BloomPayload added in v0.12.0 

type BloomPayload struct {
	Indices []uint32 `msgpack:"i"`
}

type CMSPayload added in v0.12.0 

type CMSPayload struct {
	Element string `msgpack:"el"`
	Count   int64  `msgpack:"c"`
}

type Capacitor 

type Capacitor struct {
	// contains filtered or unexported fields
}

Capacitor represents an active-active, local-first replicated caching node. It manages local key-value storage, cluster membership discovery, logical clocks, and replication orchestration.

func New 

func New(cfg Config) (*Capacitor, error)

New initializes, configures, and starts a local Capacitor node. This constructs the storage, starts the replication stream listener, and registers the node with the gossip cluster.

func (*Capacitor) BloomAdd added in v0.12.0 

func (f *Capacitor) BloomAdd(ctx context.Context, key string, element string) (bool, error)

BloomAdd inserts an element into the Bloom Filter.

func (*Capacitor) BloomExists added in v0.12.0 

func (f *Capacitor) BloomExists(ctx context.Context, key string, element string) (bool, error)

BloomExists checks if the element is likely present in the Bloom Filter.

func (*Capacitor) CMSIncrement added in v0.12.0 

func (f *Capacitor) CMSIncrement(ctx context.Context, key string, element string, count int64) error

CMSIncrement increments the count of an element in the Count-Min Sketch grid.

func (*Capacitor) CMSQuery added in v0.12.0 

func (f *Capacitor) CMSQuery(ctx context.Context, key string, element string) (int64, error)

CMSQuery returns the estimated frequency of the element in the Count-Min Sketch grid.

func (*Capacitor) Close 

func (f *Capacitor) Close() error

Close gracefully shuts down the node, leaving the SWIM gossip group, shutting down active replication streams, and closing local storage engines.

func (*Capacitor) Delete added in v0.5.0 

func (f *Capacitor) Delete(ctx context.Context, key string) error

Delete removes a key-value pair from the local store and replicates the deletion tombstone to the cluster.

func (*Capacitor) Exists added in v0.4.0 

func (f *Capacitor) Exists(ctx context.Context, key string) (bool, error)

Exists checks if a key exists in the cache and has not expired.

func (*Capacitor) Get 

func (f *Capacitor) Get(ctx context.Context, key string) (string, error)

Get retrieves a key-value pair's serialized value from the local cache database. Returns an empty string and nil error if the key is not found or has expired.

func (*Capacitor) GetCount 

func (f *Capacitor) GetCount(ctx context.Context, key string) (int64, error)

GetCount retrieves the converged aggregate sum of a distributed counter across all nodes.

func (*Capacitor) GetMetric 

func (f *Capacitor) GetMetric(ctx context.Context, key string) (Metric, error)

GetMetric retrieves the aggregated Metric details (count, sum, average) for the specified key.

func (*Capacitor) GetMetrics 

func (f *Capacitor) GetMetrics() []Summary

GetMetrics returns a snapshot summary of all built-in metrics (latencies, counts).

func (*Capacitor) GetScan added in v0.3.0 

func (f *Capacitor) GetScan(ctx context.Context, key string, dest any) error

GetScan retrieves a key's value and unmarshals it into the destination pointer dest (similar to json.Unmarshal or database rows.Scan).

func (*Capacitor) GossipAddr added in v0.24.8 

func (f *Capacitor) GossipAddr() string

GossipAddr returns the actual local address and port bound by the gossip layer.

func (*Capacitor) HLLAdd added in v0.12.0 

func (f *Capacitor) HLLAdd(ctx context.Context, key string, elements ...string) (bool, error)

HLLAdd adds one or more elements to the HyperLogLog cardinality estimator.

func (*Capacitor) HLLCount added in v0.12.0 

func (f *Capacitor) HLLCount(ctx context.Context, key string) (int64, error)

HLLCount returns the estimated cardinality of the HyperLogLog structure.

func (*Capacitor) Increment 

func (f *Capacitor) Increment(ctx context.Context, key string) (int64, error)

Increment increments a distributed PN-Counter key by 1.

func (*Capacitor) IncrementBy 

func (f *Capacitor) IncrementBy(ctx context.Context, key string, delta int64) (int64, error)

IncrementBy increments a distributed PN-Counter key by the specified delta. It tracks counts per-node to construct CRDT conflict-free convergence.

func (*Capacitor) IncrementMetric 

func (f *Capacitor) IncrementMetric(ctx context.Context, key string, delta float64) (Metric, error)

IncrementMetric records a floating-point update to a distributed aggregate metric (tracking both hit frequency and aggregated sums).

func (*Capacitor) IncrementMetricParallel 

func (f *Capacitor) IncrementMetricParallel(ctx context.Context, keys map[string]float64) (map[string]Metric, error)

IncrementMetricParallel updates multiple aggregate metrics concurrently.

func (*Capacitor) IncrementParallel 

func (f *Capacitor) IncrementParallel(ctx context.Context, keys []string) (map[string]int64, error)

IncrementParallel performs concurrent increment calls for multiple counter keys.

func (*Capacitor) IncrementSlidingWindow 

func (f *Capacitor) IncrementSlidingWindow(ctx context.Context, key string, window time.Duration) (int64, error)

IncrementSlidingWindow appends an event timestamp for rate limiting or hit tracking, and returns the count of active occurrences within the rolling window duration.

func (*Capacitor) ListLeftPopScan added in v0.11.0 

func (f *Capacitor) ListLeftPopScan(ctx context.Context, key string, destPtr any) (bool, error)

ListLeftPopScan pops and returns the first element from the head of the list, unmarshaling it into destPtr.

func (*Capacitor) ListLeftPush added in v0.11.0 

func (f *Capacitor) ListLeftPush(ctx context.Context, key string, value any) error

ListLeftPush prepends a value to the list.

func (*Capacitor) ListLen added in v0.11.0 

func (f *Capacitor) ListLen(ctx context.Context, key string) (int, error)

ListLen returns the length of the list.

func (*Capacitor) ListRangeScan added in v0.11.0 

func (f *Capacitor) ListRangeScan(ctx context.Context, key string, start, stop int, destSlicePtr any) error

ListRangeScan retrieves a sub-range of elements and unmarshals them into destSlicePtr.

func (*Capacitor) ListRightPopScan added in v0.11.0 

func (f *Capacitor) ListRightPopScan(ctx context.Context, key string, destPtr any) (bool, error)

ListRightPopScan pops and returns the last element from the tail of the list, unmarshaling it into destPtr.

func (*Capacitor) ListRightPush added in v0.11.0 

func (f *Capacitor) ListRightPush(ctx context.Context, key string, value any) error

ListRightPush appends a value to the list.

func (*Capacitor) MapExists added in v0.9.0 

func (f *Capacitor) MapExists(ctx context.Context, key, field string) (bool, error)

MapExists returns true if a field exists and is active in a Map.

func (*Capacitor) MapGetAll added in v0.9.0 

func (f *Capacitor) MapGetAll(ctx context.Context, key string) (map[string]string, error)

MapGetAll returns a copy of the raw map containing all non-expired fields as string values.

func (*Capacitor) MapGetAllScan added in v0.9.0 

func (f *Capacitor) MapGetAllScan(ctx context.Context, key string, destStructPtr any) (bool, error)

MapGetAllScan retrieves all active fields of a Map and unmarshals them into destStructPtr using `map` tags. Returns true if the map exists and contains active fields, false otherwise.

func (*Capacitor) MapGetExpiry added in v0.9.0 

func (f *Capacitor) MapGetExpiry(ctx context.Context, key, field string) (int64, bool, error)

MapGetExpiry retrieves the expiry Unix timestamp in milliseconds of a field.

func (*Capacitor) MapGetMScan added in v0.9.0 

func (f *Capacitor) MapGetMScan(ctx context.Context, key string, fieldDestMap map[string]any) (int, error)

MapGetMScan retrieves multiple fields and unmarshals them into pointers mapped in fieldDestMap. Returns the number of successfully scanned fields.

func (*Capacitor) MapGetScan added in v0.9.0 

func (f *Capacitor) MapGetScan(ctx context.Context, key, field string, destPtr any) (bool, error)

MapGetScan retrieves a field's value and unmarshals it into destPtr. Returns true if found and scanned, false if the field does not exist or expired.

func (*Capacitor) MapGetTTL added in v0.9.0 

func (f *Capacitor) MapGetTTL(ctx context.Context, key, field string) (time.Duration, bool, error)

MapGetTTL retrieves the remaining TTL duration of a field.

func (*Capacitor) MapIncrementBy added in v0.9.0 

func (f *Capacitor) MapIncrementBy(ctx context.Context, key, field string, delta float64) (float64, error)

MapIncrementBy increments a numeric field by a delta. Returns the final score.

func (*Capacitor) MapKeysScan added in v0.9.0 

func (f *Capacitor) MapKeysScan(ctx context.Context, key string, destSlicePtr any) error

MapKeysScan unmarshals all active field keys of a Map into destSlicePtr.

func (*Capacitor) MapLen added in v0.9.0 

func (f *Capacitor) MapLen(ctx context.Context, key string) (int, error)

MapLen returns the count of active fields in a Map.

func (*Capacitor) MapRemove added in v0.9.0 

func (f *Capacitor) MapRemove(ctx context.Context, key, field string) (bool, error)

MapRemove deletes a field from a map.

func (*Capacitor) MapRemoveExpiry added in v0.9.0 

func (f *Capacitor) MapRemoveExpiry(ctx context.Context, key, field string) (bool, error)

MapRemoveExpiry is an alias to MapRemoveTTL.

func (*Capacitor) MapRemoveTTL added in v0.9.0 

func (f *Capacitor) MapRemoveTTL(ctx context.Context, key, field string) (bool, error)

MapRemoveTTL removes the TTL of a field, converting it to persistent.

func (*Capacitor) MapSet added in v0.9.0 

func (f *Capacitor) MapSet(ctx context.Context, key, field string, value any, ttl time.Duration) (bool, error)

MapSet sets a field in a map to a primitive value with an optional TTL. Returns true if the field is newly created, false if updated.

func (*Capacitor) MapSetExpiry added in v0.9.0 

func (f *Capacitor) MapSetExpiry(ctx context.Context, key, field string, expireTimeMs int64) (bool, error)

MapSetExpiry sets a field-level expiry timestamp in Unix milliseconds.

func (*Capacitor) MapSetTTL added in v0.9.0 

func (f *Capacitor) MapSetTTL(ctx context.Context, key, field string, ttl time.Duration) (bool, error)

MapSetTTL sets a field-level TTL on a map.

func (*Capacitor) MapValuesScan added in v0.9.0 

func (f *Capacitor) MapValuesScan(ctx context.Context, key string, destSlicePtr any) error

MapValuesScan unmarshals all active field values of a Map into destSlicePtr.

func (*Capacitor) NMapExists added in v0.10.0 

func (f *Capacitor) NMapExists(ctx context.Context, key, field string) (bool, error)

NMapExists returns true if a field exists and is active in an NMap.

func (*Capacitor) NMapGetAll added in v0.10.0 

func (f *Capacitor) NMapGetAll(ctx context.Context, key string) (map[string]any, error)

NMapGetAll retrieves all active fields of an NMap as deserialized values.

func (*Capacitor) NMapGetAllScan added in v0.10.0 

func (f *Capacitor) NMapGetAllScan(ctx context.Context, key string, destStructPtr any) (bool, error)

NMapGetAllScan retrieves all active fields of an NMap and unmarshals them into destStructPtr using `map` tags. Returns true if the map exists and contains active fields, false otherwise.

func (*Capacitor) NMapGetExpiry added in v0.10.0 

func (f *Capacitor) NMapGetExpiry(ctx context.Context, key, field string) (int64, bool, error)

NMapGetExpiry retrieves the expiry Unix timestamp in milliseconds of a field.

func (*Capacitor) NMapGetMScan added in v0.10.0 

func (f *Capacitor) NMapGetMScan(ctx context.Context, key string, fieldDestMap map[string]any) (int, error)

NMapGetMScan retrieves multiple fields and unmarshals them into pointers mapped in fieldDestMap. Returns the number of successfully scanned fields.

func (*Capacitor) NMapGetScan added in v0.10.0 

func (f *Capacitor) NMapGetScan(ctx context.Context, key, field string, destPtr any) (bool, error)

NMapGetScan retrieves a field's value (primitive or nested structure) and unmarshals it into destPtr. Returns true if found and scanned, false if the field does not exist or expired.

func (*Capacitor) NMapGetTTL added in v0.10.0 

func (f *Capacitor) NMapGetTTL(ctx context.Context, key, field string) (time.Duration, bool, error)

NMapGetTTL retrieves the remaining TTL duration of a field.

func (*Capacitor) NMapIncrementBy added in v0.10.0 

func (f *Capacitor) NMapIncrementBy(ctx context.Context, key, field string, delta float64) (float64, error)

NMapIncrementBy increments a numeric field by a delta. Returns the final score.

func (*Capacitor) NMapKeysScan added in v0.10.0 

func (f *Capacitor) NMapKeysScan(ctx context.Context, key string, destSlicePtr any) error

NMapKeysScan unmarshals all active field keys of an NMap into destSlicePtr.

func (*Capacitor) NMapLen added in v0.10.0 

func (f *Capacitor) NMapLen(ctx context.Context, key string) (int, error)

NMapLen returns the count of active fields in an NMap.

func (*Capacitor) NMapRemove added in v0.10.0 

func (f *Capacitor) NMapRemove(ctx context.Context, key, field string) (bool, error)

NMapRemove deletes a field from an NMap.

func (*Capacitor) NMapRemoveExpiry added in v0.10.0 

func (f *Capacitor) NMapRemoveExpiry(ctx context.Context, key, field string) (bool, error)

NMapRemoveExpiry is an alias to NMapRemoveTTL.

func (*Capacitor) NMapRemoveTTL added in v0.10.0 

func (f *Capacitor) NMapRemoveTTL(ctx context.Context, key, field string) (bool, error)

NMapRemoveTTL removes the TTL of a field, converting it to persistent.

func (*Capacitor) NMapSet added in v0.10.0 

func (f *Capacitor) NMapSet(ctx context.Context, key, field string, value any, ttl time.Duration) (bool, error)

NMapSet sets a field in a nested map to any value (nested map/struct/slice/primitive) with an optional TTL. Returns true if the field is newly created, false if updated.

func (*Capacitor) NMapSetExpiry added in v0.10.0 

func (f *Capacitor) NMapSetExpiry(ctx context.Context, key, field string, expireTimeMs int64) (bool, error)

NMapSetExpiry sets a field-level expiry timestamp in Unix milliseconds.

func (*Capacitor) NMapSetTTL added in v0.10.0 

func (f *Capacitor) NMapSetTTL(ctx context.Context, key, field string, ttl time.Duration) (bool, error)

NMapSetTTL sets a field-level TTL on an NMap.

func (*Capacitor) NMapValuesScan added in v0.10.0 

func (f *Capacitor) NMapValuesScan(ctx context.Context, key string, destSlicePtr any) error

NMapValuesScan unmarshals all active field values of an NMap into destSlicePtr.

func (*Capacitor) Publish added in v0.13.0 

func (f *Capacitor) Publish(ctx context.Context, topic string, message any) error

Publish distributes a message payload to all subscribers of a topic across the cluster.

func (*Capacitor) Set 

func (f *Capacitor) Set(ctx context.Context, key string, value any, ttl time.Duration) error

Set writes a key-value pair to the local store and appends it to the replication log to propagate it to other nodes in the cluster. If a positive TTL is specified, the key-value pair will automatically expire.

func (*Capacitor) SetAdd added in v0.6.0 

func (f *Capacitor) SetAdd(ctx context.Context, key string, member any) (bool, error)

SetAdd adds a member to a distributed replicated set. The member can be of any comparable or serializable type (e.g. string, int, struct).

func (*Capacitor) SetCard added in v0.6.0 

func (f *Capacitor) SetCard(ctx context.Context, key string) (int64, error)

SetCard returns the cardinality (number of active members) of the set.

func (*Capacitor) SetIsMember added in v0.6.0 

func (f *Capacitor) SetIsMember(ctx context.Context, key string, member any) (bool, error)

SetIsMember checks if a member is currently present in the set.

func (*Capacitor) SetMIsMember added in v0.6.0 

func (f *Capacitor) SetMIsMember(ctx context.Context, key string, members ...any) ([]bool, error)

SetMIsMember checks membership for multiple members in a single call.

func (*Capacitor) SetMembers added in v0.6.0 

func (f *Capacitor) SetMembers(ctx context.Context, key string) ([]string, error)

SetMembers returns all active members in the set as strings (or serialized representations).

func (*Capacitor) SetMembersScan added in v0.6.0 

func (f *Capacitor) SetMembersScan(ctx context.Context, key string, destSlicePtr any) error

SetMembersScan unmarshals all active members in the set into the slice pointed to by destSlicePtr.

func (*Capacitor) SetMove added in v0.6.0 

func (f *Capacitor) SetMove(ctx context.Context, source string, destination string, member any) (bool, error)

SetMove moves a member from a source set to a destination set.

func (*Capacitor) SetPop added in v0.6.0 

func (f *Capacitor) SetPop(ctx context.Context, key string) (string, error)

SetPop removes and returns a random member from the set as a string.

func (*Capacitor) SetPopScan added in v0.6.0 

func (f *Capacitor) SetPopScan(ctx context.Context, key string, dest any) (bool, error)

SetPopScan removes a random member from the set and unmarshals it into dest. Returns true if a member was popped, false if the set was empty.

func (*Capacitor) SetRandMember added in v0.6.0 

func (f *Capacitor) SetRandMember(ctx context.Context, key string) (string, error)

SetRandMember returns a random member from the set as a string.

func (*Capacitor) SetRandMemberScan added in v0.6.0 

func (f *Capacitor) SetRandMemberScan(ctx context.Context, key string, dest any) (bool, error)

SetRandMemberScan retrieves a random member from the set and unmarshals it into dest. Returns true if a member was found, false if the set was empty.

func (*Capacitor) SetRemove added in v0.6.0 

func (f *Capacitor) SetRemove(ctx context.Context, key string, member any) (bool, error)

SetRemove removes a member from a distributed replicated set.

func (*Capacitor) SortedSetAdd added in v0.7.0 

func (f *Capacitor) SortedSetAdd(ctx context.Context, key string, score float64, member any) (bool, error)

SortedSetAdd adds a member with a score to a sorted set, or updates its score. Returns true if the member is newly added to the set, or false if its score was updated.

func (*Capacitor) SortedSetCard added in v0.7.0 

func (f *Capacitor) SortedSetCard(ctx context.Context, key string) (int64, error)

SortedSetCard returns the cardinality (number of active members) of a sorted set.

func (*Capacitor) SortedSetCount added in v0.7.0 

func (f *Capacitor) SortedSetCount(ctx context.Context, key string, min, max float64) (int64, error)

SortedSetCount returns the number of members with a score in [min, max].

func (*Capacitor) SortedSetIncrementBy added in v0.7.0 

func (f *Capacitor) SortedSetIncrementBy(ctx context.Context, key string, member any, delta float64) (float64, error)

SortedSetIncrementBy increments the score of a member by a delta. Returns the new score of the member.

func (*Capacitor) SortedSetPopScan added in v0.7.0 

func (f *Capacitor) SortedSetPopScan(ctx context.Context, key string, dest any, outScore *float64) (bool, error)

SortedSetPopScan removes a random member from the sorted set and unmarshals it into dest. Returns true if a member was popped and its score is returned, false if the set was empty.

func (*Capacitor) SortedSetRangeByScoreScan added in v0.7.0 

func (f *Capacitor) SortedSetRangeByScoreScan(ctx context.Context, key string, min, max float64, destSlicePtr any, destScoresPtr *[]float64) error

SortedSetRangeByScoreScan unmarshals a range of members (by score range min to max) into destSlicePtr. It also optionally populates destScoresPtr if not nil.

func (*Capacitor) SortedSetRangeScan added in v0.7.0 

func (f *Capacitor) SortedSetRangeScan(ctx context.Context, key string, start, stop int64, destSlicePtr any, destScoresPtr *[]float64) error

SortedSetRangeScan unmarshals a range of members (by rank index start to stop) into destSlicePtr. It also optionally populates destScoresPtr if not nil.

func (*Capacitor) SortedSetRank added in v0.7.0 

func (f *Capacitor) SortedSetRank(ctx context.Context, key string, member any) (int64, error)

SortedSetRank returns the 0-based rank of a member (ordered ascending by score). Returns -1 if the member is not present in the set.

func (*Capacitor) SortedSetRemove added in v0.7.0 

func (f *Capacitor) SortedSetRemove(ctx context.Context, key string, member any) (bool, error)

SortedSetRemove removes a member from a sorted set. Returns true if the member was present and removed, or false if it was not in the set.

func (*Capacitor) SortedSetRevRangeByScoreScan added in v0.8.0 

func (f *Capacitor) SortedSetRevRangeByScoreScan(ctx context.Context, key string, max, min float64, destSlicePtr any, destScoresPtr *[]float64) error

SortedSetRevRangeByScoreScan unmarshals a range of members in reverse score order (max-to-min) into destSlicePtr. It also optionally populates destScoresPtr if not nil.

func (*Capacitor) SortedSetRevRangeScan added in v0.8.0 

func (f *Capacitor) SortedSetRevRangeScan(ctx context.Context, key string, start, stop int64, destSlicePtr any, destScoresPtr *[]float64) error

SortedSetRevRangeScan unmarshals a range of members in reverse score order (high-to-low) into destSlicePtr. It also optionally populates destScoresPtr if not nil.

func (*Capacitor) SortedSetScore added in v0.7.0 

func (f *Capacitor) SortedSetScore(ctx context.Context, key string, member any) (float64, bool, error)

SortedSetScore retrieves the score of a member in a sorted set. Returns the score and a boolean indicating whether the member is present.

func (*Capacitor) Subscribe added in v0.13.0 

func (f *Capacitor) Subscribe(ctx context.Context, topic string) (*Subscription, error)

Subscribe registers a subscriber Go channel for a topic.

type Config 

type Config struct {
	// NodeID is the unique identifier for this node in the cluster. If left empty,
	// a hostname-based identifier will be generated automatically.
	NodeID string

	// BindAddr is the network address for the gossip memberlist to bind to.
	BindAddr string

	// BindPort is the port number utilized for gossip memberlist communications.
	BindPort int

	// StreamPort is the port number utilized for replication TCP streams.
	StreamPort int

	// AdvertiseAddr is the IP address advertised to other nodes for establishing
	// replication TCP streams.
	AdvertiseAddr string

	// Peers is the initial list of bootstrap addresses ("IP:port") of active cluster nodes.
	Peers []string

	// DataPath is the local directory path where BadgerDB files are persistently stored.
	DataPath string

	// LogSize is the capacity (maximum number of entries) of the in-memory circular Delta Log.
	LogSize uint64

	// TLSConfig is the optional configuration used to secure node-to-node replication streams using mTLS.
	TLSConfig *tls.Config

	// AuthToken is a shared secret token used to authenticate gossip join requests and TCP streams.
	AuthToken string

	// Logger is the structured logging engine injected into the Capacitor instance.
	Logger Logger

	// DisableMetrics disables internal metrics latency tracking for maximum read/write performance.
	DisableMetrics bool
}

Config defines the configuration parameters for a Capacitor node instance.

type DeltaLog 

type DeltaLog struct {
	Overflows uint64 // Count of overflow capacity eviction events
	// contains filtered or unexported fields
}

DeltaLog is an in-memory binary circular buffer of operations.

func NewDeltaLog 

func NewDeltaLog(capacity uint64) *DeltaLog

func (*DeltaLog) Append 

func (l *DeltaLog) Append(entry LogEntry) uint64

func (*DeltaLog) GetEntriesRaw 

func (l *DeltaLog) GetEntriesRaw(startSeq uint64, limit int, maxBytes int) [][]byte

func (*DeltaLog) Head 

func (l *DeltaLog) Head() uint64

func (*DeltaLog) PutEntriesRaw added in v0.24.6 

func (l *DeltaLog) PutEntriesRaw(entries [][]byte)

type HLC 

type HLC struct {
	ClockSmashes uint64
	// contains filtered or unexported fields
}

HLC (Hybrid Logical Clock) implements a causality-preserving clock.

func NewHLC 

func NewHLC() *HLC

func (*HLC) Now 

func (h *HLC) Now() Timestamp

Now generates a new timestamp.

func (*HLC) SetMaxOffset 

func (h *HLC) SetMaxOffset(offset time.Duration)

SetMaxOffset sets the maximum allowed clock drift.

func (*HLC) Update 

func (h *HLC) Update(remote Timestamp) (Timestamp, error)

Update updates the local clock based on a received remote timestamp.

type HLLData added in v0.25.1 

type HLLData struct {
	Registers []byte `msgpack:"r"`
}

type HLLPayload added in v0.12.0 

type HLLPayload struct {
	Index uint32 `msgpack:"i"`
	Value byte   `msgpack:"v"`
}

type Handshake 

type Handshake struct {
	LastSeenSeq uint64 `msg:"ls"`
	AuthToken   string `msg:"at,omitempty"`
}

func (*Handshake) DecodeMsg 

func (z *Handshake) DecodeMsg(dc *msgp.Reader) (err error)

DecodeMsg implements msgp.Decodable

func (Handshake) EncodeMsg 

func (z Handshake) EncodeMsg(en *msgp.Writer) (err error)

EncodeMsg implements msgp.Encodable

func (Handshake) MarshalMsg 

func (z Handshake) MarshalMsg(b []byte) (o []byte, err error)

MarshalMsg implements msgp.Marshaler

func (Handshake) Msgsize 

func (z Handshake) Msgsize() (s int)

Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message

func (*Handshake) UnmarshalMsg 

func (z *Handshake) UnmarshalMsg(bts []byte) (o []byte, err error)

UnmarshalMsg implements msgp.Unmarshaler

type ListElement added in v0.11.0 

type ListElement struct {
	ID        Timestamp `msgpack:"id"`
	ParentID  Timestamp `msgpack:"p"`
	Value     []byte    `msgpack:"v"`
	DeletedAt Timestamp `msgpack:"d"`
}

type ListPayload added in v0.11.0 

type ListPayload struct {
	ID        Timestamp `msgpack:"id"`
	ParentID  Timestamp `msgpack:"p"`
	Value     []byte    `msgpack:"v"`
	DeletedAt Timestamp `msgpack:"d"`
}

type LogEntry 

type LogEntry struct {
	Seq    uint64    `json:"s" msg:"s"`
	TS     Timestamp `json:"t" msg:"t"`
	BornAt int64     `json:"ba" msg:"ba"` // NEW: For end-to-end latency tracking
	Op     MsgType   `json:"o" msg:"o"`
	Key    string    `json:"k" msg:"k"`
	NodeID string    `json:"n,omitempty" msg:"n,omitempty"`
	Value  []byte    `json:"v,omitempty" msg:"v,omitempty"`
	Delta  float64   `json:"d,omitempty" msg:"d,omitempty"`
	TTL    int64     `json:"ttl,omitempty" msg:"ttl,omitempty"`
}

LogEntry represents a single operation in the delta log.

func (*LogEntry) MarshalMsg 

func (z *LogEntry) MarshalMsg(b []byte) (o []byte, err error)

MarshalMsg implements msgp.Marshaler

func (*LogEntry) Msgsize 

func (z *LogEntry) Msgsize() (s int)

Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message

func (*LogEntry) UnmarshalMsg 

func (z *LogEntry) UnmarshalMsg(bts []byte) (o []byte, err error)

UnmarshalMsg implements msgp.Unmarshaler

type Logger 

type Logger interface {
	Debug(msg string, args ...any)
	Info(msg string, args ...any)
	Warn(msg string, args ...any)
	Error(msg string, args ...any)
}

Logger is a generic structured logging interface.

type MapField added in v0.9.0 

type MapField struct {
	Value     string    `msgpack:"v"`
	AddedAt   Timestamp `msgpack:"a"`
	RemovedAt Timestamp `msgpack:"r"`
	ExpiresAt int64     `msgpack:"e"` // Unix nanoseconds (0 means no expiry)
}

type MapPayload added in v0.9.0 

type MapPayload struct {
	Field     string `msgpack:"f"`
	Value     string `msgpack:"v"`
	ExpiresAt int64  `msgpack:"e"`
}

type Message 

type Message struct {
	Type      MsgType `msg:"t"`
	Key       string  `msg:"k"`
	Value     []byte  `msg:"v,omitempty"`
	ValueObj  any     `msg:"-"` // Used for lazy serialization
	Delta     float64 `msg:"d,omitempty"`
	TTL       int64   `msg:"ttl,omitempty"` // Seconds
	Timestamp int64   `msg:"ts"`            // UnixNano for LWW
	NodeID    string  `msg:"n"`             // Source node
}

Message is the container for all gossip data.

func Decode 

func Decode(data []byte) (*Message, error)

Decode deserializes the message from MsgPack.

func (*Message) DecodeMsg 

func (z *Message) DecodeMsg(dc *msgp.Reader) (err error)

DecodeMsg implements msgp.Decodable

func (*Message) Encode 

func (m *Message) Encode() ([]byte, error)

Encode serializes the message to MsgPack.

func (*Message) EncodeMsg 

func (z *Message) EncodeMsg(en *msgp.Writer) (err error)

EncodeMsg implements msgp.Encodable

func (*Message) MarshalMsg 

func (z *Message) MarshalMsg(b []byte) (o []byte, err error)

MarshalMsg implements msgp.Marshaler

func (*Message) Msgsize 

func (z *Message) Msgsize() (s int)

Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message

func (*Message) UnmarshalMsg 

func (z *Message) UnmarshalMsg(bts []byte) (o []byte, err error)

UnmarshalMsg implements msgp.Unmarshaler

type Metric 

type Metric struct {
	Count int64   `json:"count" msg:"c"`
	Sum   float64 `json:"sum" msg:"s"`
}

Metric represents a composite historical value tracking frequency and volume.

type MetricsTracker 

type MetricsTracker struct {

	// Latencies
	SetLat         Stat
	GetLat         Stat
	IncrLat        Stat
	ReplicateLat   Stat // End-to-end (BornAt to Apply)
	BadgerWriteLat Stat // BadgerDB write latency during flush

	// Counters
	ReplicationFailures uint64
	PeerConnectFailures uint64
	ActiveSubscriptions int64

	// Memory (Peak)
	PeakAlloc     uint64
	PeakHeapAlloc uint64
	PeakSys       uint64
	// contains filtered or unexported fields
}

func NewMetricsTracker 

func NewMetricsTracker() *MetricsTracker

func (*MetricsTracker) GetSummary 

func (m *MetricsTracker) GetSummary() []Summary

func (*MetricsTracker) Stop 

func (m *MetricsTracker) Stop()

type MsgType 

type MsgType byte

MsgType defines the type of gossip message. 

const (
	MsgSet MsgType = iota
	MsgIncr
	MsgMetric
	MsgWindow
	MsgDelete
	MsgSetAdd
	MsgSetRemove
	MsgSortedSetAdd
	MsgSortedSetRemove
	MsgSortedSetIncrement
	MsgMapSet
	MsgMapRemove
	MsgNMapSet
	MsgNMapRemove
	MsgListInsert
	MsgListDelete
	MsgHLLAdd
	MsgBloomAdd
	MsgCMSIncrement
	MsgPubSubPublish
)

func (*MsgType) DecodeMsg 

func (z *MsgType) DecodeMsg(dc *msgp.Reader) (err error)

DecodeMsg implements msgp.Decodable

func (MsgType) EncodeMsg 

func (z MsgType) EncodeMsg(en *msgp.Writer) (err error)

EncodeMsg implements msgp.Encodable

func (MsgType) MarshalMsg 

func (z MsgType) MarshalMsg(b []byte) (o []byte, err error)

MarshalMsg implements msgp.Marshaler

func (MsgType) Msgsize 

func (z MsgType) Msgsize() (s int)

Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message

func (*MsgType) UnmarshalMsg 

func (z *MsgType) UnmarshalMsg(bts []byte) (o []byte, err error)

UnmarshalMsg implements msgp.Unmarshaler

type NMapField added in v0.10.0 

type NMapField struct {
	Value     []byte    `msgpack:"v"`
	AddedAt   Timestamp `msgpack:"a"`
	RemovedAt Timestamp `msgpack:"r"`
	ExpiresAt int64     `msgpack:"e"` // Unix nanoseconds (0 means no expiry)
}

type NMapPayload added in v0.10.0 

type NMapPayload struct {
	Field     string `msgpack:"f"`
	Value     []byte `msgpack:"v"`
	ExpiresAt int64  `msgpack:"e"`
}

type PubSubMessage added in v0.13.0 

type PubSubMessage struct {
	Topic   string
	Payload []byte
}

func (*PubSubMessage) Scan added in v0.13.0 

func (m *PubSubMessage) Scan(destPtr any) error

type PubSubPayload added in v0.13.0 

type PubSubPayload struct {
	Topic   string `msgpack:"t"`
	Payload []byte `msgpack:"p"`
}

type SetElement added in v0.6.0 

type SetElement struct {
	AddedAt   Timestamp `msgpack:"a"`
	RemovedAt Timestamp `msgpack:"r"`
}

type SortedSetElement added in v0.7.0 

type SortedSetElement struct {
	Score     float64   `msgpack:"s"`
	AddedAt   Timestamp `msgpack:"a"`
	RemovedAt Timestamp `msgpack:"r"`
}

type Stat 

type Stat struct {
	Count int64
	Sum   int64 // Nanoseconds
	Max   int64 // Nanoseconds
}

func (*Stat) Record 

func (s *Stat) Record(duration time.Duration)

type StreamClient 

type StreamClient struct {
	// contains filtered or unexported fields
}

func NewStreamClient 

func NewStreamClient(tlsConfig *tls.Config, authToken string, metrics *MetricsTracker) *StreamClient

func (*StreamClient) Close 

func (c *StreamClient) Close()

func (*StreamClient) CloseConn 

func (c *StreamClient) CloseConn(nodeID string)

func (*StreamClient) SendBatch 

func (c *StreamClient) SendBatch(ctx context.Context, nodeID string, addr string, batch Batch, lastSeenPeerSeq uint64) error

type StreamServer 

type StreamServer struct {
	// contains filtered or unexported fields
}

func NewStreamServer 

func NewStreamServer(cp *Capacitor, addr string, tlsConfig *tls.Config) (*StreamServer, error)

func (*StreamServer) Start 

func (s *StreamServer) Start()

func (*StreamServer) Stop 

func (s *StreamServer) Stop()

type Subscription added in v0.13.0 

type Subscription struct {
	Channel <-chan PubSubMessage
	// contains filtered or unexported fields
}

func (*Subscription) Unsubscribe added in v0.13.0 

func (s *Subscription) Unsubscribe()

type Summary 

type Summary struct {
	Metric     string
	Average    time.Duration
	Peak       time.Duration
	Count      int64
	PeakMemMB  float64
	PeakHeapMB float64
}

type Timestamp 

type Timestamp struct {
	Physical int64 `json:"p" msg:"p"`
	Logical  int32 `json:"l" msg:"l"`
}

func (Timestamp) After 

func (t Timestamp) After(other Timestamp) bool

func (*Timestamp) DecodeMsg 

func (z *Timestamp) DecodeMsg(dc *msgp.Reader) (err error)

DecodeMsg implements msgp.Decodable

func (Timestamp) EncodeMsg 

func (z Timestamp) EncodeMsg(en *msgp.Writer) (err error)

EncodeMsg implements msgp.Encodable

func (Timestamp) GreaterOrEqual 

func (t Timestamp) GreaterOrEqual(other Timestamp) bool

GreaterOrEqual returns true if this timestamp is logically greater than or equal to the other timestamp.

func (Timestamp) MarshalMsg 

func (z Timestamp) MarshalMsg(b []byte) (o []byte, err error)

MarshalMsg implements msgp.Marshaler

func (Timestamp) Msgsize 

func (z Timestamp) Msgsize() (s int)

Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message

func (Timestamp) String 

func (t Timestamp) String() string

func (*Timestamp) UnmarshalMsg 

func (z *Timestamp) UnmarshalMsg(bts []byte) (o []byte, err error)

UnmarshalMsg implements msgp.Unmarshaler

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.