Scan every forward edge across all tenants in current-state, yielding `(TenantId, collection, src, label, dst, properties)`. Tombstoned and erased versions are skipped; only the Ceiling resolution at `system_as_of` is returned per base. `None` means "current state" (ordinal = `i64::MAX`).
(
&self,
system_as_of: Option<i64>,
)
| 44 | /// at `system_as_of` is returned per base. `None` means "current state" |
| 45 | /// (ordinal = `i64::MAX`). |
| 46 | pub fn scan_all_edges_decoded( |
| 47 | &self, |
| 48 | system_as_of: Option<i64>, |
| 49 | ) -> crate::Result<Vec<EdgeRecord>> { |
| 50 | let cutoff = system_as_of.unwrap_or(i64::MAX); |
| 51 | let read_txn = self |
| 52 | .db |
| 53 | .begin_read() |
| 54 | .map_err(|e| redb_err("begin_read", e))?; |
| 55 | let table = read_txn |
| 56 | .open_table(EDGES) |
| 57 | .map_err(|e| redb_err("open edges", e))?; |
| 58 | |
| 59 | let mut latest: HashMap<TenantBaseKey, (i64, Vec<u8>)> = HashMap::new(); |
| 60 | let mut tombstoned: HashSet<TenantBaseKey> = HashSet::new(); |
| 61 | |
| 62 | for entry in table.iter().map_err(|e| redb_err("iter", e))? { |
| 63 | let (k, v) = entry.map_err(|e| redb_err("iter entry", e))?; |
| 64 | let (t, composite) = k.value(); |
| 65 | let Some((coll, src, label, dst, sys)) = parse_versioned_edge_key(composite) else { |
| 66 | continue; |
| 67 | }; |
| 68 | if sys > cutoff { |
| 69 | continue; |
| 70 | } |
| 71 | let base: TenantBaseKey = ( |
| 72 | t, |
| 73 | coll.to_string(), |
| 74 | src.to_string(), |
| 75 | label.to_string(), |
| 76 | dst.to_string(), |
| 77 | ); |
| 78 | let bytes = v.value(); |
| 79 | if is_sentinel(bytes) { |
| 80 | match latest.get(&base) { |
| 81 | Some((cur_sys, _)) if *cur_sys > sys => {} |
| 82 | _ => { |
| 83 | latest.remove(&base); |
| 84 | tombstoned.insert(base); |
| 85 | } |
| 86 | } |
| 87 | continue; |
| 88 | } |
| 89 | if tombstoned.contains(&base) { |
| 90 | continue; |
| 91 | } |
| 92 | match latest.get(&base) { |
| 93 | Some((cur_sys, _)) if *cur_sys >= sys => {} |
| 94 | _ => match EdgeValuePayload::decode(bytes) { |
| 95 | Ok(payload) => { |
| 96 | latest.insert(base, (sys, payload.properties)); |
| 97 | } |
| 98 | Err(_) => continue, |
| 99 | }, |
| 100 | } |
| 101 | } |
| 102 | |
| 103 | let mut out = Vec::with_capacity(latest.len()); |