effects

package
v1.0.1 Latest Latest
Warning

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

 Go to latest Published: Aug 28, 2026 License: AGPL-3.0 Imports: 39 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// BloomMinBytes is the smallest bitmap: 4KB ≈ 3k keys at target density.
	BloomMinBytes = 4096

	// BloomTargetBitsPerKey sizes a build: m = next power of two ≥ 10·n bits
	// lands the fill between ~0.3 and ~0.5 for n keys.
	BloomTargetBitsPerKey = 10

	// BloomResizeFill is the set-bit fraction past which the holder should
	// rebuild larger — above it the FP rate degrades past ~2%.
	BloomResizeFill = 0.55
)
View Source 
const DefaultSerializationThreshold = 3

DefaultSerializationThreshold is the number of consecutive aborts on a key before escalating to serialized coordination.

View Source 
const FlushKey = "__swytch:flush"

FlushKey is the special key used to signal a full index wipe (FLUSHDB/FLUSHALL). Lives under the __swytch: namespace so it's pinned in the cache and can't collide with a user-chosen key.

Variables

View Source 
var ErrAuthorityDropped = errors.New("authority dropped: not subscribed to this key")

ErrAuthorityDropped signals that HandleRemote rejected the inbound effect because this node has no authority over the key. The transport layer interprets this as "do not respond" — neither ACK nor NACK is sent — so the sender's tracked replication times out rather than counting us as a successful first-ACK replica. The sender's other peers (those with authority) still get the chance to accept the write.

View Source 
var ErrBootstrapIncomplete = errors.New("bootstrap incomplete: some peers unreachable")

ErrBootstrapIncomplete is returned by ensureSubscribed when the bootstrap could not fetch the full causal chain because some peers are unreachable. A background retry continues until the chain is complete.

View Source 
var ErrCDNBlobMissing = errors.New("cdn blob missing")

ErrCDNBlobMissing marks a CDN fetch that answered 404: the blob is provably absent from cloud storage, as opposed to the cloud being unreachable. Cloud tip markers are candidates, never an authoritative frontier — a node builds its own truth by walking the DAG it can actually read — so a hole under this sentinel is not this engine's to repair. classifyCloudTips reports the holed tip as pending; the cluster layer's reconcile loop retries it, and Cloud's own billing sweep is what eventually heals or prunes the marker.

View Source 
var ErrCloudUnavailable = errors.New("cloud unavailable")

ErrCloudUnavailable marks a read that could not be answered because the Cloud consult failed or returned a frontier we could not fully fetch. It must surface to the client as an error, never as a miss: Cloud provably holds data for the key, so answering "no such key" would be indistinguishable from data loss.

View Source 
var ErrCuckooFull = errors.New("cuckoo filter is full")

ErrCuckooFull is returned by Add when the filter could not place a fingerprint after the maximum number of evictions. The own-filter recovers by rebuilding at a larger capacity from the index, which is the authoritative key set; peer filters are never Added to (they arrive pre-built) so they never see this.

View Source 
var ErrRegionPartitioned = errors.New("region partitioned: not all same-region peers are reachable")

ErrRegionPartitioned is returned by Flush in SafeMode when not all same-region peers are reachable.

View Source 
var ErrTxnAborted = errors.New("transaction aborted")

ErrTxnAborted is returned by Flush when a transaction loses FWW or encounters a real conflict that cannot be resolved.

Functions

func BloomHash 

func BloomHash(name []byte) uint64

BloomHash is the one hash of a name every position derives from — the only thing a caller needs to retain per name to rebuild a filter.

func BuildOffsetNotify 

func BuildOffsetNotify(nodeID pb.NodeID, offset Tip, eff *pb.Effect, data []byte, traceCtx context.Context) *pb.OffsetNotify

BuildOffsetNotify constructs a single-effect notification. EffectData is wire format: [4-byte LE keyLen][key][protoData]. traceCtx is optional; when non-nil, OTel trace context is injected into the notification.

func Compress 

func Compress(b []byte) []byte

Compress returns the zstd-compressed form of b. Safe for concurrent use.

func CompressEffectValues 

func CompressEffectValues(eff *pb.Effect)

CompressEffectValues applies value compression to an effect about to be emitted: the data arm directly, and a snapshot's materialized state via copy — snapshot state may share element pointers with cached reduced state, which is immutable by contract, so compression swaps in fresh structs rather than writing through shared ones.

func ComputeForkChoiceHash 

func ComputeForkChoiceHash(nodeID proto.NodeID, hlc *timestamppb.Timestamp) []byte

ComputeForkChoiceHash computes SHA-256(nodeID_LE8 || hlc_LE8). The result is a deterministic 32-byte hash used for fork-choice tiebreaking. Lower hash wins, eliminating systematic advantage from raw HLC comparison.

func Decompress 

func Decompress(b []byte) ([]byte, error)

Decompress reverses Compress. Decoded size is bounded (see decoder init) to guard against decompression bombs from a misbehaving peer.

func ForkChoiceLess 

func ForkChoiceLess(a, b []byte) bool

ForkChoiceLess returns true if hash a is lexicographically less than hash b. Lower hash wins the fork-choice election.

func ForwardingEnabled 

func ForwardingEnabled() bool

ForwardingEnabled reports whether adaptive-serialization forwarding is compiled in. The redis handler checks it before extracting command keys that would only feed Forward/ForwardExec's early return — as a constant, the whole forwarding block dead-code-eliminates while disabled.

func InflatedSizeDelta 

func InflatedSizeDelta(eff *pb.Effect) uint64

InflatedSizeDelta is the byte count value compression hid from this effect's marshal: Σ (raw_len − len(stored)) over every compressed value. Billing adds it to the marshal length so raw_size keeps meaning "the customer's bytes" no matter how well their data compresses.

func MarshalEffect 

func MarshalEffect(eff *pb.Effect) ([]byte, error)

MarshalEffect serializes an Effect to its wire bytes, hex-encoding any map<string, X> keys inside an embedded ReducedEffect (Snapshot state). Proto3 requires string map keys to be valid UTF-8, but Redis-style collections (hashes, sets, zsets, streams) accept arbitrary byte sequences as element IDs — and the cluster membership encodes a uint64 as raw little-endian bytes. Without encoding, proto.Marshal silently rejects snapshots whose state has any binary-keyed elements.

The encoding is hex (always-on, no discriminator). The reverse step is in UnmarshalEffect. The original *pb.Effect is not mutated; if any sanitization is required the snapshot state is cloned first.

func ReduceBranch 

func ReduceBranch(effects []*pb.Effect) *pb.ReducedEffect

