capacitor

package module
v0.7.0 Latest Latest
Warning

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

 Go to latest Published: Jul 4, 2026 License: Apache-2.0 Imports: 24 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
SortedSetPopScan ~930 (1k set size) 0.00093 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)

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

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

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
	MsgDelete
	MsgSetAdd
	MsgSetRemove
	MsgSortedSetAdd
	MsgSortedSetRemove
)

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