capacitor

package module
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Jun 30, 2026 License: Apache-2.0 Imports: 23 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

💻 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")

// 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)

🧪 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

Variables

View Source 
var ErrClockSmash = errors.New("HLC clock smash detected")

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 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) 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) 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) 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) 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.

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 {
	// 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

type HLC 

type HLC struct {
	// 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 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 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 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)

	// 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
)

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 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) *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 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.