ReduceBranch reduces a linear chain of effects (oldest-first) into a single ReducedEffect.

func ReduceChain 

func ReduceChain(seed *pb.ReducedEffect, effects []*pb.Effect) *pb.ReducedEffect

ReduceChain reduces effects sequentially on top of a seed ReducedEffect. If seed is nil, behaves identically to ReduceBranch. This is used at DAG merge points: the seed is the merged result of concurrent dep subtrees, and effects are the linear chain above. The seed is never mutated; a clone is made before any modifications.

Virtual partitions: effects carry an optional `virtual` partition key that addresses nested state WITHIN the key (empty = root partition). Flat keys (no virtual effects, no seed partitions) take the fast path and reduce byte-identically to the pre-nesting engine. Otherwise the chain is split by virtual and each partition reduces independently via reduceFlat, with a root key-level DEL wiping all partitions (it deletes the whole key).

func UnmarshalEffect 

func UnmarshalEffect(data []byte, eff *pb.Effect) error

UnmarshalEffect parses wire bytes into eff and reverses the hex encoding applied by MarshalEffect on any embedded ReducedEffect (Snapshot state).

Types

type Bloom 

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

Bloom is an approximate set of key names with no false negatives: Has returns true for every name ever Set, and false-positives at a rate set by the fill fraction (FP ≈ fill^bloomK). It backs the cloud key-name filter on both sides of the wire — the dataplane builds one per cluster and ships it as a frame; the node queries it to gate read-miss cloud consults — so the hashing is fixed: a filter built on one process and queried on another must agree. Unlike CuckooChain it cannot grow; the holder rebuilds at a larger size when Fill crosses BloomResizeFill.

Not internally synchronized: a holder either treats a built filter as immutable (build fully, then publish by pointer swap — the frame path) or guards Set/Has under its own lock (CloudSync.filterMu).

func NewBloom 

func NewBloom(sizeBytes int) *Bloom

NewBloom returns a filter over sizeBytes of bitmap, rounded up to a power of two no smaller than BloomMinBytes.

func NewBloomForCount 

func NewBloomForCount(count int) *Bloom

NewBloomForCount returns a filter sized for count names at the target build density.

func ParseBloomFrame 

func ParseBloomFrame(data []byte) (*Bloom, error)

ParseBloomFrame decodes a Frame, copying the bitmap out of the (possibly transport-owned) buffer. Anything malformed — wrong version, undersized or non-power-of-two body — is rejected; the input crosses process boundaries.

func (*Bloom) Fill 

func (b *Bloom) Fill() float64

Fill is the set-bit fraction — the resize signal and the FP-rate base.

func (*Bloom) Frame 

func (b *Bloom) Frame() []byte

Frame serializes the filter for the wire: one version byte, then the bitmap.

func (*Bloom) Has 

func (b *Bloom) Has(name []byte) bool

Has is HasHash over a raw name.

func (*Bloom) HasHash 

func (b *Bloom) HasHash(h uint64) bool

HasHash reports whether the name may be in the set; false is authoritative.

func (*Bloom) Set 

func (b *Bloom) Set(name []byte) bool

Set is SetHash over a raw name.

func (*Bloom) SetHash 

func (b *Bloom) SetHash(h uint64) bool

SetHash sets the name's bits, reporting whether any bit flipped (false means the filter already covered the name — nothing to re-push).

func (*Bloom) SizeBytes 

func (b *Bloom) SizeBytes() int

SizeBytes is the bitmap size.

type Broadcaster 

type Broadcaster interface {
	Broadcast(notify *pb.OffsetNotify)
	BroadcastWithData(notify *pb.OffsetNotify, effectData []byte)
	Replicate(notify *pb.OffsetNotify, wireData []byte) error
	// ReplicateTo sends to a specific peer and waits for ACK/NACK.
	// Returns NackNotify slice on conflict, nil on ACK.
	ReplicateTo(notify *pb.OffsetNotify, wireData []byte, targetNodeID pb.NodeID) ([]*pb.NackNotify, error)
	// ReplicateMarshalled is ReplicateTo with a pre-marshalled notify body, so
	// a fan-out (the bind to every subscriber) marshals the body once instead
	// of once per peer. notify is passed for inspection; the body is the wire
	// payload.
	ReplicateMarshalled(notify *pb.OffsetNotify, notifyBody []byte, targetNodeID pb.NodeID) ([]*pb.NackNotify, error)
	// SendNack sends an enriched NACK to the originator.
	SendNack(nack *pb.NackNotify, targetNodeID pb.NodeID)
	// FetchFromAny fetches effect bytes from peers or the cloud CDN. hint
	// orders the two sources; the second is a fallback tried only after the
	// first actually failed — never raced.
	FetchFromAny(ref *pb.EffectRef, hint FetchHint) ([]byte, error)
	Fetch(ref *pb.EffectRef) ([]byte, error)
	PeerIDs() []pb.NodeID
	// AllRegionPeersReachable returns true if every same-region peer is
	// alive and has a verified symmetric path. Used by SafeMode to gate
	// writes: a key must not be written unless all region peers are reachable.
	AllRegionPeersReachable() bool
	// InMajorityPartition returns true if this node can reach a strict
	// majority of same-region nodes (including itself). Used by SafeMode
	// to block transactions when in a minority partition.
	InMajorityPartition() bool
	// ForwardTransaction sends a transaction to a specific peer for execution
	// (adaptive serialization §5). Returns the leader's response.
	ForwardTransaction(ctx context.Context, targetNodeID pb.NodeID, tx *pb.ForwardedTransaction) (*pb.ForwardedResponse, error)
}

Broadcaster sends effect notifications to cluster peers. Nil means standalone mode.

type BunnyStorage 

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

BunnyStorage implements ObjectStorage using Bunny CDN's storage API.

func NewBunnyStorage 

func NewBunnyStorage(uploadEndpoint, cdnEndpoint, accessKey string) *BunnyStorage

NewBunnyStorage creates a BunnyStorage instance.

func (*BunnyStorage) Download 

func (b *BunnyStorage) Download(ctx context.Context, path string) ([]byte, error)

func (*BunnyStorage) Upload 

func (b *BunnyStorage) Upload(ctx context.Context, path string, data []byte) error

type CloudEffect 

type CloudEffect struct {
	Tip      Tip
	Eff      *pb.Effect
	ProtoLen int
}

CloudEffect is one closure effect delivered inline by a GetTips response: the decrypted inner effect plus its marshaled proto length for cache accounting.

type CloudReader 

