cacheclient

package module
v1.13.0 Latest Latest
Warning

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

 Go to latest Published: Sep 8, 2026 License: Apache-2.0 Imports: 22 Imported by: 1

Documentation

Index

Constants

View Source 
const (
	// ModeAuto automatically detects cluster vs simple mode (default)
	ModeAuto ConnectionMode = "auto"
	// ModeSimple uses direct connections without topology service
	ModeSimple ConnectionMode = "simple"
	// ModeCluster uses topology service for smart routing
	ModeCluster ConnectionMode = "cluster"

	// DefaultRefreshInterval is the default topology refresh interval
	DefaultRefreshInterval = 30 * time.Second
	// MaxMessageSize is the maximum message size for gRPC
	MaxMessageSize = 128 * 1024 * 1024 // 128MB
	// TopologyDetectTimeout is the timeout for detecting cluster topology
	TopologyDetectTimeout = 2 * time.Second
	// DefaultBufferSize is the default buffer size for streaming operations
	DefaultBufferSize = 64 * 1024 // 64KB
	// ConnectionHealthCheckInterval is the interval for checking connection health
	ConnectionHealthCheckInterval = 30 * time.Second
	// ConnectionErrorWindow is the time window for tracking connection errors
	ConnectionErrorWindow = 30 * time.Second
	// DefaultConnectionPoolSize is the default number of connections per address
	DefaultConnectionPoolSize = 4
	// DefaultKeepaliveTime is the default keepalive time for client connections
	DefaultKeepaliveTime = 30 * time.Second
	// DefaultKeepaliveTimeout is the default keepalive timeout for client connections
	DefaultKeepaliveTimeout = 10 * time.Second

	// MaxPageLimit is the maximum number of keys to return in a single page
	MaxPageLimit = 1000
)

Variables

This section is empty.

Functions

func DefaultDialOptions 

func DefaultDialOptions() []grpc.DialOption

DefaultDialOptions returns the default gRPC dial options

Types

type CacheClient 

type CacheClient interface {
	// Basic operations
	Put(ctx context.Context, key string, data []byte, ttlSeconds int64) error
	Get(ctx context.Context, key string) ([]byte, error)
	Delete(ctx context.Context, key string) error

	// Conditional (compare-and-swap) operations (issue #254). A write applies
	// only if the key's current version equals the caller's expectation; a lost
	// race returns a *VersionMismatchError carrying the current version. An
	// absent key reads with an observation token rather than 0 (issue #267):
	// passing it back as expected orders the write against any CAS delete
	// stamped after the observation, while expected == 0 remains the unordered
	// put-if-absent. A CAS delete of a missing or dead key records that fence.
	// Note: adding these methods extends CacheClient — external implementers
	// must add them too.
	GetWithVersion(ctx context.Context, key string) (data []byte, version uint64, found bool, err error)
	PutIfVersion(ctx context.Context, key string, data []byte, ttlSeconds int64, expected uint64) (newVersion uint64, err error)
	DeleteIfVersion(ctx context.Context, key string, expected uint64) error

	// Streaming CAS for values larger than the unary message cap (issue #258).
	// Same semantics as PutIfVersion/GetWithVersion; the value is streamed
	// rather than buffered.
	PutStreamIfVersion(ctx context.Context, key string, r io.Reader, ttlSeconds int64, expected uint64) (newVersion uint64, err error)
	GetStreamWithVersion(ctx context.Context, key string, w io.Writer) (version uint64, found bool, err error)
	List(ctx context.Context, prefix string) ([]string, error)
	ListPage(ctx context.Context, prefix string, limit int, continuationToken string) (keys []string, nextToken string, hasMore bool, err error)
	ListPageWithValues(ctx context.Context, prefix string, limit int, continuationToken string) (entries []KeyValue, nextToken string, hasMore bool, err error)

	// Streaming operations
	PutStream(ctx context.Context, key string, r io.Reader, ttlSeconds int64) error
	GetStream(ctx context.Context, key string, w io.Writer) error

	// Range operations
	GetRange(ctx context.Context, key string, start, end int64) ([]byte, error)
	GetRangeStream(ctx context.Context, key string, start, end int64, w io.Writer) error

	// Lifecycle
	Close() error

	// Info
	GetMode() ConnectionMode
	GetConnectedNodes() []string
}

