graphrefly/
batch.rs

1//! Batch — declarative transactional wave grouping (D12 / L2.E).
2//!
3//! `batch` runs a closure that may drive multiple nodes; **commit is always
4//! implicit** on success, **rollback on panic or explicit rollback**. This eliminates the
5//! forgot-to-commit class of bugs, and the three languages align naturally.
6//! `bctx.rollback()` is an explicit escape hatch within the closure.
7//!
8//! ## Interaction with the tier table (D34)
9//!
10//! Batch-deferred tiers (`>= Value`, see [`crate::protocol::Tier::is_batch_deferred`])
11//! are held until the batch commits, then flushed as the committed boundary;
12//! immediate tiers (`< Value`) still propagate within the batch. The deferred
13//! self-rewire drain (`ctx.rewire_next`, D47) applies AT the committed boundary —
14//! never on an un-committed/paused view (C-11/C-22; pause drain timing = B24).
15//!
16//! Conformance touchpoints: C-3 (INVALIDATE fan-in idempotency exercised via
17//! batch), C-19 (undirty timing), and C-22 (batch-before-rewire).
18
19use std::cell::{Cell, RefCell};
20use std::panic::{catch_unwind, resume_unwind, AssertUnwindSafe};
21use std::rc::Rc;
22
23use crate::host_boundary::is_host_boundary_abort_payload;
24use crate::node::{
25    boundary_root_for, clear_all_deferred_boundary_root, clear_deferred_boundary_root,
26    drain_committed_boundary_root, BatchTarget, BoundaryRoot, Core,
27};
28use crate::protocol::{AnyValue, Wave};
29
30/// Explicit rollback escape hatch for [`batch`] (D12).
31pub struct BatchCtx {
32    rolled_back: Rc<Cell<bool>>,
33}
34
35impl BatchCtx {
36    /// Discard all deferred settle slices instead of committing them.
37    pub fn rollback(&self) {
38        self.rolled_back.set(true);
39    }
40}
41
42struct ActiveBatch {
43    targets: Vec<BatchTarget>,
44    order: Vec<usize>,
45    deferred: Vec<Option<Wave<AnyValue>>>,
46    boundary_roots: Vec<BoundaryRoot>,
47    committed: Rc<Cell<bool>>,
48    rolled_back: Rc<Cell<bool>>,
49    collecting: bool,
50}
51
52impl ActiveBatch {
53    fn new(committed: Rc<Cell<bool>>, rolled_back: Rc<Cell<bool>>) -> Self {
54        Self {
55            targets: Vec::new(),
56            order: Vec::new(),
57            deferred: Vec::new(),
58            boundary_roots: Vec::new(),
59            committed,
60            rolled_back,
61            collecting: true,
62        }
63    }
64}
65
66thread_local! {
67    static ACTIVE_BATCH: RefCell<Option<ActiveBatch>> = const { RefCell::new(None) };
68}
69
70/// True while a batch frame is open. Boundary drains are blocked for the whole frame,
71/// including the commit loop, so queued topology applies only after the committed view.
72pub(crate) fn boundary_drains_blocked() -> bool {
73    ACTIVE_BATCH.with(|b| b.borrow().is_some())
74}
75
76pub(crate) fn collecting_batch() -> bool {
77    ACTIVE_BATCH.with(|b| b.borrow().as_ref().is_some_and(|batch| batch.collecting))
78}
79
80/// Return the active batch's commit token for boundary tasks caused while the
81/// batch frame is open. The token flips only after commit succeeds; rollback or
82/// commit failure leaves it false so cleanup can drop only uncommitted tasks.
83pub(crate) fn active_batch_committed_token() -> Option<Rc<Cell<bool>>> {
84    ACTIVE_BATCH.with(|b| b.borrow().as_ref().map(|batch| batch.committed.clone()))
85}
86
87/// Remember that a graph-local committed-boundary task was queued during this batch.
88/// Some tasks do not also create a tier>=3 batched wave, so they need their graph root
89/// included in the post-commit drain/rollback cleanup set (R-rewire-batch-boundary).
90pub(crate) fn register_boundary_root(target: &Core) {
91    ACTIVE_BATCH.with(|b| {
92        let mut active = b.borrow_mut();
93        let Some(batch) = active.as_mut() else {
94            return;
95        };
96        if batch.boundary_roots.iter().any(|r| r.same_graph(target)) {
97            return;
98        }
99        batch.boundary_roots.push(boundary_root_for(target));
100    });
101}
102
103/// Capture a tier>=3 settle slice into the current collecting batch. Returns false when
104/// there is no open collecting batch (including the commit phase).
105pub(crate) fn defer_to_batch(target: &Core, wave: Wave<AnyValue>) -> bool {
106    ACTIVE_BATCH.with(|b| {
107        let mut active = b.borrow_mut();
108        let Some(batch) = active.as_mut() else {
109            return false;
110        };
111        if !batch.collecting {
112            return false;
113        }
114        let index = match batch.targets.iter().position(|entry| entry.matches(target)) {
115            Some(index) => index,
116            None => {
117                let Some(target) = BatchTarget::from_core(target) else {
118                    return false;
119                };
120                let index = batch.targets.len();
121                batch.targets.push(target);
122                batch.deferred.push(None);
123                batch.order.push(index);
124                index
125            }
126        };
127        batch.deferred[index] = Some(wave);
128        true
129    })
130}
131
132/// Return the active batch's commit token, but only when this target has an
133/// uncommitted settle slice in that batch (R-rewire-batch-boundary / D67).
134pub(crate) fn committed_after_batch_for_target(target: &Core) -> Option<Rc<Cell<bool>>> {
135    ACTIVE_BATCH.with(|b| {
136        let active = b.borrow();
137        let batch = active.as_ref()?;
138        if !batch.collecting
139            || !batch.targets.iter().enumerate().any(|(index, existing)| {
140                existing.matches(target) && batch.deferred.get(index).is_some_and(Option::is_some)
141            })
142        {
143            return None;
144        }
145        Some(batch.committed.clone())
146    })
147}
148
149struct BatchFinish {
150    targets: Vec<BatchTarget>,
151    order: Vec<usize>,
152    deferred: Vec<Option<Wave<AnyValue>>>,
153    boundary_roots: Vec<BoundaryRoot>,
154    committed: Rc<Cell<bool>>,
155    explicit_rollback: bool,
156}
157
158fn take_for_finish() -> BatchFinish {
159    ACTIVE_BATCH.with(|b| {
160        let mut active = b.borrow_mut();
161        let batch = active.as_mut().expect("batch frame installed");
162        batch.collecting = false;
163        BatchFinish {
164            targets: std::mem::take(&mut batch.targets),
165            order: std::mem::take(&mut batch.order),
166            deferred: std::mem::take(&mut batch.deferred),
167            boundary_roots: std::mem::take(&mut batch.boundary_roots),
168            committed: batch.committed.clone(),
169            explicit_rollback: batch.rolled_back.get(),
170        }
171    })
172}
173
174fn clear_active() {
175    ACTIVE_BATCH.with(|b| {
176        *b.borrow_mut() = None;
177    });
178}
179
180fn commit(targets: &[BatchTarget], deferred: Vec<Option<Wave<AnyValue>>>) {
181    for (target_index, wave) in deferred.into_iter().enumerate() {
182        if let Some(wave) = wave {
183            targets[target_index].commit_batched_wave(wave);
184        }
185    }
186}
187
188fn rollback(targets: &[BatchTarget], order: Vec<usize>) {
189    for target_index in order {
190        targets[target_index].rollback_batched();
191    }
192}
193
194fn boundary_graph_roots(
195    targets: &[BatchTarget],
196    order: &[usize],
197    deferred: &[Option<Wave<AnyValue>>],
198) -> Vec<BoundaryRoot> {
199    let mut roots: Vec<BoundaryRoot> = Vec::new();
200    let mut push_unique = |target: &BatchTarget| {
201        let root = target.boundary_root();
202        if roots.iter().any(|r| r.same_root(&root)) {
203            return;
204        }
205        roots.push(root);
206    };
207    for target_index in order {
208        if let Some(target) = targets.get(*target_index) {
209            push_unique(target);
210        }
211    }
212    for (target_index, wave) in deferred.iter().enumerate() {
213        if wave.is_some() {
214            let Some(target) = targets.get(target_index) else {
215                continue;
216            };
217            push_unique(target);
218        }
219    }
220    roots
221}
222
223fn take_late_boundary_roots() -> Vec<BoundaryRoot> {
224    ACTIVE_BATCH.with(|b| {
225        let mut active = b.borrow_mut();
226        active
227            .as_mut()
228            .map(|batch| std::mem::take(&mut batch.boundary_roots))
229            .unwrap_or_default()
230    })
231}
232
233fn append_boundary_roots(
234    task_roots: &mut Vec<BoundaryRoot>,
235    core_roots: &[BoundaryRoot],
236    boundary_roots: Vec<BoundaryRoot>,
237) {
238    for root in boundary_roots {
239        if core_roots.iter().any(|r| r.same_root(&root))
240            || task_roots.iter().any(|r| r.same_root(&root))
241        {
242            continue;
243        }
244        task_roots.push(root);
245    }
246}
247
248fn clear_boundary_for_roots(core_roots: &[BoundaryRoot], task_roots: &[BoundaryRoot]) {
249    for root in core_roots {
250        clear_deferred_boundary_root(root);
251    }
252    for root in task_roots {
253        clear_deferred_boundary_root(root);
254    }
255}
256
257fn drain_boundary_for_roots(core_roots: &[BoundaryRoot], task_roots: &[BoundaryRoot]) {
258    let mut escaped: Option<Box<dyn std::any::Any + Send>> = None;
259    for root in core_roots {
260        let result = catch_unwind(AssertUnwindSafe(|| drain_committed_boundary_root(root)));
261        if let Err(payload) = result {
262            if is_host_boundary_abort_payload(payload.as_ref()) {
263                clear_all_boundary_for_roots(core_roots, task_roots);
264                resume_unwind(payload);
265            }
266            remember_first_boundary_panic(&mut escaped, Err(payload));
267        }
268    }
269    for root in task_roots {
270        let result = catch_unwind(AssertUnwindSafe(|| drain_committed_boundary_root(root)));
271        if let Err(payload) = result {
272            if is_host_boundary_abort_payload(payload.as_ref()) {
273                clear_all_boundary_for_roots(core_roots, task_roots);
274                resume_unwind(payload);
275            }
276            remember_first_boundary_panic(&mut escaped, Err(payload));
277        }
278    }
279    if let Some(e) = escaped {
280        std::panic::resume_unwind(e);
281    }
282}
283
284fn clear_all_boundary_for_roots(core_roots: &[BoundaryRoot], task_roots: &[BoundaryRoot]) {
285    for root in core_roots {
286        clear_all_deferred_boundary_root(root);
287    }
288    for root in task_roots {
289        clear_all_deferred_boundary_root(root);
290    }
291}
292
293fn remember_first_boundary_panic(
294    escaped: &mut Option<Box<dyn std::any::Any + Send>>,
295    result: std::thread::Result<()>,
296) {
297    if let Err(e) = result {
298        if escaped.is_none() {
299            *escaped = Some(e);
300        }
301    }
302}
303
304fn rollback_and_cleanup(
305    targets: Vec<BatchTarget>,
306    order: Vec<usize>,
307    mut task_roots: Vec<BoundaryRoot>,
308) -> std::thread::Result<()> {
309    let core_roots = boundary_graph_roots(&targets, &order, &[]);
310    let result = catch_unwind(AssertUnwindSafe(|| rollback(&targets, order)));
311    let late_roots = take_late_boundary_roots();
312    append_boundary_roots(&mut task_roots, &core_roots, late_roots);
313    clear_active();
314    clear_boundary_for_roots(&core_roots, &task_roots);
315    result
316}
317
318/// Run `f` as a declarative batch (D12): success commits, panic/rollback discards.
319///
320/// Nested batches join the outer frame; the outermost frame owns commit/rollback.
321pub fn batch<R>(f: impl FnOnce(&BatchCtx) -> R) -> R {
322    if let Some(rolled_back) =
323        ACTIVE_BATCH.with(|b| b.borrow().as_ref().map(|batch| batch.rolled_back.clone()))
324    {
325        return f(&BatchCtx { rolled_back });
326    }
327
328    let committed = Rc::new(Cell::new(false));
329    let rolled_back = Rc::new(Cell::new(false));
330    ACTIVE_BATCH.with(|b| {
331        *b.borrow_mut() = Some(ActiveBatch::new(committed.clone(), rolled_back.clone()));
332    });
333
334    let ctx = BatchCtx {
335        rolled_back: rolled_back.clone(),
336    };
337    let result = catch_unwind(AssertUnwindSafe(|| f(&ctx)));
338    let finish = take_for_finish();
339    let boundary_core_roots =
340        boundary_graph_roots(&finish.targets, &finish.order, &finish.deferred);
341    let mut boundary_task_roots = Vec::new();
342    append_boundary_roots(
343        &mut boundary_task_roots,
344        &boundary_core_roots,
345        finish.boundary_roots,
346    );
347
348    match result {
349        Ok(value) => {
350            if finish.explicit_rollback {
351                if let Err(payload) =
352                    rollback_and_cleanup(finish.targets, finish.order, boundary_task_roots)
353                {
354                    resume_unwind(payload);
355                }
356            } else {
357                let commit_result = catch_unwind(AssertUnwindSafe(|| {
358                    commit(&finish.targets, finish.deferred)
359                }));
360                let late_roots = take_late_boundary_roots();
361                append_boundary_roots(&mut boundary_task_roots, &boundary_core_roots, late_roots);
362                if let Err(payload) = commit_result {
363                    clear_active();
364                    clear_boundary_for_roots(&boundary_core_roots, &boundary_task_roots);
365                    resume_unwind(payload);
366                }
367                finish.committed.set(true);
368                clear_active();
369                drain_boundary_for_roots(&boundary_core_roots, &boundary_task_roots);
370            }
371            value
372        }
373        Err(payload) => {
374            if let Err(rollback_payload) =
375                rollback_and_cleanup(finish.targets, finish.order, boundary_task_roots)
376            {
377                resume_unwind(rollback_payload);
378            }
379            resume_unwind(payload);
380        }
381    }
382}
graphrefly/batch.rs

graphrefly/
batch.rs