type CloudReader interface {
	// MayHold reports whether Cloud may hold key, answered from the pushed
	// key-name filter — free, no RPC. False is the filter's definite no and
	// lets a read free-miss without subscribing or consulting; true routes
	// the read through the subscribe + consult path, whose per-leaf
	// cloudConsulted marker caps the WAN round-trips.
	MayHold(key string) bool
	// CloudTips returns the candidate tip frontier Cloud's index holds for
	// key, or nil (with nil error) if Cloud holds nothing for it. These are
	// candidates, never authoritative: Cloud's tips directory is an index
	// that can lag or hole, and the engine's own DAG walk is what decides
	// what is actually installable. sidecar is the closure — every effect
	// reachable from those tips down to the LCA snapshot — delivered inline
	// by GetTips and installed into the effect cache before the tip walk, so
	// the walk runs locally instead of one WAN fetch per dep. The sidecar may
	// be partial (capped, or missing a blob the cloud is still fetching
	// back) — the walk stays the authority and pulls anything missing on
	// demand via FetchFromAny. missing names refs GetTips' own closure walk
	// found reachable but does not hold the blob for — a fact declared by
	// the storage side, letting the engine skip a redundant CDN round trip to
	// re-learn the same hole; it is a hint, not a complete inventory, so the
	// engine's walk still treats anything not listed here as normal. Content-
	// blind on the wire: the implementation maps key to its Cloud PRF image
	// and calls GetTips.
	CloudTips(ctx context.Context, key string) (tips []Tip, sidecar []CloudEffect, missing []Tip, err error)
	// MarkPending reports that key's Cloud frontier has a durable hole this
	// node could not walk past — see effects.ErrCDNBlobMissing. Cloud's own
	// reconcile loop owns retrying and eventually healing or pruning the
	// affected markers; this is not a request the engine expects synchronous
	// action on.
	MarkPending(key string)
}

CloudReader is the tiered-storage backstop: it reports the tip frontier that durable Cloud storage holds for a key. A key evicted from every live peer (e.g. its writer departed) may still live on Cloud, so a read that finds nothing cluster-wide asks Cloud for the frontier, installs it, and lets the cluster take over. Nil (unset) means no Cloud is configured and the read free-misses as before.

type Context 

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

Context tracks per-key state for dep chaining (Emit) and deferred index update + broadcast (Flush). One Context per command invocation.

func (*Context) Abort 

func (c *Context) Abort()

Abort discards pending index updates and broadcasts. Effects already written to the log remain durable but invisible (index not updated).

func (*Context) BeginTx 

func (c *Context) BeginTx()

BeginTx marks subsequent effects as transactional. This is called by read-modify-write commands (INCR, LPUSH, etc.) for atomicity AND by handleExec for MULTI/EXEC. Watch processing is NOT done here — use CheckWatches after BeginTx for EXEC.

func (*Context) CheckWatches 

func (c *Context) CheckWatches() bool

CheckWatches validates WATCH observations and emits transactional NOOPs for the Bind's read set. Must be called AFTER BeginTx (so NOOPs get IsTransactional=true). Only called from handleExec.

Returns false if any watched key was modified — the caller should abort the transaction and return a null array to the client.

func (*Context) ClearWatches 

func (c *Context) ClearWatches()

ClearWatches removes all watched keys. Called from UNWATCH, DISCARD, and EXEC's pre-BeginTx abort paths — all cases where the upcoming transaction is being discarded. Also drops the lazily-generated txnID so the next WATCH/MULTI starts fresh; if we're already inside MULTI/EXEC (c.inTx) the id is load-bearing for in-flight effects and must be preserved.

func (*Context) Emit 

func (c *Context) Emit(eff *pb.Effect, snapshotTips ...[]Tip) error

Emit writes a single effect to the log. The context tracks per-key state so that consecutive effects on the same key form a dep chain. The index is NOT updated until Flush.

Emit takes ownership of eff: it fills causality fields in place and caches the message itself, so the caller must not retain, reuse, or mutate eff (or its submessages) after the call. Byte-slice fields may alias pooled buffers (e.g. RESP parser args) — Emit copies those before caching.

For read-modify-write commands, pass the tip offsets returned by GetSnapshot as snapshotTips so that the first effect depends on the tips the handler actually read, not whatever the index contains now. Pure writes (SET, LPUSH) omit snapshotTips and Emit reads the index.

func (*Context) Flush 

func (c *Context) Flush() error

Flush updates the index for all touched keys and broadcasts all notifications per key for durability, then resets the context for reuse.

func (*Context) GetSnapshot 

func (c *Context) GetSnapshot(key string) (*pb.ReducedEffect, []Tip, error)

GetSnapshot returns the current materialized state of a key, including any unflushed effects from this context. Within MULTI/EXEC, earlier commands' effects are in the log but not yet in the index; this method reconstructs from the log so commands within the same transaction can see each other's writes. The returned ReducedEffect is immutable and may be shared by later reads.

When a txSnapshot is active (MULTI/EXEC with SSI), reads for keys not yet in the context use the snapshot instead of the live index, and a NOOP is emitted to record the read in the causal structure.

func (*Context) PendingKeys 

func (c *Context) PendingKeys() []string

PendingKeys returns the names of all keys with pending effects in this Context. Callers use this to acquire per-key locks before calling Flush — Flush's fork-choice critical section races with any other Flush on the same key, and the handler layer (redis handler, sql handler) is where that serialisation belongs.

func (*Context) RestoreSavepoint 

func (c *Context) RestoreSavepoint(sp *ContextSavepoint)

RestoreSavepoint replaces the Context's pending-key state with the snapshot. Any Emit issued after TakeSavepoint is discarded from the Context; its log offsets remain in the engine's log but never reach the index (identical to what happens to all effects on Abort).

func (*Context) SetTraceCtx 

func (c *Context) SetTraceCtx(ctx context.Context)

SetTraceCtx sets the OTel trace context for this command.

func (*Context) TakeSavepoint 

func (c *Context) TakeSavepoint() *ContextSavepoint

TakeSavepoint captures a deep-enough copy of the Context's pending key state that RestoreSavepoint can revert every Emit issued between Take and Restore. Cheap: each contextKey is a small struct with slice fields that we clone by reference (the underlying values — notifies, elementIDs — are immutable once added).

func (*Context) TraceCtx 

func (c *Context) TraceCtx() context.Context

TraceCtx returns the OTel trace context, or context.Background() if none set.

func (*Context) Watch 

func (c *Context) Watch(key string) error

Watch records a key for optimistic locking by emitting a NoopEffect immediately and flushing it. This records the observation in the causal log so all nodes can verify whether the key was modified between WATCH and EXEC. The NOOP offset and post-flush TipSet pointer are stored for comparison at BeginTx time.