CacheClient is the common interface for both SimpleClient and ClusterClient

type Client 

type Client struct {
	CacheClient
	// contains filtered or unexported fields
}

Client is a wrapper that maintains backward compatibility It delegates to either SimpleClient or ClusterClient based on mode detection

func New 

func New(addrs ...string) (*Client, error)

New creates a new client with default configuration

func NewWithConfig 

func NewWithConfig(config *ClientConfig) (*Client, error)

NewWithConfig creates a new client with custom configuration

func (*Client) FetchClusterState 

func (c *Client) FetchClusterState() (*clusterpb.ClusterState, error)

FetchClusterState fetches the current cluster state (cluster mode only)

func (*Client) FetchTopology 

func (c *Client) FetchTopology() (*clusterpb.ClusterTopology, error)

FetchTopology fetches the current cluster topology (cluster mode only)

func (*Client) GetMode 

func (c *Client) GetMode() ConnectionMode

GetMode returns the actual connection mode being used

func (*Client) GetNodeIDForKey 

func (c *Client) GetNodeIDForKey(key string) (string, error)

GetNodeIDForKey returns the node ID that owns the given key (cluster mode only)

func (*Client) GetNodeInfoForKey 

func (c *Client) GetNodeInfoForKey(key string) (nodeID, address string, err error)

GetNodeInfoForKey returns both the node ID and address that owns the given key (cluster mode only)

func (*Client) GetTopologyEpoch 

func (c *Client) GetTopologyEpoch() uint64

GetTopologyEpoch returns the current topology epoch (cluster mode only)

func (*Client) HasRing 

func (c *Client) HasRing() bool

HasRing returns true if the consistent hash ring is initialized (cluster mode only)

func (*Client) IsClusterMode 

func (c *Client) IsClusterMode() bool

IsClusterMode returns true if the client is in cluster mode

type ClientConfig 

type ClientConfig struct {
	Addrs              []string          // One or more server addresses
	Mode               ConnectionMode    // Connection mode (default: "auto")
	RefreshInterval    time.Duration     // Topology refresh for cluster mode (default: 30s)
	ConnectionPoolSize int               // Number of connections per address (default: 4)
	DialOpts           []grpc.DialOption // Optional gRPC dial options
}

ClientConfig contains configuration for the unified Client

func (*ClientConfig) SetDefaults 

func (c *ClientConfig) SetDefaults()

SetDefaults sets default values for unspecified config fields

type ClusterClient 

type ClusterClient struct {
	*Operations // Embedded for shared operations
	// contains filtered or unexported fields
}

ClusterClient implements a cluster-aware cache client with topology support

func NewClusterClient 

func NewClusterClient(config *ClientConfig) (*ClusterClient, error)

NewClusterClient creates a new ClusterClient with the given configuration

func (*ClusterClient) Close 

func (c *ClusterClient) Close() error

Close closes all connections and stops background goroutines

func (*ClusterClient) Delete 

func (c *ClusterClient) Delete(ctx context.Context, key string) error

Delete removes a key with retry logic

func (*ClusterClient) FetchClusterState 

func (c *ClusterClient) FetchClusterState() (*clusterpb.ClusterState, error)

FetchClusterState fetches the current cluster state from any available node.

func (*ClusterClient) FetchTopology 

func (c *ClusterClient) FetchTopology() (*clusterpb.ClusterTopology, error)

FetchTopology fetches the current topology (exposed for testing)

func (*ClusterClient) Get 

func (c *ClusterClient) Get(ctx context.Context, key string) ([]byte, error)

Get retrieves a value with retry logic

func (*ClusterClient) GetConnectedNodes 

func (c *ClusterClient) GetConnectedNodes() []string

