velesdb_core/collection/graph/edge_concurrent/snapshot.rs
1//! CSR snapshot management for `ConcurrentEdgeStore`.
2//!
3//! Extracted from `edge_concurrent/mod.rs` to reduce NLOC below the 500 threshold.
4//! Contains: `invalidate_snapshot()`, `rebuild_snapshot_best_effort()`,
5//! `rebuild_snapshot()`, `build_read_snapshot()`, `has_read_snapshot()`.
6
7use super::super::clustered_index::ClusteredIndex;
8use super::super::csr_snapshot::SnapshotBuilder;
9use super::super::edge::EdgeStore;
10use super::ConcurrentEdgeStore;
11use crate::error::Result;
12use std::sync::atomic::Ordering;
13use std::sync::Arc;
14
15impl ConcurrentEdgeStore {
16 /// Invalidates the clustered read snapshot.
17 ///
18 /// Called by every write method so that stale data is never served.
19 /// Readers fall back to per-shard lookup when the clustered snapshot
20 /// is absent.
21 #[inline]
22 pub(super) fn invalidate_snapshot(&self) {
23 // Fast path: skip write lock if snapshot is already absent.
24 let guard = self.clustered_snapshot.read();
25 if guard.is_some() {
26 drop(guard);
27 *self.clustered_snapshot.write() = None;
28 }
29 }
30
31 /// Best-effort CSR snapshot rebuild after a mutation.
32 ///
33 /// Sets the `csr_dirty` flag and increments the pending-write counter so
34 /// the lazy rebuild can be debounced (issue #905). The actual O(N+E)
35 /// rebuild is deferred until either a reader observes
36 /// [`CSR_REBUILD_WRITE_THRESHOLD`](super::CSR_REBUILD_WRITE_THRESHOLD)
37 /// accumulated writes, or a reader that has no per-shard fallback forces
38 /// it. Until then readers fall back to the authoritative per-shard data.
39 #[inline]
40 pub(super) fn rebuild_snapshot_best_effort(&self) {
41 self.record_pending_writes(1);
42 }
43
44 /// Records `count` accumulated edge mutations toward the CSR rebuild
45 /// debounce threshold and marks the snapshot dirty.
46 ///
47 /// Batch writers (`add_edges_batch`) must report the actual number of
48 /// edges inserted (issue #905 follow-up): reporting a flat `1` per batch
49 /// would let a bulk-loaded graph stay permanently below
50 /// [`CSR_REBUILD_WRITE_THRESHOLD`](super::CSR_REBUILD_WRITE_THRESHOLD),
51 /// so the CSR fast path would never engage.
52 #[inline]
53 pub(super) fn record_pending_writes(&self, count: u64) {
54 if count == 0 {
55 return;
56 }
57 self.pending_writes.fetch_add(count, Ordering::AcqRel);
58 self.csr_dirty.store(true, Ordering::Release);
59 }
60
61 /// Rebuilds the lock-free `CsrSnapshot` from all shards.
62 ///
63 /// Acquires read locks on all shards sequentially, merges outgoing edges
64 /// into a single `EdgeStore`, builds a `CsrSnapshot` via `SnapshotBuilder`,
65 /// and swaps it atomically into `self.csr_snapshot`.
66 ///
67 /// On failure the previous snapshot is retained (readers see stale but
68 /// structurally valid data).
69 ///
70 /// # Locking contract (must-read)
71 ///
72 /// The caller **must not** hold a write lock on `edge_ids` **or** any
73 /// `shards[*]` lock (read or write) when invoking this method. The
74 /// method walks every shard and takes a read lock on each one in turn;
75 /// holding a same-shard write lock deadlocks against the reader, and
76 /// holding an `edge_ids` write lock deadlocks against the downstream
77 /// `label_table` / snapshot consumers in the same lock-order chain.
78 ///
79 /// The only two supported call sites are:
80 ///
81 /// * [`build_read_snapshot`](Self::build_read_snapshot) (this file) —
82 /// acquires `edge_ids` as **read-only** and releases per-shard read
83 /// locks between loop iterations.
84 /// * The lazy-rebuild path in
85 /// `collection/graph/edge_concurrent/query.rs::ensure_csr_fresh`
86 /// (reachable from `get_csr_snapshot`) — runs with no outer locks
87 /// held.
88 ///
89 /// Mutation methods (`add_edge`, `remove_edge`, `flush`, …) must
90 /// instead call
91 /// [`rebuild_snapshot_best_effort`](Self::rebuild_snapshot_best_effort)
92 /// which only flips the dirty flag and defers the actual rebuild to
93 /// the next reader. Cross-reference: `docs/CONCURRENCY_MODEL.md`
94 /// (graph collection lock-ordering section).
95 ///
96 /// # Errors
97 ///
98 /// Returns `Error::SnapshotBuildFailed` if the merge or build fails.
99 #[allow(clippy::unnecessary_wraps)] // Reason: Result kept for future allocation-failure propagation
100 pub(crate) fn rebuild_snapshot(&self) -> Result<()> {
101 // Build a merged EdgeStore from all shards (outgoing edges only).
102 // We iterate shards directly instead of using `to_merged_edge_store()`
103 // to avoid acquiring `edge_ids` (which may already be write-locked
104 // by the calling mutation method).
105 let mut merged = EdgeStore::new();
106 for shard in &self.shards {
107 let guard = shard.read();
108 for edge in guard.all_edges() {
109 // Ignore duplicates — cross-shard edges appear in both shards
110 // but `add_edge` deduplicates by edge ID.
111 let _ = merged.add_edge(edge.clone());
112 }
113 }
114 let label_table = self.label_table.read();
115 let new_snapshot = SnapshotBuilder::build(&merged, &label_table);
116 self.csr_snapshot.store(Arc::new(new_snapshot));
117 Ok(())
118 }
119
120 /// Builds a CSR-like read snapshot from current shard state.
121 ///
122 /// The snapshot stores only outgoing neighbor **target node IDs** per source
123 /// node in contiguous memory, enabling [`with_neighbors()`](Self::with_neighbors)
124 /// to provide zero-copy `&[u64]` access without shard locking.
125 ///
126 /// # Limitation — target IDs only
127 ///
128 /// The snapshot does **not** store edge IDs, labels, or properties.
129 /// It is optimized for BFS neighbor expansion where only connectivity
130 /// matters. To retrieve full edge metadata (edge ID, label, properties),
131 /// use [`get_outgoing()`](Self::get_outgoing) which reads from the
132 /// authoritative shard data.
133 ///
134 /// Call this after bulk inserts, after `flush()`, or after loading
135 /// from disk. The snapshot is automatically invalidated on any write.
136 pub fn build_read_snapshot(&self) {
137 let ids = self.edge_ids.read();
138 let edge_count = ids.len();
139 // Rough estimate: each edge contributes one outgoing target entry.
140 let mut snapshot = ClusteredIndex::with_capacity(edge_count, edge_count);
141
142 for (&edge_id, &source_id) in ids.iter() {
143 let shard_idx = self.shard_index(source_id);
144 let guard = self.shards[shard_idx].read();
145 if let Some(edge) = guard.get_edge(edge_id) {
146 snapshot.insert(source_id, edge.target());
147 }
148 }
149
150 // Compact once to eliminate any fragmentation from insert order.
151 snapshot.compact();
152
153 *self.clustered_snapshot.write() = Some(snapshot);
154
155 // Also rebuild the lock-free CSR snapshot.
156 let _ = self.rebuild_snapshot();
157
158 // The freshly built snapshot reflects all edges, so clear the dirty
159 // flag and reset the debounce counter (issue #905). Without this the
160 // next reader would needlessly rebuild again even though the snapshot
161 // is already authoritative.
162 self.pending_writes.store(0, Ordering::Release);
163 self.csr_dirty.store(false, Ordering::Release);
164 }
165
166 /// Returns `true` if a CSR read snapshot is currently available.
167 #[must_use]
168 pub fn has_read_snapshot(&self) -> bool {
169 self.clustered_snapshot.read().is_some()
170 }
171}