The noop is tx-marked with the upcoming transaction's id (generated lazily here). This is what makes write-skew detection work across nodes: a competing write on a watched key becomes a structural fork sibling of the watching tx's bind, surfaced by reconstruct's pairwise fork-choice on the key. Without the TxnId, the noop is an anonymous committed effect that other writers can chain off, producing false sequential relationships (the run-25890153673 G1a).

Errors fail the WATCH: a read that can't be answered (e.g. the Cloud consult failed for an evicted key) must not silently record hadData=false — the EXEC verdict would then rest on a fabricated observation.

type ContextSavepoint 

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

ContextSavepoint is an opaque snapshot of a Context's pending per-key state. Restore it (via Context.RestoreSavepoint) to discard every Emit made after the snapshot was taken; already- written log offsets become orphans (never indexed), same as Abort. Used by higher-level SQL SAVEPOINT semantics.

Savepoints do NOT snapshot the engine's committed index — the engine's state is unchanged until Flush, so reverting the Context's pending keys is sufficient to roll back.

type CuckooChain 

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

CuckooChain is a queryable approximate set with NO false negatives, built from a sequence of CuckooFilters split into:

  • active: the single writable segment, always one we allocated and kept below cuckooChainChunk distinct keys, so Add on it never fails.
  • sealed: query-only segments — filled-up former active segments and any spliced peer segments. Never written to, so a near-full or foreign segment can't be corrupted by an Add.

It never shrinks; a deleted key lingers as a safe false-positive (a needless subscribe, never a wrong miss) until the chain is rebuilt.

func (*CuckooChain) Add 

func (cc *CuckooChain) Add(key string) bool

Add inserts key into the active segment, rotating first if that segment is at capacity. Returns true iff a new fingerprint was actually stored (false for an idempotent re-add), so callers can skip version bumps when the set did not change.

func (*CuckooChain) MarshalBinary 

func (cc *CuckooChain) MarshalBinary() ([]byte, error)

func (*CuckooChain) MaybeContains 

func (cc *CuckooChain) MaybeContains(key string) bool

MaybeContains reports whether key may be in the set; false is authoritative.

func (*CuckooChain) UnmarshalBinary 

func (cc *CuckooChain) UnmarshalBinary(data []byte) error

UnmarshalBinary decodes a chain as query-only segments (active stays nil) — a decoded peer chain is never written to.

type CuckooFilter 

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

CuckooFilter is an approximate set of keys with no false negatives: MaybeContains returns true for every key ever successfully Added, and false-positives at a bounded rate. It backs the per-node cluster key filters that let a read-only handler answer a miss without subscribing.

The filter is serialized and shipped between nodes (on NACKs), so the hashing and layout are fixed: a filter built on one node and queried on another must agree. Item placement (which candidate bucket a fingerprint landed in) does not affect queries — MaybeContains always probes both candidate buckets — so Add may relocate fingerprints freely.

func NewCuckooFilter 

func NewCuckooFilter(capacity int) *CuckooFilter

NewCuckooFilter returns a filter sized to hold at least capacity keys before relocations begin to fail. numBuckets is rounded up to a power of two so the index masks are cheap.

func (*CuckooFilter) Add 

func (c *CuckooFilter) Add(key string) error

Add inserts key. It is idempotent at the fingerprint level: re-adding a key whose fingerprint already sits in a candidate bucket is a no-op. Returns ErrCuckooFull if no slot could be freed.

func (*CuckooFilter) Count 

func (c *CuckooFilter) Count() uint64

Count returns the number of live fingerprints.

func (*CuckooFilter) LoadFactor 

func (c *CuckooFilter) LoadFactor() float64

LoadFactor is the fraction of slots occupied.

func (*CuckooFilter) MarshalBinary 

func (c *CuckooFilter) MarshalBinary() ([]byte, error)

MarshalBinary serializes the filter: numBuckets (8 bytes, big-endian) followed by every slot as a big-endian uint16.

func (*CuckooFilter) MaybeContains 

func (c *CuckooFilter) MaybeContains(key string) bool

MaybeContains reports whether key may be present. False positives are possible; false negatives are not (for keys successfully Added).

func (*CuckooFilter) UnmarshalBinary 

func (c *CuckooFilter) UnmarshalBinary(data []byte) error

UnmarshalBinary restores a filter produced by MarshalBinary. The decoded filter is query-only in practice (peer filters are never Added to), but Add still works should that change.

type Encryptor 

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

Encryptor seals payloads with XChaCha20-Poly1305 under a key derived from shared input keying material, with zstd compression inside the seal.

Encryption is symmetric on purpose: every sealing party already holds the connection secret the key derives from, so asymmetric sealing would add per-blob key-encapsulation overhead without separating any capabilities. Read-only consumers are carved out by the secret-derivation tree instead — hand them the encryption key and the key-name key but not the master secret, and they can decrypt blobs yet never derive the auth key a write requires. Sealed blobs are stored durably; changing this format strands them.

func NewEncryptorFromIKM 

func NewEncryptorFromIKM(ikm []byte) (*Encryptor, error)

NewEncryptorFromIKM derives an Encryptor's key deterministically from input keying material. Every holder of the same IKM arrives at the same key independently — this is how a cluster shares one cloud-payload key derived from the connection secret, with no key exchange and nothing for the cloud to see.

func (*Encryptor) OpenAndDecompress 

func (enc *Encryptor) OpenAndDecompress(sealed, info []byte) ([]byte, error)

OpenAndDecompress reverses SealAndCompress: authenticates and decrypts under the same info domain, then decompresses.

func (*Encryptor) SealAndCompress 

func (enc *Encryptor) SealAndCompress(plaintext, info []byte) ([]byte, error)

SealAndCompress compresses with zstd, then seals with XChaCha20-Poly1305 under a random nonce, returning nonce ‖ ciphertext. The info parameter is bound as additional data for domain separation (e.g. "effect" vs "tip-recovery"): a blob sealed under one domain does not open under another.

type Engine 

type Engine struct {

	// Notification callbacks — fired after effects are durable
	OnKeyDataAdded func(key string) // wake oldest waiter (data inserted)
	OnKeyDeleted   func(key string) // wake all waiters (key removed)
	OnFlushAll     func()           // wake all waiters across all keys

	OnLocalEffect func(offset Tip, eff *pb.Effect)

	// Ephemeral pub/sub callbacks — fired from HandleRemote on
	// receive of wire-only effects that are never stored or indexed.
	// OnPubSubMessage delivers an inbound PUBLISH to local subscribers.
	OnPubSubMessage func(channel, payload []byte)
	// OnEphemeralSubscribe records / removes a remote peer's interest
	// in a routing key. unsubscribe=false means register, true means
	// drop. The routing key is the raw Effect.Key bytes — encoding is
	// the cluster router's concern, not the engine's.
	OnEphemeralSubscribe func(subscriberNodeID uint64, routingKey []byte, unsubscribe bool)
	// contains filtered or unexported fields
}