GetConnectedNodes returns the addresses of all connected nodes

func (*ClusterClient) GetConnectionCount 

func (c *ClusterClient) GetConnectionCount() int

GetConnectionCount returns the number of active connections (exposed for testing)

func (*ClusterClient) GetMode 

func (c *ClusterClient) GetMode() ConnectionMode

GetMode returns the connection mode

func (*ClusterClient) GetNodeIDForKey 

func (c *ClusterClient) GetNodeIDForKey(key string) (string, error)

GetNodeIDForKey returns the node ID that owns the given key.

func (*ClusterClient) GetNodeInfoForKey 

func (c *ClusterClient) GetNodeInfoForKey(key string) (nodeID, address string, err error)

GetNodeInfoForKey returns both the node ID and address that owns the given key.

func (*ClusterClient) GetRange 

func (c *ClusterClient) GetRange(ctx context.Context, key string, start, end int64) ([]byte, error)

GetRange retrieves a byte range with retry logic

func (*ClusterClient) GetRangeStream 

func (c *ClusterClient) GetRangeStream(ctx context.Context, key string, start, end int64, w io.Writer) error

GetRangeStream streams a byte range with retry logic

func (*ClusterClient) GetStream 

func (c *ClusterClient) GetStream(ctx context.Context, key string, w io.Writer) error

GetStream streams a value with retry logic

func (*ClusterClient) GetStreamWithVersion added in v1.12.0 

func (c *ClusterClient) GetStreamWithVersion(ctx context.Context, key string, w io.Writer) (uint64, bool, error)

GetStreamWithVersion streams a value and its CAS version with routing-error retry, mirroring plain GetStream: the read is idempotent, but a routing error after partial output cannot be retried without corrupting w, so we retry only when nothing has been written yet.

func (*ClusterClient) GetTopologyEpoch 

func (c *ClusterClient) GetTopologyEpoch() uint64

GetTopologyEpoch returns the current topology epoch

func (*ClusterClient) GetWithVersion added in v1.12.0 

func (c *ClusterClient) GetWithVersion(ctx context.Context, key string) ([]byte, uint64, bool, error)

GetWithVersion reads a key's value and CAS version with routing-error retry. A read is idempotent, so on a routing error we refresh topology and retry once — the same treatment plain Get gets. A version mismatch is not a routing error and never occurs on a read.

func (*ClusterClient) HasRing 

func (c *ClusterClient) HasRing() bool

HasRing returns true if the token ring is initialized and has tokens

func (*ClusterClient) Put 

func (c *ClusterClient) Put(ctx context.Context, key string, data []byte, ttlSeconds int64) error

Put stores a value with retry logic for routing errors

func (*ClusterClient) RoundRobinRoute 

func (c *ClusterClient) RoundRobinRoute() (*connection, error)

RoundRobinRoute selects a connection using round-robin (for List operation) Implements Router interface

func (*ClusterClient) Route 

func (c *ClusterClient) Route(key string) (*connection, error)

Route determines which connection to use for a given key Implements Router interface Optimized to minimize lock contention using cached routing decisions

func (*ClusterClient) UpdateTopology 

func (c *ClusterClient) UpdateTopology(topology *clusterpb.ClusterTopology) error

UpdateTopology manually updates the topology (exposed for testing)

type ConnectionMode 

type ConnectionMode string

ConnectionMode defines how the client connects to servers

type EpochGetter 

type EpochGetter func() uint64

EpochGetter is a function that returns the current client epoch

type EpochMismatchHandler 

type EpochMismatchHandler func(clientEpoch, serverEpoch uint64)

EpochMismatchHandler is called when server epoch differs from client epoch

type KeyValue 

type KeyValue struct {
	Key   string
	Value []byte
	// ValueLength is the size of the value in bytes, set even when the value was
	// omitted for exceeding the List-with-values per-value size cap.
	ValueLength int64
	// ValueOmitted is true when the value was omitted for exceeding that cap;
	// Value is nil in that case.
	ValueOmitted bool
}

