Resolve the Ceiling: the latest version of `(collection, src, label, dst)` whose `system_from ≤ system_as_of`. Returns `Ok(None)` if no version exists at or before the cutoff, or if the latest qualifying version is a tombstone/GDPR erasure. When `valid_at_ms` is supplied, the resolved version must also satisfy `valid_from_ms ≤ valid_at_ms < valid_until_ms`; otherwise the method continues scannin
(
&self,
edge: EdgeRef<'_>,
system_as_of: i64,
valid_at_ms: Option<i64>,
)
| 20 | /// `valid_from_ms ≤ valid_at_ms < valid_until_ms`; otherwise the method |
| 21 | /// continues scanning to earlier system-time versions. |
| 22 | pub fn ceiling_resolve_edge( |
| 23 | &self, |
| 24 | edge: EdgeRef<'_>, |
| 25 | system_as_of: i64, |
| 26 | valid_at_ms: Option<i64>, |
| 27 | ) -> crate::Result<Option<Vec<u8>>> { |
| 28 | if system_as_of < 0 { |
| 29 | return Err(crate::Error::BadRequest { |
| 30 | detail: format!("ceiling_resolve_edge: negative system_as_of={system_as_of}"), |
| 31 | }); |
| 32 | } |
| 33 | let prefix = edge_version_prefix(edge.collection, edge.src, edge.label, edge.dst); |
| 34 | let upper = versioned_edge_key( |
| 35 | edge.collection, |
| 36 | edge.src, |
| 37 | edge.label, |
| 38 | edge.dst, |
| 39 | system_as_of, |
| 40 | )?; |
| 41 | let t = edge.tid.as_u64(); |
| 42 | |
| 43 | let read_txn = self |
| 44 | .db |
| 45 | .begin_read() |
| 46 | .map_err(|e| redb_err("begin_read", e))?; |
| 47 | let table = read_txn |
| 48 | .open_table(EDGES) |
| 49 | .map_err(|e| redb_err("open edges", e))?; |
| 50 | |
| 51 | // Inclusive upper — the exact key at system_as_of is a valid ceiling. |
| 52 | let range = table |
| 53 | .range((t, prefix.as_str())..=(t, upper.as_str())) |
| 54 | .map_err(|e| redb_err("ceiling range", e))?; |
| 55 | |
| 56 | // Walk newest-first by reversing the iterator. |
| 57 | for entry in range.rev() { |
| 58 | let (k, v) = entry.map_err(|e| redb_err("ceiling iter", e))?; |
| 59 | let (kt, composite) = k.value(); |
| 60 | if kt != t || !composite.starts_with(&prefix) { |
| 61 | break; |
| 62 | } |
| 63 | let bytes = v.value(); |
| 64 | if is_tombstone(bytes) || is_gdpr_erasure(bytes) { |
| 65 | return Ok(None); |
| 66 | } |
| 67 | let payload = EdgeValuePayload::decode(bytes)?; |
| 68 | match valid_at_ms { |
| 69 | Some(vt) if !(payload.valid_from_ms <= vt && vt < payload.valid_until_ms) => { |
| 70 | // This system-time version didn't assert the fact at `vt` — |
| 71 | // scan older versions. |
| 72 | continue; |
| 73 | } |
| 74 | _ => return Ok(Some(payload.properties)), |
| 75 | } |
| 76 | } |
| 77 | Ok(None) |
| 78 | } |
| 79 | } |
no test coverage detected