Engine is the central coordinator for the causal effect log. Lock-free: the log uses CAS, the index manages its own concurrency, and safety config is swapped atomically.

func NewEngine 

func NewEngine(cfg EngineConfig) *Engine

NewEngine creates a new Engine from the given configuration.

func NewTestEngine 

func NewTestEngine() *Engine

NewTestEngine creates a minimal Engine for use in tests outside this package.

func (*Engine) ArenaBytes 

func (e *Engine) ArenaBytes() int64

ArenaBytes returns the critbit index's slot-array footprint (the trie skeleton), distinct from the vertex pool's effect bytes (EffectCache().Bytes()). Exposed for telemetry so the two memory consumers can be compared directly.

func (*Engine) AverageK 

func (e *Engine) AverageK() float64

AverageK exposes the index's current eviction threshold for telemetry. k now lives on the critbit index (the vertex pool has no eviction policy of its own), so heartbeat stats read it from here rather than the effect cache.

func (*Engine) CachePeerFilter 

func (e *Engine) CachePeerFilter(peer pb.NodeID, data []byte, version uint64)

CachePeerFilter splices a peer's bulk filter into our view of that peer, if it's newer than what we've already applied. Delivered by the cluster layer at connection establishment; the arrival releases the peer from the "presumed to hold everything" default.

func (*Engine) CheckSerializationLeader 

func (e *Engine) CheckSerializationLeader(key string) *pb.NodeID

CheckSerializationLeader returns the serialization leader for a key, or nil if no serialization is active.

func (*Engine) Close 

func (e *Engine) Close() error

Close performs graceful shutdown of the engine and its background components.

func (*Engine) DropPeer 

func (e *Engine) DropPeer(peer pb.NodeID)

DropPeer releases per-peer cached state when a peer permanently leaves the cluster. Wired into PeerManager.SetPeerLifecycleHooks on the onRemoved hook (which fires only on genuine membership removal, not transient unreachability). Subscriber sets are NOT scrubbed here: they live in per-key leafState (reclaimed on leaf eviction) and are filtered against current membership at broadcast time, so a departed peer's id simply falls out of collectSubscribers. The key filter, by contrast, is real per-peer cached data that cannot be lazily re-derived, so it must be dropped explicitly.

func (*Engine) EffectCache 

func (e *Engine) EffectCache() *VertexPool

EffectCache returns the engine's deserialized effect cache for use by the fetch handler (serves effects from cache when the log is unavailable).

func (*Engine) EvictStats 

func (e *Engine) EvictStats() keytrie.EvictStats

EvictStats exposes the index's adaptive-eviction internals for telemetry.

func (*Engine) FlushIndex 

func (e *Engine) FlushIndex()

FlushIndex deletes all keys from the index and evicts all cache entries.

func (*Engine) Forward 

func (e *Engine) Forward(commandName string, args [][]byte, keys []string, username string) []byte

Forward checks if any key has a serialization leader on another node. If so, forwards the command and returns raw RESP bytes. Returns nil if the command should execute locally.

func (*Engine) ForwardExec 

func (e *Engine) ForwardExec(commands []ForwardCommand, watchedKeys []string, username string) []byte

ForwardExec checks if any queued command touches a serialized key. If so, forwards the entire transaction and returns raw RESP bytes. Returns nil for local execution.

func (*Engine) GetLock 

func (e *Engine) GetLock(key string) *sync.Mutex

GetLock returns a striped lock for the given key using FNV-1a hashing.

func (*Engine) GetSnapshot 

func (e *Engine) GetSnapshot(key string) (*pb.ReducedEffect, []Tip, int, error)

GetSnapshot returns the current materialized state of a key and the tip offsets the snapshot was derived from. Callers that perform read-modify-write (SETBIT, INCR, etc.) must pass the returned tips to Emit so the first effect depends on the tips the snapshot was actually computed from, not whatever the index contains at Emit time. The returned ReducedEffect is an immutable snapshot and may be shared by subsequent reads; callers that need to change it must clone it first.

Cache hit returns immediately. On miss, walks the causal DAG from the index tip set and reconstructs via ReduceBranch + canonical merge.

func (*Engine) HandleNack 

func (e *Engine) HandleNack(nack *pb.NackNotify) error

HandleNack processes a NACK from a remote peer.

func (*Engine) HandleRemote 

func (e *Engine) HandleRemote(notify *pb.OffsetNotify) ([]*pb.NackNotify, error)

HandleRemote processes a remote effect notification: stores the effect in the log, updates the index, and returns NACKs if deps don't match tips.

EffectData may be in wire format [4-byte LE keyLen][key][protoData] or raw proto bytes. Both are handled transparently.

Returns all NACKs generated (one per diverged key) so the caller can send them synchronously as the ReplicateTo response.

func (*Engine) InstallCloudTips 

func (e *Engine) InstallCloudTips(key string, tips []Tip, sidecar []CloudEffect, missing []Tip) (pending []Tip, err error)

InstallCloudTips classifies Cloud-provided marker candidates into the subset that is locally walkable (install) and the subset still holed (pending), merges the walkable subset into the index alongside whatever is already there, and returns the holed subset for the caller to track. Besides the read-miss rehydrate it serves the cloud-sync reconcile path: a read served from the outbox during a Cloud outage re-installs the missed Cloud frontier here once Cloud answers again.

It never mints and never consumes local tips: Cloud's tips directory is a candidate index, not authoritative state, so nothing here may supersede local ancestry. A hole is not this engine's to repair — see ErrCDNBlobMissing.

func (*Engine) KeyCount 

func (e *Engine) KeyCount() int64

KeyCount returns the number of keys in the index.

func (*Engine) MatchKeys 

func (e *Engine) MatchKeys(pattern string) []string

MatchKeys returns all keys matching the glob pattern. Uses a point-in-time snapshot for consistency.

func (*Engine) MemoryTarget 

func (e *Engine) MemoryTarget() int64

MemoryTarget returns the byte budget the governor holds the vertex pool under (0 = unbounded, no --maxmemory configured). The pool is byte-bounded rather than slot-bounded, so this is its capacity for heartbeat telemetry.

func (*Engine) NewContext 

func (e *Engine) NewContext() *Context