KeyValue holds a key and its associated value bytes.

type MemoryCache 

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

MemoryCache implements CacheClient with in-memory storage. Useful for testing without a real cache server.

func NewMemoryCache 

func NewMemoryCache() *MemoryCache

NewMemoryCache creates a new in-memory cache.

func (*MemoryCache) Close 

func (m *MemoryCache) Close() error

Close clears the cache. This is a no-op for cleanup purposes.

func (*MemoryCache) Delete 

func (m *MemoryCache) Delete(ctx context.Context, key string) error

Delete removes a key from the cache.

func (*MemoryCache) DeleteIfVersion added in v1.12.0 

func (m *MemoryCache) DeleteIfVersion(ctx context.Context, key string, expected uint64) error

DeleteIfVersion deletes only if key admits expected (see PutIfVersion), and records the delete as the key's fence (issue #267): delete-if-absent on a missing or dead key is not a no-op, it writes or moves the fence so a populate that observed absence before this delete loses to it.

func (*MemoryCache) Get 

func (m *MemoryCache) Get(ctx context.Context, key string) ([]byte, error)

Get retrieves data by key.

func (*MemoryCache) GetConnectedNodes 

func (m *MemoryCache) GetConnectedNodes() []string

GetConnectedNodes returns a single "memory" node identifier.

func (*MemoryCache) GetMode 

func (m *MemoryCache) GetMode() ConnectionMode

GetMode returns ModeSimple since MemoryCache is a single-node implementation.

func (*MemoryCache) GetRange 

func (m *MemoryCache) GetRange(ctx context.Context, key string, start, end int64) ([]byte, error)

GetRange retrieves a byte range from the cached data.

func (*MemoryCache) GetRangeStream 

func (m *MemoryCache) GetRangeStream(ctx context.Context, key string, start, end int64, w io.Writer) error

GetRangeStream retrieves a byte range and writes it to the writer.

func (*MemoryCache) GetStream 

func (m *MemoryCache) GetStream(ctx context.Context, key string, w io.Writer) error

GetStream retrieves data and writes it to the writer.

func (*MemoryCache) GetStreamWithVersion added in v1.12.0 

func (m *MemoryCache) GetStreamWithVersion(ctx context.Context, key string, w io.Writer) (uint64, bool, error)

GetStreamWithVersion writes the value to w and returns its version/presence.

func (*MemoryCache) GetWithVersion added in v1.12.0 

func (m *MemoryCache) GetWithVersion(ctx context.Context, key string) ([]byte, uint64, bool, error)

GetWithVersion returns key's value and version. For an absent or expired key found is false and the version is an observation token (issue #267): the fence stamp of the CAS delete that removed it, else a fresh stamp. Pass it back to PutIfVersion to order the write against any later delete.

func (*MemoryCache) List 

func (m *MemoryCache) List(ctx context.Context, prefix string) ([]string, error)

List returns all keys matching the prefix.

func (*MemoryCache) ListPage 

func (m *MemoryCache) ListPage(ctx context.Context, prefix string, limit int, continuationToken string) (keys []string, nextToken string, hasMore bool, err error)

ListPage returns a paginated list of keys matching the prefix.

func (*MemoryCache) ListPageWithValues 

func (m *MemoryCache) ListPageWithValues(ctx context.Context, prefix string, limit int, continuationToken string) (entries []KeyValue, nextToken string, hasMore bool, err error)

ListPageWithValues returns a paginated list of key-value pairs matching the prefix.

func (*MemoryCache) Put 

func (m *MemoryCache) Put(ctx context.Context, key string, data []byte, ttlSeconds int64) error

Put stores data with an optional TTL (0 means no expiration).

func (*MemoryCache) PutIfVersion added in v1.12.0 

func (m *MemoryCache) PutIfVersion(ctx context.Context, key string, data []byte, ttlSeconds int64, expected uint64) (uint64, error)

