linkrs_storage/edge/single_mutable_csr/
write.rs

1//! Write path for `SingleMutableCsr`: insert, delete, revert and rollback.
2
3use linkrs_core::{StorageError, StorageResult};
4
5use super::super::csr_shared::{
6    can_revert_delete, decide_slot_delete, decode_endpoint_pair, DeleteSlotOutcome,
7};
8use super::super::{EdgeId, Timestamp, VertexId, INVALID_EDGE_ID};
9use super::{empty_slot, SingleMutableCsr};
10
11impl SingleMutableCsr {
12    pub fn insert_edge(
13        &mut self,
14        src: u32,
15        dst: VertexId,
16        edge_id: EdgeId,
17        _ts: Timestamp,
18    ) -> StorageResult<()> {
19        let src_idx = src as usize;
20
21        if src_idx >= self.vertex_capacity() {
22            self.ensure_vertex_capacity(src_idx + 1);
23        }
24
25        let (existing_hot, existing_cold) = match self.slot_at(src_idx) {
26            Some(probe) => (probe.hot(), probe.cold()),
27            None => (
28                super::super::HotNbr::dead_gap(),
29                super::super::ColdStamps::dead_gap(),
30            ),
31        };
32
33        // Physical uniqueness only: a live slot rejects the second insert
34        // regardless of timestamp, while a tombstoned or empty slot accepts
35        // any timestamp. Snapshot visibility is decided by the version
36        // authority above this layer.
37        if existing_cold.is_live() && existing_hot.edge_id != INVALID_EDGE_ID {
38            return Err(StorageError::conflict(format!(
39                "[SingleMutableCsr] insert conflict on src={}: slot holds live edge {:?}",
40                src, existing_hot.edge_id
41            )));
42        }
43
44        let was_empty = existing_hot.edge_id == INVALID_EDGE_ID || !existing_cold.is_live();
45        let (decoded_endpoint, rank) = decode_endpoint_pair(dst).ok_or_else(|| {
46            StorageError::invalid_input(format!("Malformed edge endpoint key: {}", dst))
47        })?;
48        let segment = self.slot_mut_or_alloc(src_idx);
49        let (_, off) = Self::locate(src_idx);
50        segment.hot[off] = super::super::HotNbr {
51            endpoint: decoded_endpoint,
52            rank,
53            edge_id,
54        };
55        segment.cold[off] = super::super::ColdStamps {
56            delete_ts: Timestamp::MAX,
57        };
58        self.set_present(src_idx, true);
59
60        if was_empty {
61            self.edge_count += 1;
62        }
63
64        Ok(())
65    }
66
67    pub fn delete_edge(&mut self, src: u32, edge_id: EdgeId, ts: Timestamp) -> StorageResult<bool> {
68        let src_idx = src as usize;
69
70        if src_idx >= self.vertex_capacity() {
71            return Ok(false);
72        }
73
74        let probe = match self.slot_at(src_idx) {
75            Some(probe) => probe,
76            None => return Ok(false),
77        };
78
79        if probe.edge_id == INVALID_EDGE_ID {
80            return Ok(false);
81        }
82
83        if probe.edge_id != edge_id {
84            return Ok(false);
85        }
86
87        if matches!(
88            decide_slot_delete(&probe, probe.edge_id, ts)?,
89            DeleteSlotOutcome::AlreadyStamped
90        ) {
91            return Ok(false);
92        }
93
94        self.stamp_delete(src_idx, ts);
95        self.edge_count -= 1;
96        Ok(true)
97    }
98
99    /// Delete the single matching live entry for full-match endpoint semantics.
100    ///
101    /// Returns the deleted count (0 or 1) so callers can reconcile.
102    /// Reporting variant stamps the slot and hands the id to `on_deleted`
103    /// in the same step, so no second lookup is needed.
104    pub fn delete_edge_by_dst_reporting(
105        &mut self,
106        src: u32,
107        dst: VertexId,
108        ts: Timestamp,
109        on_deleted: &mut dyn FnMut(EdgeId),
110    ) -> usize {
111        let src_idx = src as usize;
112
113        if src_idx >= self.vertex_capacity() {
114            return 0;
115        }
116
117        let (dst_ep, dst_rank) = match decode_endpoint_pair(dst) {
118            Some(pair) => pair,
119            None => return 0,
120        };
121        let probe = match self.slot_at(src_idx) {
122            Some(probe) => probe,
123            None => return 0,
124        };
125
126        if probe.edge_id == INVALID_EDGE_ID
127            || probe.endpoint != dst_ep
128            || probe.rank != dst_rank
129            || !self.cold_is_live_at(src_idx)
130        {
131            return 0;
132        }
133
134        self.stamp_delete(src_idx, ts);
135        self.edge_count -= 1;
136        on_deleted(probe.edge_id);
137        1
138    }
139
140    /// Delete the single matching live entry for full-match endpoint semantics.
141    ///
142    /// Returns the deleted count (0 or 1) so callers can reconcile.
143    pub fn delete_edge_by_dst(&mut self, src: u32, dst: VertexId, ts: Timestamp) -> usize {
144        let mut noop = |_: EdgeId| {};
145        self.delete_edge_by_dst_reporting(src, dst, ts, &mut noop)
146    }
147
148    pub fn revert_delete_by_offset(&mut self, src: u32, offset: i32, ts: Timestamp) -> bool {
149        if offset != 0 {
150            return false;
151        }
152
153        let src_idx = src as usize;
154
155        if src_idx >= self.vertex_capacity() {
156            return false;
157        }
158
159        let probe = match self.slot_at(src_idx) {
160            Some(probe) => probe,
161            None => return false,
162        };
163
164        // Only revert deletions that happened at or before rollback time.
165        if can_revert_delete(&probe, ts) {
166            self.clear_delete(src_idx);
167            self.edge_count += 1;
168            return true;
169        }
170
171        false
172    }
173
174    pub fn delete_edge_by_offset(
175        &mut self,
176        src: u32,
177        offset: i32,
178        ts: Timestamp,
179    ) -> StorageResult<bool> {
180        if offset != 0 {
181            return Ok(false);
182        }
183        let src_idx = src as usize;
184        if src_idx >= self.vertex_capacity() {
185            return Ok(false);
186        }
187        let edge_id = self
188            .hot_at(src_idx)
189            .map(|hot| hot.edge_id)
190            .unwrap_or(INVALID_EDGE_ID);
191        self.delete_edge(src, edge_id, ts)
192    }
193
194    pub fn rollback_insert(&mut self, src: u32, edge_id: EdgeId) -> bool {
195        let src_idx = src as usize;
196        if src_idx >= self.vertex_capacity() {
197            return false;
198        }
199        let hot = match self.hot_at(src_idx) {
200            Some(h) => h,
201            None => return false,
202        };
203        if hot.edge_id == INVALID_EDGE_ID {
204            return false;
205        }
206        if hot.edge_id != edge_id {
207            return false;
208        }
209        let was_live = self.cold_is_live_at(src_idx);
210        self.set_slot(src_idx, empty_slot());
211        if was_live {
212            self.edge_count -= 1;
213        }
214        true
215    }
216
217    pub fn revert_delete_by_edge_id(&mut self, src: u32, edge_id: EdgeId, ts: Timestamp) -> bool {
218        let src_idx = src as usize;
219        if src_idx >= self.vertex_capacity() {
220            return false;
221        }
222        let probe = match self.slot_at(src_idx) {
223            Some(probe) => probe,
224            None => return false,
225        };
226        if probe.edge_id == INVALID_EDGE_ID {
227            return false;
228        }
229        if probe.edge_id != edge_id {
230            return false;
231        }
232        if can_revert_delete(&probe, ts) {
233            self.clear_delete(src_idx);
234            self.edge_count += 1;
235            return true;
236        }
237        false
238    }
239}
linkrs_storage/edge/single_mutable_csr/write.rs

linkrs_storage/edge/single_mutable_csr/
write.rs