NewContext creates a new write context bound to the engine.

func (*Engine) NewReadOnlyContext 

func (e *Engine) NewReadOnlyContext() *Context

NewReadOnlyContext creates a context for read-only, non-transactional commands. It uses the per-node cluster key filters to answer a read-miss without issuing a subscription — a key no peer holds returns nil for free. It must not be used inside MULTI/EXEC (SSI reads must subscribe).

func (*Engine) NodeID 

func (e *Engine) NodeID() pb.NodeID

func (*Engine) NotePeerConnected 

func (e *Engine) NotePeerConnected(peer pb.NodeID)

NotePeerConnected marks peer as a live cluster member whose key filter we may not have yet. Until its bulk filter arrives, the peer is presumed to hold every key (clusterMaybeHasKey answers true), so reads bootstrap for real instead of fabricating misses. Called by the cluster layer when a peer connection is registered.

func (*Engine) OwnKeyFilterSnapshot 

func (e *Engine) OwnKeyFilterSnapshot() ([]byte, uint64)

OwnKeyFilterSnapshot returns the serialized own filter and its version, re-marshaling only when the filter changed since the last call. The cluster layer sends this to a peer at connection establishment.

func (*Engine) PeerSubscribers 

func (e *Engine) PeerSubscribers(key string) []pb.NodeID

PeerSubscribers returns the current subscriber set for key. The returned slice is a snapshot; callers may iterate without holding any lock. It is unfiltered by membership — flushTx's collectSubscribers applies the current-membership filter at broadcast time.

func (*Engine) PinKey 

func (e *Engine) PinKey(key string) bool

PinKey adds a dynamic do-not-evict hold on key; UnpinKey releases it. The cloud outbox is the caller: a key with un-acked uploads must stay findable — evicting it frees almost nothing (the outbox pins the effect bytes) while unsubscribing makes the committed data invisible to the whole cluster until the ack lands. Reports whether a live leaf carried the operation.

func (*Engine) ReconvergeAllKeys 

func (e *Engine) ReconvergeAllKeys()

ReconvergeAllKeys triggers a debounced background anti-entropy pass to fetch any effects missed during the partition. Multiple calls within the debounce window are coalesced into a single pass. This replaces the previous approach of deleting all subscriptions, which caused a thundering herd of blocking re-bootstraps on the read path and cascading peer timeouts.

func (*Engine) RecordSerializationActivity 

func (e *Engine) RecordSerializationActivity(key string)

RecordSerializationActivity updates the last activity timestamp for a serialized key. Called when the leader processes a forwarded command or performs a local write on a serialized key.

func (*Engine) ReleaseQueueDepth 

func (e *Engine) ReleaseQueueDepth() int

ReleaseQueueDepth returns the number of cold-evicted keys whose deferred ref-release walk has not yet run. Each pending entry pins its key's cached subdag and resident effect chain, so a persistently deep queue means the governor's reclaim pass is falling behind eviction and live heap will read high for memory eviction can't actually free yet.

func (*Engine) ScanKeys 

func (e *Engine) ScanKeys(after string, pattern string, fn func(key string) bool)

ScanKeys iterates over keys matching the glob pattern, starting after `after`. Pass empty string for `after` to start from the beginning. Return false from fn to stop iteration.

func (*Engine) SetBroadcaster 

func (e *Engine) SetBroadcaster(b Broadcaster)

SetBroadcaster sets the broadcaster for replicating effects to peers. Must be called before any Emit/Flush if cluster mode is desired.

Also lazy-inits the horizon set if not already present: NewEngine gates horizon on cfg.Broadcaster != nil, but the bootstrap path in beacon/runtime.go constructs the engine first (with a nil Broadcaster) and wires the PeerManager as broadcaster afterwards (chicken-and-egg: PeerManager needs engine to build the effect handler). Without this lazy init, horizon stays nil for the life of the engine and bind-arrival visibility isn't deferred — peers see aborting-txn effects before fork-choice has settled, which surfaces as Elle :incompatible-order on cross-node reads.

func (*Engine) SetCloudReader 

func (e *Engine) SetCloudReader(r CloudReader)

SetCloudReader installs the Cloud read backstop consulted on a read-miss. Nil leaves the engine free-missing as before (standalone / no cloud).

func (*Engine) SetOnLocalEffect 

func (e *Engine) SetOnLocalEffect(hook func(offset Tip, eff *pb.Effect))

SetOnLocalEffect replaces the local-mint hook. It waits for callbacks already in flight, which gives lifecycle owners a clean boundary before draining or tearing down the hook's resources.

func (*Engine) SetRTTProvider 

func (e *Engine) SetRTTProvider(p PeerRTTProvider)

SetRTTProvider sets the RTT provider for adaptive serialization leader selection.

func (*Engine) StartAntiEntropy 

func (e *Engine) StartAntiEntropy(interval time.Duration)

StartAntiEntropy launches a background goroutine that periodically exchanges tips with peers and fetches missing effect chains. This ensures effects missed during partitions are eventually discovered without polluting the log with redundant subscription effects. Also starts the reconvergence debounce loop that coalesces peer recovery events into background anti-entropy passes.

func (*Engine) UnpinKey 

func (e *Engine) UnpinKey(key string) bool

func (*Engine) UpdateSafetyRules 

func (e *Engine) UpdateSafetyRules(defaultMode SafetyMode, rules []KeyRangeRule)

UpdateSafetyRules atomically replaces the key-range safety configuration.

func (*Engine) VertexCount 

func (e *Engine) VertexCount() int

VertexCount returns the number of effects resident in the vertex pool.

type EngineConfig 

type EngineConfig struct {
	NodeID        pb.NodeID
	Broadcaster   Broadcaster     // nil for standalone
	RTTProvider   PeerRTTProvider // nil disables RTT-based leader selection
	DefaultMode   SafetyMode
	KeyRangeRules []KeyRangeRule
	MemoryLimit   int64 // memory budget for effect cache (0 = 10MB default)
	// MemoryLimitPercent, when non-zero, expresses the cache budget as a
	// fraction (0,1] of the memory available to this process. Takes
	// precedence over MemoryLimit. The cache enforces this via a live
	// re-evaluating tick so cgroup/system changes propagate.
	MemoryLimitPercent float64
	// CompressValues makes this node store data values zstd-compressed inside
	// the effects it emits, trading decompression on read for more resident
	// DAG per byte. Write-side only: the per-effect compression flag drives
	// reading, so nodes with different settings share one DAG.
	CompressValues bool
}

EngineConfig holds configuration for creating an Engine.

func (*EngineConfig) ModeForKey 

