graphcache

package
v0.14.0 Latest Latest
Warning

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

 Go to latest Published: Jun 24, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Overview

Package graphcache provides Lantern's in-memory graph cache: a TTL-backed vertex store plus additive, expiring directed edges.

GraphCache is the aggregate consistency boundary. Public write methods take GraphCache.mu before mutating the vertex cache, edge cache, dictionary, replication watermarks, tombstones, or secondary indexes. The outer rule is: 

GraphCache.mu is acquired before any aggregate member is made inconsistent.

The inner structures also synchronize themselves because a few hot paths read or compact per-edge weights without taking a write lock on the whole aggregate. Low-level types must not call back into GraphCache while holding their own locks. User callbacks have narrower contracts: ScanByPrefix snapshots under GraphCache.mu and invokes the visitor after unlocking, while ScanEdgesByPrefix invokes the visitor while holding GraphCache.mu.RLock.

Mutation lifecycle, at the aggregate level:

  • vertex writes intern the key, update the vertex cache, then update enabled secondary indexes;
  • vertex eviction through Delete, Clear, or Flush releases dictionary references and drops prefix/search postings through the vertex cache's eviction hook;
  • edge writes auto-create endpoint vertices, mutate edge buckets, then update the per-tail head index when a bucket is created or deleted;
  • replicated writes additionally pass through HLC/tombstone guards before mutating storage;
  • GC first flushes expired vertices, then sweeps tombstones and stale HLC watermarks, then removes zero-weight and dangling edges.

The package deliberately keeps all index maintenance synchronous. There is no background indexer or asynchronous event bus: a public write returns only after every observable in-memory view has been updated.

Index

Constants

View Source 
const (
	// DefaultPPRAlpha is the restart/teleport-to-seed probability used when a
	// caller passes a non-positive (or out-of-range) alpha. 0.15 is the
	// canonical PageRank damping complement — a broad, well-behaved locality.
	DefaultPPRAlpha = 0.15
	// DefaultPPREpsilon is the forward-push residual threshold used when a
	// caller passes a non-positive epsilon. 1e-4 keeps the push budget small
	// (O(1/(alpha*eps))) while staying accurate enough for ranking.
	DefaultPPREpsilon = 1e-4
)