PutIfVersion writes only if key admits expected: a live key needs an exact match; an absent key admits 0 (put-if-absent) or an observation token, which a CAS delete stamped after the observation rejects (issue #267). Returns the new version or a *VersionMismatchError carrying the current version (the fence stamp, for a fenced key).

func (*MemoryCache) PutStream 

func (m *MemoryCache) PutStream(ctx context.Context, key string, r io.Reader, ttlSeconds int64) error

PutStream reads all data from the reader and stores it.

func (*MemoryCache) PutStreamIfVersion added in v1.12.0 

func (m *MemoryCache) PutStreamIfVersion(ctx context.Context, key string, r io.Reader, ttlSeconds int64, expected uint64) (uint64, error)

PutStreamIfVersion buffers the stream and delegates to PutIfVersion (this test double keeps everything in memory).

type Operations 

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

Operations provides shared implementation of cache operations

func NewOperations 

func NewOperations(router Router) *Operations

NewOperations creates a new Operations instance

func (*Operations) Delete 

func (o *Operations) Delete(ctx context.Context, key string) error

Delete removes a key from the cache

func (*Operations) DeleteIfVersion added in v1.12.0 

func (o *Operations) DeleteIfVersion(ctx context.Context, key string, expected uint64) error

DeleteIfVersion deletes the key only if it admits expected (see PutIfVersion). A CAS delete leaves a fence (issue #267): delete-if-absent on a missing or dead key is not a no-op, it records the delete so a put that observed absence before it loses. A lost race returns a *VersionMismatchError.

func (*Operations) Get 

func (o *Operations) Get(ctx context.Context, key string) ([]byte, error)

Get retrieves a value from the cache

func (*Operations) GetRange 

func (o *Operations) GetRange(ctx context.Context, key string, start, end int64) ([]byte, error)

GetRange retrieves a byte range from the cache

func (*Operations) GetRangeStream 

func (o *Operations) GetRangeStream(ctx context.Context, key string, start, end int64, w io.Writer) error

GetRangeStream streams a byte range from the cache

func (*Operations) GetStream 

func (o *Operations) GetStream(ctx context.Context, key string, w io.Writer) error

GetStream streams a value from the cache

func (*Operations) GetStreamWithVersion added in v1.12.0 

func (o *Operations) GetStreamWithVersion(ctx context.Context, key string, w io.Writer) (uint64, bool, error)

GetStreamWithVersion streams a key's value to w and returns its CAS version and presence, without buffering the whole value (issue #258). The version and data are one atomic snapshot: the server sends version+found first, then the bytes.

func (*Operations) GetWithVersion added in v1.12.0 

func (o *Operations) GetWithVersion(ctx context.Context, key string) (data []byte, version uint64, found bool, err error)

GetWithVersion returns a key's value together with its current CAS version. found is false when the key is absent, expired, or deleted; version is then an observation token (issue #267): the fence stamp of the CAS delete that removed the key, or a fresh stamp. Pass it back as expected to PutIfVersion to order the write against any later delete; pass 0 to skip ordering.

func (*Operations) List 

func (o *Operations) List(ctx context.Context, prefix string) ([]string, error)

List lists keys with optional prefix Returns all keys matching the prefix (automatically handles pagination)

func (*Operations) ListPage 

func (o *Operations) ListPage(ctx context.Context, prefix string, limit int, continuationToken string) ([]string, string, bool, error)

ListPage returns a single page of keys with pagination support Returns: (keys, continuationToken, hasMore, error)

func (*Operations) ListPageWithValues 

func (o *Operations) ListPageWithValues(ctx context.Context, prefix string, limit int, continuationToken string) ([]KeyValue, string, bool, error)

ListPageWithValues returns a single page of key-value pairs with pagination support. Returns: (entries, continuationToken, hasMore, error)

func (*Operations) Put 

func (o *Operations) Put(ctx context.Context, key string, data []byte, ttlSeconds int64) error

Put stores a value in the cache

func (*Operations) PutIfVersion added in v1.12.0 

func (o *Operations) PutIfVersion(ctx context.Context, key string, data []byte, ttlSeconds int64, expected uint64) (uint64, error)

PutIfVersion writes the value only if the key admits expected: a live key needs an exact version match; an absent key admits 0 (put-if-absent) or an observation token from GetWithVersion, which a CAS delete stamped after that observation rejects (issue #267). It returns the new version on success, or a *VersionMismatchError carrying the current version — for a fenced key, the fence stamp to refetch and retry with — on a lost race.

func (*Operations) PutStream 

func (o *Operations) PutStream(ctx context.Context, key string, r io.Reader, ttlSeconds int64) error

PutStream streams data to the cache

func (*Operations) PutStreamIfVersion added in v1.12.0 

func (o *Operations) PutStreamIfVersion(ctx context.Context, key string, r io.Reader, ttlSeconds int64, expected uint64) (uint64, error)

PutStreamIfVersion streams a value and applies it only if the key's current version equals expected (0 = put-if-absent) — the streaming form of PutIfVersion for values larger than the unary message cap (issue #258).

type Router 

type Router interface {
	Route(key string) (*connection, error)
	RoundRobinRoute() (*connection, error)
}

Router is an interface for routing keys to connections

type SimpleClient 

type SimpleClient struct {
	*Operations // Embedded for shared operations
	// contains filtered or unexported fields
}

SimpleClient implements a simple round-robin cache client

func NewSimpleClient 

func NewSimpleClient(config *ClientConfig) (*SimpleClient, error)

NewSimpleClient creates a new SimpleClient with the given configuration

func (*SimpleClient) Close 

func (c *SimpleClient) Close() error

Close closes all connections

func (*SimpleClient) GetConnectedNodes 

func (c *SimpleClient) GetConnectedNodes() []string

GetConnectedNodes returns the addresses of all connected nodes

func (*SimpleClient) GetMode 

func (c *SimpleClient) GetMode() ConnectionMode

GetMode returns the connection mode

func (*SimpleClient) RoundRobinRoute 

func (c *SimpleClient) RoundRobinRoute() (*connection, error)

RoundRobinRoute selects a connection using round-robin (for operations without keys) Implements Router interface

func (*SimpleClient) Route 

func (c *SimpleClient) Route(key string) (*connection, error)

Route selects a connection using hash-based routing for better key locality Implements Router interface

type TokenRing 

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

TokenRing implements token-based consistent hashing matching the server's dskit ring. It uses FNV-1a 32-bit hash and binary search for O(log n) lookups. The ring is structured as a sorted array of tokens. Each token is owned by a specific node.

Thread Safety: Updates use atomic pointer swapping, so reads are completely lock-free.

func NewTokenRing 

func NewTokenRing() *TokenRing

NewTokenRing creates a new empty token ring

func (*TokenRing) GetNodeAddresses 

func (r *TokenRing) GetNodeAddresses() map[string]string

GetNodeAddresses returns a copy of all node addresses

func (*TokenRing) GetNodeForKey 

func (r *TokenRing) GetNodeForKey(key string) (string, error)

GetNodeForKey returns the node address that owns the given key. Uses FNV-1a 32-bit hash (same as server) + binary search.

The algorithm: 1. Hash the key to get a 32-bit token 2. Binary search for the first token in the ring >= our hash 3. If we're past the last token, wrap around to the first (ring semantics) 4. Return the address of the node that owns that token

This method is lock-free - it reads from an atomically swapped pointer.

func (*TokenRing) GetNodeIDForKey 

func (r *TokenRing) GetNodeIDForKey(key string) (string, error)

GetNodeIDForKey returns the node ID that owns the given key. This is useful for debugging and testing. This method is lock-free.

func (*TokenRing) GetNodeInfoForKey 

func (r *TokenRing) GetNodeInfoForKey(key string) (nodeID, address string, err error)

GetNodeInfoForKey returns both the node ID and address that owns the given key. This method is lock-free.

func (*TokenRing) IsEmpty 

func (r *TokenRing) IsEmpty() bool

IsEmpty returns true if the ring has no tokens

func (*TokenRing) NodeCount 

func (r *TokenRing) NodeCount() int

NodeCount returns the number of unique nodes in the ring

func (*TokenRing) TokenCount 

func (r *TokenRing) TokenCount() int

TokenCount returns the total number of tokens in the ring

func (*TokenRing) Update 

func (r *TokenRing) Update(nodeTokens map[string][]uint32, nodeAddresses map[string]string)

Update rebuilds the ring with new token assignments. nodeTokens is a map of nodeID -> list of tokens owned by that node. nodeAddresses is a map of nodeID -> listen address for that node.

This method uses atomic pointer swapping, so concurrent reads see either the old state or the new state atomically - never a partially updated state. This eliminates contention between topology updates and client requests.

type TopologyManager 

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

TopologyManager manages cluster topology for ClusterClient. It uses a token-based ring that matches the server's dskit ring implementation, ensuring consistent key routing between client and server.

func NewTopologyManager 

func NewTopologyManager(seedAddrs []string, refreshInterval time.Duration, dialOpts []grpc.DialOption) (*TopologyManager, error)

NewTopologyManager creates a new topology manager

func (*TopologyManager) FetchTopology 

func (tm *TopologyManager) FetchTopology(ctx context.Context) (*clusterpb.ClusterTopology, error)

FetchTopology fetches the cluster topology from available nodes

func (*TopologyManager) GetNodeAddresses 

func (tm *TopologyManager) GetNodeAddresses() map[string]string

GetNodeAddresses returns all node addresses

func (*TopologyManager) GetNodeForKey 

func (tm *TopologyManager) GetNodeForKey(key string) (string, error)

GetNodeForKey returns the node address for a given key. Uses FNV-1a 32-bit hash + binary search (same as server).

func (*TopologyManager) GetNodeIDForKey 

func (tm *TopologyManager) GetNodeIDForKey(key string) (string, error)

GetNodeIDForKey returns the node ID for a given key. Useful for debugging and testing.

func (*TopologyManager) GetNodeInfoForKey 

func (tm *TopologyManager) GetNodeInfoForKey(key string) (nodeID, address string, err error)

GetNodeInfoForKey returns both the node ID and address for a given key.

func (*TopologyManager) GetRing 

func (tm *TopologyManager) GetRing() *TokenRing

GetRing returns the underlying token ring

func (*TopologyManager) GetTopologyEpoch 

func (tm *TopologyManager) GetTopologyEpoch() uint64

GetTopologyEpoch returns the current topology epoch. Uses atomic load for lock-free access.

func (*TopologyManager) RefreshTopology 

func (tm *TopologyManager) RefreshTopology(ctx context.Context) (bool, error)

RefreshTopology refreshes the topology

func (*TopologyManager) TopologyRefreshLoop 

func (tm *TopologyManager) TopologyRefreshLoop(ctx context.Context, updateFn func())

TopologyRefreshLoop periodically refreshes the cluster topology

func (*TopologyManager) UpdateTopology 

func (tm *TopologyManager) UpdateTopology(topology *clusterpb.ClusterTopology) (map[string]bool, bool)

UpdateTopology updates the internal state based on new topology. With content-addressable epochs, same epoch = same ring state, so we use equality check (not >=) to detect changes.

type VersionMismatchError added in v1.12.0 

type VersionMismatchError struct {
	Key            string
	CurrentVersion uint64
}

VersionMismatchError reports a lost conditional (CAS) operation: the key's current version did not equal the caller's expected version (issue #254). CurrentVersion carries the version the server observed, so the caller can re-read, re-decide, and retry with a fresh precondition. It is a normal outcome, not a transport failure.

func IsVersionMismatch added in v1.12.0 

func IsVersionMismatch(err error) (*VersionMismatchError, bool)

IsVersionMismatch reports whether err is (or wraps) a VersionMismatchError, returning it for access to CurrentVersion.

func (*VersionMismatchError) Error added in v1.12.0 

func (e *VersionMismatchError) Error() string

Source Files

View all Source files

Directories

Path Synopsis
cmd
ycsb

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.