func (c *EngineConfig) ModeForKey(key string) SafetyMode

ModeForKey returns the safety mode for a key. First matching rule wins.

type FetchHint 

type FetchHint uint8

FetchHint orders the two effect-byte sources for FetchFromAny. The engine derives it from knowledge it already holds: the per-peer key filters claim the key → PreferPeers; otherwise the cluster provably lacks it → PreferCDN. The unpreferred source is a failure fallback, never raced. 

const (
	// PreferPeers tries connected peers first, cloud CDN on total peer
	// failure. The default for peer-origin fetches (NACK ingest, backfill)
	// and any fetch without key context.
	PreferPeers FetchHint = iota
	// PreferCDN tries the cloud CDN first, peers on failure. Chosen when no
	// peer's key filter claims the key — a cloud rehydrate of state the
	// cluster no longer holds.
	PreferCDN
)

type ForwardCommand 

type ForwardCommand struct {
	Name string
	Args [][]byte
	Keys []string
}

ForwardCommand describes a single command to be forwarded.

type HorizonSet 

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

HorizonSet tracks Binds in their horizon wait period. Effects from invisible Binds are excluded from reconstruction until the horizon completes.

func (*HorizonSet) Abort 

func (h *HorizonSet) Abort(txnID string)

Abort removes a single entry from the horizon set without promoting it. Used by flushTx when the originator decides to abort after NACK processing. The bind effect is still in the local DAG and at peers; reconstruct's cross-key reachability is what skips it on read.

Other entries in the same group are unaffected — they continue waiting on their own visibility trigger (timer or explicit MakeVisible).

func (*HorizonSet) Add 

func (h *HorizonSet) Add(txnID string, bindOffset Tip, bind *pb.TransactionalBindEffect)

Add registers a Bind in the invisible set. The bind stays invisible until MakeVisible, Abort, or a timer scheduled via ScheduleMakeVisible fires. If the bind's keys overlap with an existing group's consumed tips, it joins that group; otherwise a new group is created.

func (*HorizonSet) Empty 

func (h *HorizonSet) Empty() bool

Empty reports whether no txn is currently held invisible. When true, reconstruct can skip its per-read invisibility precompute: any bind reachable from a read's captured tips had its horizon entry added during ingest (before the index update that made it visible to the read), so an empty horizon at reconstruct setup proves no walked bind is invisible.

func (*HorizonSet) IsInvisible 

func (h *HorizonSet) IsInvisible(txnID string) bool

IsInvisible returns true if the given txnID is in the invisible set.

func (*HorizonSet) MakeVisible 

func (h *HorizonSet) MakeVisible(txnID string)

MakeVisible promotes an entire group: removes all txnIDs from the invisible set, cleans up pendingTxTips, evicts cache, and fires OnKeyDataAdded callbacks.

func (*HorizonSet) ScheduleMakeVisible 

func (h *HorizonSet) ScheduleMakeVisible(txnID string, wait time.Duration)

ScheduleMakeVisible schedules MakeVisible(txnID) to fire after `wait`. Used by handleRemoteBind as a crash-fallback for the remote-arrival wait — the primary signal is the originator's verdict-snapshot arriving via applySnapshotVerdicts, which short-circuits this timer. The timer only fires when the originator died between broadcasting the bind and broadcasting the verdict; we keep it generous (~5s) to avoid racing the snapshot under normal conditions. Resets any existing timer on the group; a later-joining bind extends the wait.

func (*HorizonSet) WaitForClear 

func (h *HorizonSet) WaitForClear(txnID string) *HorizonWait