PPR (Personalized PageRank) defaults and guards (#801).

Variables

This section is empty.

Functions

This section is empty.

Types

type ContribID 

type ContribID [24]byte

ContribID is a globally-unique identifier for an additive edge contribution. It packs the 16-byte HLC NodeID of the originating writer (offset 0..15) and a monotonic 8-byte big-endian sequence assigned by that writer's mutation log (offset 16..23).

The all-zero value is reserved as "no contribution identity" and is the value used by every local (non-replicated) write. Code that performs dedup MUST skip the dedup check when the incoming ContribID is zero — otherwise two distinct local Add calls would collapse into one.

ContribID exists so the replication apply path (issue #182) can ignore a Mutation it has already seen — at-least-once delivery from a peer must not double-count additive edge weight on re-delivery.

func (ContribID) IsZero 

func (c ContribID) IsZero() bool

IsZero reports whether c is the zero ContribID. Zero means "no identity" and disables dedup on the write path.

type EdgeItem 

type EdgeItem[S comparable] struct {
	Tail       S
	Head       S
	Weight     float32
	Expiration time.Time
	// ContribID is an optional dedup key for additive AddEdges* writes.
	// A non-zero id makes the contribution idempotent: re-applying an item
	// with the same id leaves the stored weight unchanged (see
	// AddEdgesWithExpirationContrib). The zero value disables dedup and keeps
	// the legacy additive semantics. PutEdgesWithExpiration ignores this
	// field — Put is already idempotent.
	ContribID ContribID
}

EdgeItem is a single (tail, head, weight, expiration) tuple supplied to AddEdgesWithExpiration / PutEdgesWithExpiration.

type EdgeKey 

type EdgeKey[S comparable] struct {
	Tail S
	Head S
}

EdgeKey identifies a directed edge for batch deletion via DeleteEdges.

type EdgeWeighting added in v0.14.0 

type EdgeWeighting uint8

EdgeWeighting selects the transform applied to a raw edge weight BEFORE the directional per-hop top/bottom-k prune (#410, #800). It replaces the historical tfidf bool threaded through the Neighbor* surface so a third strategy — and any future one — is a single enum arm rather than another boolean parameter. 

const (
	// WeightingRaw uses the live additive edge weight verbatim.
	WeightingRaw EdgeWeighting = iota
	// WeightingTFIDF applies the crude hub-suppressor w / log2(1 + df(head)).
	// It is cheap and O(1) per edge but corpus-size-blind; kept alongside
	// BM25 for its distinct, faster semantics.
	WeightingTFIDF
	// WeightingBM25 re-scores with Okapi BM25 over the per-vertex out-edge
	// distribution: TF = the live edge weight, DF = df(head) (distinct tails
	// into head), N = number of tails, DocLen = the tail's out-degree, and
	// AvgLen = mean out-degree. It runs through the shared search.BM25Score
	// kernel so graph edge weighting stays numerically identical to full-text
	// SearchVertices ranking, adding TF saturation, document-length
	// normalization, and a real N-aware IDF that TFIDF lacks.
	WeightingBM25
)

type GraphCache 

type GraphCache[S comparable, T any] struct {
	// contains filtered or unexported fields
}

func NewGraphCache 

func NewGraphCache[S comparable, T any](defaultTTL time.Duration) *GraphCache[S, T]

func (*GraphCache[S, T]) AddEdge 

func (c *GraphCache[S, T]) AddEdge(tail, head S, w float32)

func (*GraphCache[S, T]) AddEdgeWithExpiration 

func (c *GraphCache[S, T]) AddEdgeWithExpiration(tail, head S, w float32, expiration time.Time)

func (*GraphCache[S, T]) AddEdgeWithExpirationContrib 

func (c *GraphCache[S, T]) AddEdgeWithExpirationContrib(tail, head S, w float32, expiration time.Time, contribID ContribID) bool

AddEdgeWithExpirationContrib is the dedup-aware additive write used by the replication apply path (#182). A non-zero contribID makes the call idempotent: re-applying the same mutation (e.g. on peer reconnect or duplicate stream delivery) leaves the stored weight unchanged. A zero contribID falls through to ordinary additive semantics, which is the local non-replicated path. Returns applied=true when the contribution was recorded; false means dedup suppressed an already-stored contribution with the same ID.

func (*GraphCache[S, T]) AddEdgeWithExpirationContribHLC 

func (c *GraphCache[S, T]) AddEdgeWithExpirationContribHLC(tail, head S, w float32, expiration time.Time, contribID ContribID, ts hlc.Timestamp) bool

AddEdgeWithExpirationContribHLC is the tombstone-aware sibling of AddEdgeWithExpirationContrib used by the replication apply path (#183). When a live edge tombstone exists for (tail, head) whose HLC is strictly newer than ts the contribution is dropped and false is returned, preventing a late Add* from resurrecting a freshly-deleted edge. Otherwise behaviour matches AddEdgeWithExpirationContrib, including ContribID-based dedup. A successful apply clears any existing tombstone for the edge.

func (*GraphCache[S, T]) AddEdgeWithTTL 

func (c *GraphCache[S, T]) AddEdgeWithTTL(tail, head S, w float32, ttl time.Duration)

func (*GraphCache[S, T]) AddEdgesWithExpiration 

func (c *GraphCache[S, T]) AddEdgesWithExpiration(items []EdgeItem[S])

AddEdgesWithExpiration additively writes every supplied edge under a single write lock, auto-creating endpoint vertices on demand (matching the per-edge AddEdgeWithExpiration invariant). Concurrent readers see either the pre-batch or the post-batch state.

Each item's ContribID is honored: a non-zero id deduplicates the contribution (see AddEdgesWithExpirationContrib). Leaving ContribID at its zero value — the default — keeps the legacy non-idempotent additive path.

func (*GraphCache[S, T]) AddEdgesWithExpirationContrib 

func (c *GraphCache[S, T]) AddEdgesWithExpirationContrib(items []EdgeItem[S]) (deduped int)

AddEdgesWithExpirationContrib is the dedup-aware batch sibling of AddEdgesWithExpiration. It applies the whole batch under a single write lock and returns the number of items that were deduplicated — i.e. whose non-zero ContribID matched a live contribution already stored on the (tail, head) edge, so no weight was added. Items with a zero ContribID always apply (legacy additive semantics) and never count as deduped.

The dedup guarantee mirrors the per-edge AddEdgeWithExpirationContrib used by the replication apply path (#182): replaying a batch with the same ContribIDs leaves the stored weights unchanged. This is what lets a client-supplied idempotency key make AddEdge(s) safe to retry (#588).

func (*GraphCache[S, T]) AddEdgesWithExpirationContribHLC added in v0.9.0 

func (c *GraphCache[S, T]) AddEdgesWithExpirationContribHLC(items []EdgeItem[S], ts hlc.Timestamp) (deduped int)

AddEdgesWithExpirationContribHLC is the tombstone-aware, single-lock batch sibling of AddEdgesWithExpirationContrib used by the LOCAL write path when replication is enabled. Additive merge already converges via ContribID set semantics regardless of delivery order, so the ts is NOT compared against a per-edge write watermark; it is consulted ONLY against the edge tombstone so a contribution whose ts is strictly older than a delete is dropped on the origin exactly as it is on every peer. Items whose ts loses to the tombstone are skipped and counted as deduped=false (they applied nothing); items with a matching live ContribID are deduped as in the non-HLC variant. Returns the number of items that added no weight (tombstone-dropped or ContribID-deduped).

func (*GraphCache[S, T]) CountByPrefix 

func (c *GraphCache[S, T]) CountByPrefix(prefix string) int

CountByPrefix returns the number of LIVE vertices whose projected key starts with prefix. It walks the prefix index and counts only keys the vertex cache still resolves as live, so an entry that has expired but not yet been flushed is excluded — the count therefore equals len(ScanByPrefix(prefix)) and the number of live primary vertices matching prefix, independent of GC timing (#752). Returns 0 when the index has not been enabled.

func (*GraphCache[S, T]) DeleteByPrefix 

func (c *GraphCache[S, T]) DeleteByPrefix(ctx context.Context, prefix string, limit int) int

DeleteByPrefix removes every vertex whose projected key starts with prefix and returns the number of vertices deleted. limit caps the number of deletions in a single call; pass 0 (or a negative value) to delete the entire matching set. When the index has not been enabled, DeleteByPrefix returns 0 without touching the cache.

Deletion routes the whole matched set through a single vertices.DeleteMany, so the OnEvict / refcount / radix-removal chain runs once for the batch rather than once per key (#738). Edges incident to a deleted vertex are NOT removed eagerly here — they will be reclaimed by the next dangling-edge flush, matching the existing DeleteVertex contract.

func (*GraphCache[S, T]) DeleteByPrefixHLC 

func (c *GraphCache[S, T]) DeleteByPrefixHLC(ctx context.Context, prefix string, limit uint32, ts hlc.Timestamp, expiration time.Time) (int, error)

DeleteByPrefixHLC is the tombstone-aware sibling of DeleteByPrefix: every vertex whose projected key matches prefix receives a tombstone stamped at ts/expiration. Returns the number of vertices actually deleted (matching DeleteByPrefix). limit==0 means unlimited.

func (*GraphCache[S, T]) DeleteEdge 

func (c *GraphCache[S, T]) DeleteEdge(tail, head S) bool

DeleteEdge removes the (tail, head) edge and returns whether it was present.

func (*GraphCache[S, T]) DeleteEdgeHLC 

func (c *GraphCache[S, T]) DeleteEdgeHLC(tail, head S, ts hlc.Timestamp, expiration time.Time) bool

DeleteEdgeHLC removes the (tail, head) edge and stamps a tombstone. See DeleteVertexHLC for the late-replay rationale.

func (*GraphCache[S, T]) DeleteEdges 

func (c *GraphCache[S, T]) DeleteEdges(keys []EdgeKey[S]) int

DeleteEdges removes every supplied edge under a single write lock and returns the count of edges that were actually present. Concurrent readers observe either the pre-batch or the post-batch state.

func (*GraphCache[S, T]) DeleteEdgesHLC 

func (c *GraphCache[S, T]) DeleteEdgesHLC(keys []EdgeKey[S], ts hlc.Timestamp, expiration time.Time) int

DeleteEdgesHLC is the batch sibling of DeleteEdgeHLC.

func (*GraphCache[S, T]) DeleteVertex 

func (c *GraphCache[S, T]) DeleteVertex(key S) bool

DeleteVertex removes the vertex (by key) and returns whether it was present.

func (*GraphCache[S, T]) DeleteVertexHLC 

func (c *GraphCache[S, T]) DeleteVertexHLC(key S, ts hlc.Timestamp, expiration time.Time) bool

DeleteVertexHLC removes the vertex and stamps a tombstone so a late Put*/Add* with an HLC strictly older than ts is rejected for the duration of expiration. Returns whether the vertex was present at call time. Edges of the vertex are NOT tombstoned by this call — they are still GC'd via the dangling sweep when their endpoints disappear, but a late AddEdge that re-creates an endpoint can still reappear. Callers that need full subgraph fencing should pair this with DeleteEdgesHLC for the relevant (tail, head) pairs.

func (*GraphCache[S, T]) DeleteVertices 

func (c *GraphCache[S, T]) DeleteVertices(keys []S) int

DeleteVertices removes every supplied vertex under a single write lock and returns the count of keys that were actually present (and therefore deleted). Concurrent readers observe either the pre-batch or the post-batch state. Vertex-owned side indexes (dict refs, prefix radix, search postings) are cleaned in one pass via the batch eviction hook so a large delete pays one acquisition per index instead of one per key (#738).

func (*GraphCache[S, T]) DeleteVerticesHLC 

func (c *GraphCache[S, T]) DeleteVerticesHLC(keys []S, ts hlc.Timestamp, expiration time.Time) int

DeleteVerticesHLC is the batch sibling of DeleteVertexHLC. Every key receives a tombstone, including ones not present in the live cache — this is intentional so a Delete-before-Add race is still resolved by LWW once the (out-of-order) Add arrives.

func (*GraphCache[S, T]) EdgeCount 

func (c *GraphCache[S, T]) EdgeCount() int

EdgeCount returns the live (tail, head) edge count under an RLock. Like VertexCount it is intended for periodic gauge sampling, not hot paths.

func (*GraphCache[S, T]) EnablePrefixIndex 

func (c *GraphCache[S, T]) EnablePrefixIndex(extract func(S) string)

EnablePrefixIndex turns on the optional prefix index, projecting each key S through extract to obtain the string used by ScanByPrefix / CountByPrefix / DeleteByPrefix. It must be called before any vertex is stored; calling it on a non-empty cache panics so the caller cannot silently observe an index that disagrees with point reads. extract must not be nil.

EnablePrefixIndex is intentionally separate from the constructor:

  • GraphCache is generic over S comparable, but prefix semantics are well-defined only for string-like keys. Lifting the constraint into the signature would force every caller (including those that never want prefix scans) to thread a projection through;
  • opt-in keeps the put / evict hot paths free of any radix work for the existing string-keyed deployments that have not migrated.

func (*GraphCache[S, T]) EnableSearchIndex 

func (c *GraphCache[S, T]) EnableSearchIndex(extract func(T) search.Document)

EnableSearchIndex turns on the optional content-search index, projecting each stored value T into the search.Document that gets indexed (for example the vertex's string value). Like EnablePrefixIndex it must be called before any vertex is stored, is idempotent, and panics on a non-empty cache so the caller cannot silently observe an index that disagrees with point reads. extract must not be nil.

Once enabled, the index is kept in perfect lockstep with the vertex lifecycle: every put (including overwrites) re-indexes the value, and eviction — Delete, DeleteMany, Clear, and TTL Flush — drops the key through the shared SetOnEvictMany hook, so entries decay together with the vertices they describe. The index is a third opt-in secondary structure alongside the prefix index; when it is left disabled the put / evict hot paths pay only a nil check.

func (*GraphCache[S, T]) GCSweepBacklog added in v0.11.0 

func (c *GraphCache[S, T]) GCSweepBacklog() int

GCSweepBacklog reports how many tail buckets remain in the current incremental sweep cycle (#744) — those not yet visited by flush(). It is 0 when the incremental sweep is disabled or a cycle has just completed, and lets operators see when bounded cleanup is falling behind the write rate.

func (*GraphCache[S, T]) GetEdgeDetail 

func (c *GraphCache[S, T]) GetEdgeDetail(tail, head S) (float32, time.Time, bool)

GetEdgeDetail returns the current edge weight together with the latest contribution expiration. The expiration is the moment after which the edge is guaranteed to have decayed to zero. When no edge exists, ok is false.

Like GetWeight it is a lock-free point read (see edges.getDetail): it does not take GraphCache.mu and may observe either the pre- or post-write state under a concurrent edge mutation, but it never resolves a recycled endpoint id to an unrelated edge. The edge is hidden unless both endpoint vertices are still live, so a dangling edge to a deleted or expired-but-not-flushed vertex is reported as absent (#750).

func (*GraphCache[S, T]) GetVertex 

func (c *GraphCache[S, T]) GetVertex(key S) (T, bool)

GetVertex returns the live value stored for key. It is a point read that does not take GraphCache.mu: the inner vertex cache has its own RWMutex and hides expired entries, so a point lookup needs no aggregate-lock consistency with the prefix/search/edge indexes. Dropping c.mu here keeps hot reads from serializing behind unrelated writes, long scans, and GC (#740). A read may race a concurrent Put/Delete on the same key and observe either the pre- or post-write state, which is the existing point-read contract.

func (*GraphCache[S, T]) GetWeight 

func (c *GraphCache[S, T]) GetWeight(tail, head S) (float32, bool)

GetWeight returns the additive weight of the (tail, head) edge. Like GetVertex it is a lock-free point read: edges.get pins both endpoint ids (closing the dictionary ABA hazard) and reads the edge map under the edgeCache's own lock, so GraphCache.mu is not required (#740).

The edge is hidden (ok=false) unless both endpoint vertices are still live, so a dangling edge whose tail or head was deleted or expired but not yet swept is reported as absent (#750). The endpoint check reads the vertex cache under its own lock, after the edge read, so GetWeight stays free of GraphCache.mu.

func (*GraphCache[S, T]) Neighbor 

func (c *GraphCache[S, T]) Neighbor(seed S, step int, k int, weighting EdgeWeighting, selectSmallest bool, keep func(S) bool) *graph.Graph[S, T]

Neighbor walks the graph from seed and returns the visited subgraph. The per-hop top-k pruning keeps the k largest-weight edges when selectSmallest is false and the k smallest-weight edges when it is true (#560) — the caller picks the direction that matches its Objective so a cost-minimiser is not handed the costliest edges. weighting re-scores edge weights BEFORE the directional top/bottom-k selection (#800).

keep is an optional frontier predicate (nil = accept all): a candidate head is expanded into the result only when keep(head) is true. It is applied at frontier materialisation, BEFORE scoring and the directional top/bottom-k prune, so top-k selects the k best *accepted* neighbours per hop. The seed is the anchor exemption — it is always retained and never passed through keep. Because the next-hop frontier is derived from the surviving edges, a matching vertex reachable only through a rejected "bridge" is not reached (induced-subgraph semantics). core stays generic: keep is just a predicate over S; the concrete prefix/string logic lives in the caller.

func (*GraphCache[S, T]) NeighborContext 

func (c *GraphCache[S, T]) NeighborContext(ctx context.Context, seed S, step int, k int, weighting EdgeWeighting, selectSmallest bool, keep func(S) bool) (*graph.Graph[S, T], error)

NeighborContext is the context-aware variant of Neighbor. It returns ctx.Err() as soon as the context is cancelled or its deadline has expired — checked between BFS expansion steps — so handlers can short-circuit large traversals when the caller has given up. keep is the optional frontier predicate documented on Neighbor (nil = accept all).

func (*GraphCache[S, T]) NeighborWithExpirationsContext 

func (c *GraphCache[S, T]) NeighborWithExpirationsContext(ctx context.Context, seed S, step int, k int, weighting EdgeWeighting, selectSmallest bool, keep func(S) bool) (*graph.Graph[S, T], map[S]map[S]time.Time, error)

NeighborWithExpirationsContext returns the same subgraph as NeighborContext together with a parallel expirations map keyed by (tail, head). Both are computed under a single RLock so handlers can compose responses without re-acquiring the cache lock per edge.

The expirations map only contains entries for edges that ended up in the returned graph; a missing or zero value means the edge has no known expiration. keep is the optional frontier predicate documented on Neighbor (nil = accept all).

func (*GraphCache[S, T]) PersonalizedPageRank added in v0.14.0 

func (c *GraphCache[S, T]) PersonalizedPageRank(seed S, topN int, alpha, epsilon float64, weighting EdgeWeighting, keep func(S) bool) *graph.Graph[S, T]

PersonalizedPageRank is the context-free convenience wrapper over PersonalizedPageRankContext using context.Background(). See that method for the full contract.

func (*GraphCache[S, T]) PersonalizedPageRankContext added in v0.14.0 

func (c *GraphCache[S, T]) PersonalizedPageRankContext(ctx context.Context, seed S, topN int, alpha, epsilon float64, weighting EdgeWeighting, keep func(S) bool) (*graph.Graph[S, T], error)

PersonalizedPageRankContext ranks vertices by their Personalized PageRank (a.k.a. Random-Walk-with-Restart) relevance to seed and returns a relevance star: a graph whose seed→v edge weight is π[v], the stationary mass a seed-anchored random surfer accumulates on v. It replaces the greedy, per-hop Top(k) walk of Neighbor* with a single global, seed-relative score (#801), so a vertex reached via many paths outranks one reached via a single weak path, and a globally popular hub is discounted relative to this seed.

It is computed by local forward-push (Andersen–Chung–Lang): residual mass is pushed only through vertices whose residual exceeds eps·deg, touching O(1/(alpha·eps)) vertices independent of |V|/|E|. alpha is the restart probability (locality knob: higher = tighter, seed-proximate); epsilon the residual threshold (smaller = more accurate, more work). Non-positive or out-of-range alpha/epsilon fall back to DefaultPPRAlpha / DefaultPPREpsilon.

weighting transforms each transition affinity BEFORE row-normalization (raw / tfidf / bm25, via the shared scoreEdge kernel) so BM25-weighted PPR composes for free. keep is the optional frontier predicate (#601): a head is only walked into (and ranked) when keep(head) is true; the seed is the anchor exemption. ctx is polled every pprCtxCheckInterval pushes and returns ctx.Err() when cancelled. topN caps the ranked vertices returned (the request k); topN <= 0 returns every vertex with positive mass. An unknown seed yields an empty graph (the seed vertex alone is never emitted without neighbours).

func (*GraphCache[S, T]) PutEdgeWithExpiration 

func (c *GraphCache[S, T]) PutEdgeWithExpiration(tail, head S, w float32, expiration time.Time)

PutEdgeWithExpiration atomically replaces the (tail, head) edge weight. AddEdgeWithExpiration is additive (Add semantics) so a naive "DeleteEdge + AddEdgeWithExpiration" sequence performed by callers exposes a window in which concurrent GetEdge readers observe a spurious NotFound. PutEdgeWithExpiration takes the write lock once and performs the delete + add under the same lock, restoring atomicity for the idempotent Put semantics.

func (*GraphCache[S, T]) PutEdgeWithExpirationHLC 

func (c *GraphCache[S, T]) PutEdgeWithExpirationHLC(tail, head S, w float32, expiration time.Time, ts hlc.Timestamp) bool

PutEdgeWithExpirationHLC is the LWW-aware sibling of PutEdgeWithExpiration used by the replication apply path. Returns applied=false when the stored edge's last accepted HLC is strictly newer than ts. Endpoint vertices are auto-created (matching PutEdgeWithExpiration) regardless of the LWW outcome, unless an edge tombstone rejects the write before storage is touched.

func (*GraphCache[S, T]) PutEdgesWithExpiration 

func (c *GraphCache[S, T]) PutEdgesWithExpiration(items []EdgeItem[S])

PutEdgesWithExpiration replaces every supplied edge atomically under a single write lock. Each (tail, head) pair is deleted and re-added so the resulting weight is exactly the supplied weight — matching PutEdgeWithExpiration semantics for each item in the batch.

func (*GraphCache[S, T]) PutEdgesWithExpirationHLC added in v0.9.0 

func (c *GraphCache[S, T]) PutEdgesWithExpirationHLC(items []EdgeItem[S], ts hlc.Timestamp)

PutEdgesWithExpirationHLC is the LWW-aware, single-lock batch sibling of PutEdgesWithExpiration used by the LOCAL write path when replication is enabled. Like PutVerticesWithExpirationHLC it stamps every edge with the originating mutation's ts so PutEdge resolves as an LWW-Register on (tail, head) across replicas rather than diverging when two origins write the same edge concurrently. Endpoint vertices are auto-created regardless of the per-edge LWW outcome (matching PutEdgeWithExpirationHLC) so traversal always sees the endpoints.

func (*GraphCache[S, T]) PutVertex 

func (c *GraphCache[S, T]) PutVertex(key S, value T)

func (*GraphCache[S, T]) PutVertexWithExpiration 

func (c *GraphCache[S, T]) PutVertexWithExpiration(key S, value T, expiration time.Time)

func (*GraphCache[S, T]) PutVertexWithExpirationHLC 

func (c *GraphCache[S, T]) PutVertexWithExpirationHLC(key S, value T, expiration time.Time, ts hlc.Timestamp) bool

PutVertexWithExpirationHLC is the LWW-aware sibling of PutVertexWithExpiration used by the replication apply path. When the stored HLC for key is strictly newer than ts the call is a no-op and returns applied=false. A zero ts always applies and is recorded as the zero watermark (nothing can be strictly older than it). Local writers continue to use PutVertexWithExpiration; this helper is intentionally narrow to the replicated path so non-replicated workloads pay nothing.

func (*GraphCache[S, T]) PutVertexWithTTL 

func (c *GraphCache[S, T]) PutVertexWithTTL(key S, value T, ttl time.Duration)

func (*GraphCache[S, T]) PutVerticesWithExpiration 

func (c *GraphCache[S, T]) PutVerticesWithExpiration(items []VertexItem[S, T])

PutVerticesWithExpiration writes every supplied vertex under a single write lock. Concurrent readers observe either the pre-batch or the post-batch state — never an intermediate snapshot where some keys are present and others are not.

Search-document analysis (tokenization) for the whole batch runs BEFORE the lock is taken (see prepareSearchDocs) so the expensive per-vertex work never serializes other writers behind the aggregate graph lock; only the cheap store + postings mutation happens under c.mu (#739).

func (*GraphCache[S, T]) PutVerticesWithExpirationHLC added in v0.9.0 

func (c *GraphCache[S, T]) PutVerticesWithExpirationHLC(items []VertexItem[S, T], ts hlc.Timestamp)

PutVerticesWithExpirationHLC is the LWW-aware, single-lock batch sibling of PutVerticesWithExpiration used by the LOCAL write path when replication is enabled. Every item is stamped with the SAME ts — the HLC the originating mutation is logged under — so the origin's own writes participate in last-writer-wins on equal footing with the values its peers apply from its mutation log.

This closes a convergence hole: when the local path used the non-HLC PutVerticesWithExpiration it stored a value WITHOUT recording a vertexHLC watermark, so a concurrently-written OLDER value replayed from a peer would clobber the origin's newer value on the origin (the incoming write saw no watermark to lose to) while every other replica kept the newer value — permanent divergence for the same key. Stamping the watermark here makes PutVertex an LWW-Register on every replica, exactly as docs/replication.md specifies ("Higher HLC wins; same HLC ⇒ higher origin ID wins").

Per item the usual guards apply: a write whose ts is strictly older than the key's tombstone, or strictly older than the key's stored vertexHLC, is skipped. Born-expired items follow the same dead-on-arrival handling as PutVerticesWithExpiration (the entry is removed, not stored) so the #698 high-water optimisation is preserved; the watermark is still recorded so a later strictly-older write cannot resurrect the key.

func (*GraphCache[S, T]) ScanByPrefix 

func (c *GraphCache[S, T]) ScanByPrefix(ctx context.Context, prefix string, fn func(projected string, key S, value T) bool) bool

ScanByPrefix iterates every live vertex whose projected key starts with prefix, in lexicographic order, invoking fn for each one. fn may return false to stop the walk early; iteration also stops when ctx is cancelled. The boolean return is fn's last verdict (true means the caller exhausted the result set without asking to stop).

Consistency: ScanByPrefix takes a point-in-time snapshot of the matching (projected, key, value) triples under c.mu.RLock and then releases the lock BEFORE invoking fn. As a result fn is free to call back into any GraphCache method (including writers) without deadlock, at the cost of possibly observing keys whose live state has since changed \u2014 the snapshot itself is consistent, but the cache may have moved on by the time fn runs. Callers that need to react to the live state should re-fetch via GetVertex inside fn.

Releasing the lock before fn runs also avoids the sync.RWMutex-reentrancy hazard: a concurrent writer queued on c.mu would otherwise block any nested RLock from the visitor's goroutine (Go's RWMutex forbids recursive read-locking when a writer is waiting), which previously deadlocked callers whose fn touched the cache. See #202.

If the prefix index has not been enabled (see EnablePrefixIndex), ScanByPrefix returns false immediately and invokes fn zero times. The boolean is true if the prefix index is enabled and the walk completed without fn requesting a stop.

Vertices whose TTL has already expired but which have not yet been flushed are skipped at snapshot time \u2014 ScanByPrefix returns the same set of vertices a matching sequence of GetVertex calls under the snapshot lock would have returned.

fn receives both the projected string key (the prefix-index view) and the original S key (the cache view). For S = string they coincide; for other S the projection is intentionally lossy and the original key is the one suitable for downstream GraphCache calls.

func (*GraphCache[S, T]) ScanEdgesByPrefix 

func (c *GraphCache[S, T]) ScanEdgesByPrefix(
	ctx context.Context,
	tailPrefix, headPrefix string,
	fn func(tailProjected string, tail S, headProjected string, head S, weight float32, expiration time.Time) bool,
) bool

ScanEdgesByPrefix iterates every live edge whose tail projected key starts with tailPrefix AND whose head projected key starts with headPrefix, in ascending (tail, head) order, invoking fn for each one. Either prefix may be empty to disable the corresponding filter (both empty scans every edge).

fn may return false to stop the walk early; iteration also stops when ctx is cancelled. The boolean return is fn's last verdict — true means the caller exhausted the matching set without asking to stop.

Implementation: the walk reuses the vertex-side prefix index for the tail dimension. Every edge endpoint is auto-created as a vertex on insert (see AddEdgeWithExpiration / PutEdgeWithExpiration), so the radix already covers all live tails. For each matching tail the head dimension is served by a per-tail head radix (Issue #167) when available — see scanTailHeadsFast — or falls back to the v1 materialise-and-sort path (scanTailHeadsFallback) when the head index is disabled. The two paths emit identical (tailProj, headProj) sets for any given graph state.

Locking contract mirrors ScanByPrefix (#742): the matching edge rows are collected into a point-in-time snapshot under c.mu.RLock, the lock is released, and only THEN is fn invoked per row. Because fn runs after the lock is dropped, a slow visitor no longer starves writers for the whole scan, and fn MAY call back into GraphCache (including write methods) without risking sync.RWMutex reentrancy. The tradeoff is that fn observes a consistent snapshot taken at collection time, not necessarily the latest state after the lock was released, and an unbounded scan buffers one row per matching edge — callers that need to bound memory should scope the scan with prefixes or page at the service layer.

Returns false when the prefix index is not enabled, when ctx is cancelled, or when fn requested an early stop; true on a clean exhaustion of the matching set.

func (*GraphCache[S, T]) SearchIndexStats added in v0.9.0 

func (c *GraphCache[S, T]) SearchIndexStats() (terms, docs int)

SearchIndexStats returns the current number of distinct terms and indexed documents in the optional search index. Both values are 0 when the search index is disabled. Safe for concurrent use; intended for Prometheus gauge sampling.

func (*GraphCache[S, T]) SearchVertices 

func (c *GraphCache[S, T]) SearchVertices(query string, limit int, keyPrefix string) []search.Result[S]

SearchVertices returns up to limit keys whose indexed content matches query, ranked most-relevant first by BM25. An empty or unanalyzable query, a disabled index, or limit <= 0 returns nil. A non-empty keyPrefix scopes the results to a namespace, keeping only keys whose projection (the same one the prefix index uses) starts with keyPrefix.

Consistency mirrors ScanByPrefix: a hit is returned only when the vertex is still live, so a vertex whose TTL has expired but which has not yet been flushed is skipped — the returned set matches what a matching sequence of GetVertex calls under the snapshot lock would surface. Because matching is the index's boolean OR over the query terms, limit is applied after the liveness and prefix filters so a full page of live, in-scope hits is returned whenever one exists.

func (*GraphCache[S, T]) SetGCEdgeBudget added in v0.11.0 

func (c *GraphCache[S, T]) SetGCEdgeBudget(n int)

SetGCEdgeBudget bounds how many tail buckets the GC edge sweep walks per Watch tick (#744). A positive n caps each tick at n tails, spreading a large O(E) sweep across multiple ticks to bound the c.mu.Lock pause; the sweep carries a cursor across ticks so every tail is still reclaimed within bounded time. n <= 0 (the default) restores the full single-tick sweep and clears any in-flight cursor. Safe for concurrent use; takes effect on the next tick.

func (*GraphCache[S, T]) SetGCHooks 

func (c *GraphCache[S, T]) SetGCHooks(onExpire func(kind string, n int), onGCDuration func(d time.Duration))

SetGCHooks installs optional observability callbacks invoked from Watch after every GC tick. Either argument may be nil. Hooks must not call back into the cache (re-entrant locking would deadlock). Safe for concurrent use.

func (*GraphCache[S, T]) SnapshotEdges 

func (c *GraphCache[S, T]) SnapshotEdges() []SnapshotEdge[S]

SnapshotEdges returns a materialised snapshot of every live edge in the cache, preserving the per-contribution weight decomposition. The snapshot is taken under a single write-lock pass; the returned slice is independent of the cache. Edges whose every contribution has decayed are skipped (they would otherwise show up as ghost entries with zero total weight).

Memory cost is O(E + sum of bucket sizes); same scope rationale as SnapshotVertices.

func (*GraphCache[S, T]) SnapshotGraph added in v0.8.0 

func (c *GraphCache[S, T]) SnapshotGraph() GraphSnapshot[S, T]

SnapshotGraph returns a materialised whole-graph snapshot — every live vertex and edge — taken under a SINGLE write-lock pass, so the vertex and edge sides reflect the same instant. Calling SnapshotVertices and SnapshotEdges separately locks twice and can observe a vertex/edge set that never co-existed; SnapshotGraph closes that window, which is what a whole-graph backup/restore needs.

The returned slices are independent of the cache and safe to iterate without further locking. Expired vertices and fully-decayed edges are skipped, identical to the single-kind methods. Memory cost is O(live vertices + live edges); like the single-kind snapshots it is intended for bounded, infrequent operations (backup, replication bootstrap), not hot read paths.

func (*GraphCache[S, T]) SnapshotVertices 

func (c *GraphCache[S, T]) SnapshotVertices() []SnapshotVertex[S, T]

SnapshotVertices returns a materialised snapshot of every live vertex in the cache. The snapshot is taken under a single write-lock pass; the returned slice is independent of the cache and safe to iterate without further locking. Expired vertices (those past Expiration at snapshot time) are skipped.

Memory cost is O(N) in the live vertex count — the API is intended for replication bootstrap (#184), which is a bounded, infrequent operation. Hot read paths should keep using GetVertex.

func (*GraphCache[S, T]) VertexCount 

func (c *GraphCache[S, T]) VertexCount() int

VertexCount returns the live vertex count under an RLock. Intended for Prometheus gauges that sample the cache periodically.

func (*GraphCache[S, T]) VertexHLCCount added in v0.9.0 

func (c *GraphCache[S, T]) VertexHLCCount() int

VertexHLCCount returns the number of entries in the per-key LWW watermark map (vertexHLC). Under a healthy workload this equals the number of distinct vertex keys ever touched by the replication apply path and should track the live vertex count after a full GC sweep. A value that grows without bound across GC ticks is a leak signal (see issue #700). Returns 0 when the map has never been initialised (no replicated writes have arrived yet).

func (*GraphCache[S, T]) VertexHLCHighWater added in v0.10.0 

func (c *GraphCache[S, T]) VertexHLCHighWater() int

VertexHLCHighWater returns the largest number of vertexHLC entries ever observed at the start of a GC sweep — the per-cycle peak before stale watermarks are drained. Unlike VertexHLCCount (which reports the current, post-sweep len and therefore reads low right after a drain), this value is monotonic non-decreasing and records the all-time churn peak. It used to also proxy the map's retained bucket-array heap (Go never shrinks a map after delete), but sweepStaleVertexHLCLocked now reallocates the map after a large drain, so a high VertexHLCHighWater no longer implies pinned heap — it is a historical confirm-the-phenomenon signal for the ttl_churn leak gate (#700/#719/#727). Returns 0 when no sweep has run yet. Safe for concurrent use; intended for Prometheus gauge sampling.

func (*GraphCache[S, T]) Watch 

func (c *GraphCache[S, T]) Watch(ctx context.Context, interval time.Duration)

type GraphSnapshot added in v0.8.0 

type GraphSnapshot[S comparable, T any] struct {
	Vertices []SnapshotVertex[S, T]
	Edges    []SnapshotEdge[S]
}

GraphSnapshot is the result of GraphCache.SnapshotGraph: every live vertex and edge materialised under a single lock acquisition, so the two slices reflect one point-in-time. The type is named GraphSnapshot (not SnapshotGraph) so the SnapshotGraph name is free for the method.

type SnapshotContribution 

type SnapshotContribution struct {
	Weight     float32
	Expiration time.Time
	ContribID  ContribID
}

SnapshotContribution mirrors a single live entry inside an edge's weight bucket. Snapshot deliberately preserves the per-contribution decomposition so that the receiver's ContribID dedup (#182) continues to suppress duplicates when the live tail re-delivers the same contribution after bootstrap.

type SnapshotEdge 

type SnapshotEdge[S comparable] struct {
	Tail          S
	Head          S
	HLC           hlc.Timestamp
	Contributions []SnapshotContribution
}

SnapshotEdge is one entry of GraphCache.SnapshotEdges. HLC carries the bucket's last LWW position (zero when no LWW write has happened); the receiver uses it as the causal floor when replaying each contribution via AddEdgeWithExpirationContribHLC.

type SnapshotVertex 

type SnapshotVertex[S comparable, T any] struct {
	Key        S
	Value      T
	Expiration time.Time
	HLC        hlc.Timestamp
}

SnapshotVertex is one entry of GraphCache.SnapshotVertices. It carries the live vertex value, its expiration, and the last replication HLC the LWW apply path (#182) recorded for this key. HLC is the zero value when the vertex has never been touched by a replicated Put — that means "no causal floor", which is exactly what receivers feed back into PutVertexWithExpirationHLC.

type VertexItem 

type VertexItem[S comparable, T any] struct {
	Key        S
	Value      T
	Expiration time.Time
}

VertexItem is a single (key, value, expiration) tuple supplied to PutVerticesWithExpiration.

Source Files

View all Source files

Directories

Path Synopsis

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.