Sweep dangling edges: detect edges whose source or destination node has been deleted (tracked per-tenant in `deleted_nodes`). Called periodically from the idle loop. Removes dangling edges from the tenant's CSR partition and from the tenant-scoped edge store. Returns the total number of edges removed.
(&mut self)
| 223 | /// from the tenant's CSR partition and from the tenant-scoped |
| 224 | /// edge store. Returns the total number of edges removed. |
| 225 | pub fn sweep_dangling_edges(&mut self) -> usize { |
| 226 | if self.deleted_nodes.is_empty() { |
| 227 | return 0; |
| 228 | } |
| 229 | let mut removed = 0; |
| 230 | // Copy (tenant, node) pairs so we can mutate `self.csr` and |
| 231 | // `self.edge_store` without borrowing the map during |
| 232 | // iteration. |
| 233 | let work: Vec<(crate::types::TenantId, String)> = self |
| 234 | .deleted_nodes |
| 235 | .iter() |
| 236 | .flat_map(|(tid, set)| set.iter().map(move |n| (*tid, n.clone()))) |
| 237 | .collect(); |
| 238 | let swept_nodes = work.len(); |
| 239 | for (tid, node) in &work { |
| 240 | let edges = match self.csr.partition_mut(*tid) { |
| 241 | Some(partition) => partition.remove_node_edges(node), |
| 242 | None => 0, |
| 243 | }; |
| 244 | if edges > 0 { |
| 245 | let ord = self.hlc.next_ordinal(); |
| 246 | if let Err(e) = self.edge_store.delete_edges_for_node(*tid, node, ord) { |
| 247 | tracing::warn!( |
| 248 | core = self.core_id, |
| 249 | tid = tid.as_u64(), |
| 250 | node = %node, |
| 251 | error = %e, |
| 252 | "sweep: failed to delete edges from store" |
| 253 | ); |
| 254 | } |
| 255 | removed += edges; |
| 256 | } |
| 257 | } |
| 258 | if removed > 0 { |
| 259 | tracing::info!( |
| 260 | core = self.core_id, |
| 261 | removed, |
| 262 | deleted_nodes = swept_nodes, |
| 263 | "dangling edge sweep complete" |
| 264 | ); |
| 265 | } |
| 266 | removed |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | /// Lowercase `v` iff `case_insensitive` — used so COLLATE NOCASE indexes |
no test coverage detected