WaitForClear blocks until the bind for txnID has been resolved (MakeVisible or Abort releases the entry's waiter lock). Returns nil if the bind is not in the invisible set — the caller may proceed without waiting. Returns a *HorizonWait the caller must Release when done.

type HorizonWait 

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

HorizonWait is a handle returned by WaitForClear. It carries the horizonEntry pointer and the RToken so Release can RUnlock the same mutex even after the entry has been removed from h.entries by MakeVisible or Abort.

func (*HorizonWait) Release 

func (w *HorizonWait) Release()

Release returns the read-lock token to the horizonEntry. Safe on nil.

type KeyRangeRule 

type KeyRangeRule struct {
	Pattern string
	Mode    SafetyMode
}

KeyRangeRule maps a key pattern to a safety mode.

type MemoryStorage 

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

MemoryStorage is an in-memory ObjectStorage implementation for testing.

func NewMemoryStorage 

func NewMemoryStorage() *MemoryStorage

NewMemoryStorage creates a new MemoryStorage.

func (*MemoryStorage) Download 

func (m *MemoryStorage) Download(_ context.Context, path string) ([]byte, error)

func (*MemoryStorage) Keys 

func (m *MemoryStorage) Keys() []string

Keys returns all stored paths (for testing).

func (*MemoryStorage) Upload 

func (m *MemoryStorage) Upload(_ context.Context, path string, data []byte) error

type ObjectStorage 

type ObjectStorage interface {
	Upload(ctx context.Context, path string, data []byte) error
	Download(ctx context.Context, path string) ([]byte, error)
}

ObjectStorage abstracts uploading/downloading blobs to/from object storage.

type PeerRTTProvider 

type PeerRTTProvider interface {
	// GetRTT returns the estimated round-trip time to the given peer.
	// Returns 0 if the peer is unknown or RTT has not been measured.
	GetRTT(nodeID pb.NodeID) time.Duration
	// AlivePeerIDs returns the IDs of all peers that are currently alive.
	AlivePeerIDs() []pb.NodeID
}

PeerRTTProvider provides RTT measurements to peers for optimal leader selection.

type SafetyMode 

type SafetyMode int

SafetyMode determines write behavior during network partitions. 

const (
	SafeMode   SafetyMode = iota // blocks writes when quorum is unreachable
	UnsafeMode                   // allows writes during partitions, forming branches
)

type Tip 

type Tip = keytrie.EffectRef

func NormalizeCloudTipCandidates 

func NormalizeCloudTipCandidates(key string, candidates, localRoots []Tip, load func(Tip) (*pb.Effect, error)) (real, superseded []Tip, err error)

NormalizeCloudTipCandidates reduces Cloud's marker candidates to the maximal tips of the effect DAG. Cloud's tips directory is an index: stale markers can remain after a newer effect has dep-referenced them, so callers must not treat every returned marker as an independent logical frontier branch. localRoots seeds the same walk with this node's own current tips for key, so a candidate already covered by local ancestry the cluster itself produced — not yet reflected in Cloud's index — is dropped as superseded too. Only candidates, never localRoots, can be marked superseded: local tips aren't entries in Cloud's index and have nothing there to correct.

Missing blobs do not abort normalization. A missing candidate can still be proven non-maximal when another candidate's or local root's readable ancestry names it. Any missing candidate that cannot be so proven remains a real tip and the caller's classification decides what to do with it.

type VertexPool 

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

VertexPool is the engine's deserialized effect store: a map from Tip to its effect, replacing the per-offset CloxCache. Presence in the pool means "resident locally"; absence means getEffect must fetch the bytes from a peer. The pool is dumb storage — it has no eviction policy of its own. Eviction is driven by the engine memory governor, which is the only component that knows a key's active DAG path and can therefore choose safe victims.

The pool does NOT meter its own footprint: a protoLen-based sum undercounted each decoded effect graph ~3x and was blind to the off-pool per-key state (subdags, reduced memos, subscriptions) that grows alongside it. Bytes() reports the runtime's exact live heap instead — see Bytes.

DAG-reference counting lets eviction drop whole keys and reclaim only the vertices no other key or read still needs; a vertex is referenced by every tipset that reaches it and every cached subdag that adopted it.

func (*VertexPool) Bytes 

func (p *VertexPool) Bytes() int64

Bytes returns the process's live heap as marked by the last GC (/gc/heap/live:bytes). This is the exact, total memory signal the governor triggers and sizes eviction on, and it counts everything that grows per key — decoded effect graphs in the pool, off-pool subdags/reduced memos/subscription state, the trie — not just the pool's slice. A protoLen-based pool sum undercounted each decoded graph ~3x and was blind to the off-pool half, so triggering on it let the heap reach 10GB while the pool read 3GB.

It deliberately uses /gc/heap/live (live objects only) rather than HeapAlloc (/memory/classes/heap/objects), which is live + not-yet-collected garbage and sawtooths up to the GC goal (~2x live at GOGC=100). Triggering on HeapAlloc fired eviction on the garbage peaks while the true working set was under target. /gc/heap/live updates once per GC cycle — stable between cycles and, under a churning workload, fresh well within the governor's 1s tick. metrics.Read is cheap and non-STW.

func (*VertexPool) ColdEvictions 

func (p *VertexPool) ColdEvictions() uint64

ColdEvictions returns the number of keys the index's bounded sweep evicted under memory pressure — the real "evicted_keys", distinct from reclaim churn.

func (*VertexPool) Decref 

func (p *VertexPool) Decref(tip Tip)

Decref releases a DAG reference on the vertex at tip.

func (*VertexPool) EntryCount 

func (p *VertexPool) EntryCount() int

EntryCount returns the number of resident effects.

func (*VertexPool) Get 

func (p *VertexPool) Get(tip Tip) (*pb.Effect, bool)

Get returns the effect at tip.

func (*VertexPool) Incref 

func (p *VertexPool) Incref(tip Tip) bool

Incref adds a DAG reference to the vertex at tip and reports whether the reference was taken. See incref for the failure contract.

func (*VertexPool) Put 

func (p *VertexPool) Put(tip Tip, eff *pb.Effect)

Put stores eff at tip, computing its accounted size from the proto. Callers that already hold the serialized length — a marshalled buffer on emit, the wire protoData on ingest — should use PutSized to skip the proto.Size walk.

A re-Put of an already-resident tip keeps the existing vertex so its accumulated refcount survives: effects are immutable (a Tip is written once), so the payload and byte cost are identical. Replacing the vertex would reset refs to 0, letting reclaimUnreferenced free an effect the index still holds as a frontier tip — a premature free surfacing as a missing read. The resident fast-path also skips the proto.Size walk on the common re-delivery.

func (*VertexPool) PutSized 

func (p *VertexPool) PutSized(tip Tip, eff *pb.Effect, protoLen int)

PutSized stores eff at tip using a caller-supplied serialized length, avoiding the proto.Size walk Put would do. protoLen is the marshalled effect size (the length of MarshalEffect's output, or the wire protoData); vertexOverhead is added on top. Re-Put semantics match Put: an already-resident tip is kept.

A new vertex is born with its creation refcount equal to the number of tipsets it occupies — its "walkable from a tipset" references, one per reaching tip. An ordinary effect is the tip of its single key, so it starts at 1; a bind commits a whole keyset and is the tip of each key, so it starts at len(Keys). This pins the vertex (interior chain nodes included, local or backfilled from a peer) until each occupying key is evicted, where the eviction reachable-walk decrefs it once per key (releaseChainRefs). The count is stored before the vertex is published via LoadOrStore, so reclaimUnreferenced never observes a fresh vertex at 0 (a birth-race free). Read adoption layers further refs on top in publishSubdag.

If the tip is already resident as a cache entry (PutSizedCache left it at refs==0 because we did not yet serve its key), this owned Put promotes it by adding the creation ref it lacked, so reclaim won't drop a vertex we now own. The CAS makes that race-safe: it loses to reclaim's 0→tombstone claim (the vertex is being freed; we refetch on the next miss) and to a concurrent read's publishSubdag incref (refs>0; left un-promoted, resolved by a later refetch), and it never double-counts an already-owned re-put (refs>0 → CAS fails → keep the accumulated count).

func (*VertexPool) PutSizedCache 

func (p *VertexPool) PutSizedCache(tip Tip, eff *pb.Effect, protoLen int)

PutSizedCache stores a fetched effect on a key we do NOT serve as a reclaimable cache entry: refs start at 0, so it carries no creation ref and reclaim frees it the moment no subdag adopts it. These are cross-key bind adjudication fetches (and remote effects on unsubscribed keys): a read that walks them holds them via the reconstruct dag for its duration, and they are refetchable from a peer, so they must not pin memory the way an owned effect's creation ref does. An already-resident owned tip is kept (LoadOrStore), never downgraded. protoLen semantics match PutSized.

A freshly-stored entry is born at refs==0 and will never see a refs→0 transition unless first adopted (publishSubdag) and then dropped, so it is enqueued as a reclaim candidate at birth — without this an un-adopted cache fetch would never be reclaimed (there is no full-map scan to catch it).

func (*VertexPool) Reclaimed 

func (p *VertexPool) Reclaimed() uint64

Reclaimed returns the number of vertices freed by reclaimUnreferenced (below-LCA history and orphans) — storage churn, not cache eviction.

func (*VertexPool) Stats 

func (p *VertexPool) Stats() (hits, misses, reclaimed uint64)

Stats returns cumulative hit, miss, and reclaim counts. The third value is reclaimUnreferenced churn (vertices freed), not cache eviction — use ColdEvictions for keys dropped under memory pressure.

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.