native_ipc/
batch.rs

1//! One-owner transfer transactions and keyed post-commit regions.
2
3use std::collections::BTreeMap;
4
5use crate::active::{ActiveReader, ActiveWriter};
6use crate::protocol::{ManifestEntry, PeerAccess};
7use crate::region::{PreparedRegion, RegionId, WriterEndpoint};
8
9/// Portable batch construction or active-set lookup failure.
10#[derive(Clone, Copy, Debug, Eq, PartialEq)]
11pub enum BatchError {
12    /// Negotiated batch limits are zero or exceed hard maxima.
13    InvalidLimits,
14    /// A batch must contain at least one region.
15    Empty,
16    /// Region count exceeds the negotiated limit or hard maximum of sixteen.
17    TooManyRegions,
18    /// A region ID occurs more than once in the transaction.
19    DuplicateRegionId(RegionId),
20    /// Checked aggregate logical or mapped bytes overflowed or exceeded limits.
21    BatchBytesExceeded,
22    /// An expected region has zero or non-native logical length.
23    InvalidRegionLength,
24    /// The keyed active set does not contain this ID.
25    UnknownRegion(RegionId),
26    /// The requested runtime authority does not match the committed direction.
27    WrongDirection(RegionId),
28    /// Committed mappings do not exactly match pending IDs and directions.
29    CommitMismatch,
30}
31
32impl core::fmt::Display for BatchError {
33    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
34        write!(formatter, "region batch operation failed: {self:?}")
35    }
36}
37
38impl std::error::Error for BatchError {}
39
40/// Consuming owner of every prepared object in one transfer transaction.
41pub struct TransferBatch {
42    regions: Vec<PreparedRegion>,
43    max_regions: usize,
44    max_region_bytes: u64,
45    max_batch_bytes: u64,
46    total_logical: u64,
47    total_mapped: u64,
48}
49
50/// Receiver-owned coordinator-relative metadata fixed before capability I/O.
51#[derive(Clone, Copy, Debug, Eq, PartialEq)]
52pub struct ExpectedRegion {
53    pub(crate) id: RegionId,
54    pub(crate) writer: WriterEndpoint,
55    pub(crate) logical_len: usize,
56}
57
58/// Canonical receiver expectation that contains no coordinator-minted
59/// incarnation, native object, mapping, or transaction authority.
60pub struct ExpectedBatch {
61    pub(crate) regions: Vec<ExpectedRegion>,
62    pub(crate) total_logical: u64,
63}
64
65impl ExpectedRegion {
66    /// Fixes the expected identity, coordinator-relative writer, and length.
67    pub const fn new(id: RegionId, writer: WriterEndpoint, logical_len: usize) -> Self {
68        Self {
69            id,
70            writer,
71            logical_len,
72        }
73    }
74
75    /// Expected opaque identity.
76    pub const fn id(self) -> RegionId {
77        self.id
78    }
79
80    /// Expected coordinator-relative writer direction.
81    pub const fn writer(self) -> WriterEndpoint {
82        self.writer
83    }
84
85    /// Expected logical application-visible length.
86    pub const fn logical_len(self) -> usize {
87        self.logical_len
88    }
89}
90
91impl ExpectedBatch {
92    /// Validates a complete receiver expectation before any capability I/O.
93    pub fn try_from_regions(mut regions: Vec<ExpectedRegion>) -> Result<Self, BatchError> {
94        if regions.is_empty() {
95            return Err(BatchError::Empty);
96        }
97        if regions.len() > 16 {
98            return Err(BatchError::TooManyRegions);
99        }
100        regions.sort_unstable_by_key(|region| region.id);
101        if let Some(duplicate) = regions
102            .windows(2)
103            .find(|pair| pair[0].id == pair[1].id)
104            .map(|pair| pair[0].id)
105        {
106            return Err(BatchError::DuplicateRegionId(duplicate));
107        }
108        let total_logical = regions.iter().try_fold(0_u64, |total, region| {
109            let logical = u64::try_from(region.logical_len)
110                .ok()
111                .filter(|logical| *logical != 0)
112                .ok_or(BatchError::InvalidRegionLength)?;
113            total
114                .checked_add(logical)
115                .ok_or(BatchError::BatchBytesExceeded)
116        })?;
117        Ok(Self {
118            regions,
119            total_logical,
120        })
121    }
122
123    /// Number of expected regions.
124    pub fn len(&self) -> usize {
125        self.regions.len()
126    }
127
128    /// Whether the expectation is empty (always false after construction).
129    pub fn is_empty(&self) -> bool {
130        self.regions.is_empty()
131    }
132
133    pub(crate) fn try_from_specs(regions: Vec<ExpectedRegion>) -> Result<Self, BatchError> {
134        Self::try_from_regions(regions)
135    }
136}
137
138impl TransferBatch {
139    #[allow(dead_code)]
140    pub(crate) fn new(
141        max_regions: u16,
142        max_region_bytes: u64,
143        max_batch_bytes: u64,
144    ) -> Result<Self, BatchError> {
145        let max_regions = usize::from(max_regions);
146        if max_regions == 0 || max_regions > 16 {
147            return Err(BatchError::TooManyRegions);
148        }
149        if max_region_bytes == 0 || max_batch_bytes == 0 {
150            return Err(BatchError::InvalidLimits);
151        }
152        Ok(Self {
153            regions: Vec::with_capacity(max_regions),
154            max_regions,
155            max_region_bytes,
156            max_batch_bytes,
157            total_logical: 0,
158            total_mapped: 0,
159        })
160    }
161
162    /// Consumes one prepared region into this transaction.
163    pub fn add(&mut self, region: PreparedRegion) -> Result<(), BatchError> {
164        if self.regions.len() == self.max_regions {
165            return Err(BatchError::TooManyRegions);
166        }
167        let id = region.spec().id;
168        if self.regions.iter().any(|existing| existing.spec().id == id) {
169            return Err(BatchError::DuplicateRegionId(id));
170        }
171        let logical =
172            u64::try_from(region.logical_len()).map_err(|_| BatchError::BatchBytesExceeded)?;
173        let mapped =
174            u64::try_from(region.mapped_len()).map_err(|_| BatchError::BatchBytesExceeded)?;
175        if logical > self.max_region_bytes {
176            return Err(BatchError::InvalidRegionLength);
177        }
178        let total_logical = self
179            .total_logical
180            .checked_add(logical)
181            .ok_or(BatchError::BatchBytesExceeded)?;
182        let total_mapped = self
183            .total_mapped
184            .checked_add(mapped)
185            .ok_or(BatchError::BatchBytesExceeded)?;
186        if total_logical > self.max_batch_bytes || total_mapped > self.max_batch_bytes {
187            return Err(BatchError::BatchBytesExceeded);
188        }
189        self.total_logical = total_logical;
190        self.total_mapped = total_mapped;
191        self.regions.push(region);
192        Ok(())
193    }
194
195    /// Current number of transaction-owned regions.
196    pub fn len(&self) -> usize {
197        self.regions.len()
198    }
199
200    /// Whether no region has been added yet.
201    pub fn is_empty(&self) -> bool {
202        self.regions.is_empty()
203    }
204
205    pub(crate) fn reservation_lengths(&self) -> Vec<u64> {
206        let mut lengths = self
207            .regions
208            .iter()
209            .map(|region| {
210                (
211                    region.spec().id,
212                    u64::try_from(region.mapped_len()).expect("prepared length is native"),
213                )
214            })
215            .collect::<Vec<_>>();
216        lengths.sort_unstable_by_key(|(id, _)| *id);
217        lengths.into_iter().map(|(_, length)| length).collect()
218    }
219
220    pub(crate) fn manifest_entries(&self) -> Option<Vec<ManifestEntry>> {
221        self.regions
222            .iter()
223            .map(|region| {
224                let access = match region.spec().writer {
225                    WriterEndpoint::Coordinator => PeerAccess::ReadOnly,
226                    WriterEndpoint::Receiver => PeerAccess::SoleWriter,
227                };
228                Some(ManifestEntry::from_native(
229                    region.request.native_spec(region.spec().id.get())?,
230                    access,
231                ))
232            })
233            .collect()
234    }
235
236    #[allow(dead_code)]
237    pub(crate) fn into_pending(self) -> Result<PendingBatch, BatchError> {
238        if self.regions.is_empty() {
239            return Err(BatchError::Empty);
240        }
241        Ok(PendingBatch {
242            regions: self.regions,
243            total_logical: self.total_logical,
244            total_mapped: self.total_mapped,
245        })
246    }
247}
248
249/// Complete transaction state after local validation and before COMMIT.
250#[allow(dead_code)]
251pub(crate) struct PendingBatch {
252    pub(crate) regions: Vec<PreparedRegion>,
253    pub(crate) total_logical: u64,
254    pub(crate) total_mapped: u64,
255}
256
257impl PendingBatch {
258    pub(crate) fn manifest_entries(&self) -> Option<Vec<ManifestEntry>> {
259        self.regions
260            .iter()
261            .map(|region| {
262                let access = match region.spec().writer {
263                    WriterEndpoint::Coordinator => PeerAccess::ReadOnly,
264                    WriterEndpoint::Receiver => PeerAccess::SoleWriter,
265                };
266                Some(ManifestEntry::from_native(
267                    region.request.native_spec(region.spec().id.get())?,
268                    access,
269                ))
270            })
271            .collect()
272    }
273}
274
275#[allow(dead_code)]
276pub(crate) enum CommittedRegion {
277    Reader(ActiveReader),
278    Writer(ActiveWriter),
279}
280
281/// Endpoint-local authority expected for one committed native mapping.
282#[derive(Clone, Copy, Debug, Eq, PartialEq)]
283pub(crate) enum LocalRegionAuthority {
284    Reader,
285    Writer,
286}
287
288/// Keyed complete runtime set returned only after batch COMMIT.
289pub struct ActiveRegionSet {
290    regions: BTreeMap<RegionId, CommittedRegion>,
291}
292
293impl ActiveRegionSet {
294    #[allow(dead_code)]
295    pub(crate) fn from_committed(
296        pending: PendingBatch,
297        regions: impl IntoIterator<Item = (RegionId, CommittedRegion)>,
298    ) -> Result<Self, BatchError> {
299        let expected = pending.regions.iter().map(|region| {
300            let authority = match region.spec().writer {
301                WriterEndpoint::Coordinator => LocalRegionAuthority::Writer,
302                WriterEndpoint::Receiver => LocalRegionAuthority::Reader,
303            };
304            (region.spec().id, authority)
305        });
306        let result = Self::from_local_committed(expected, regions);
307        drop(pending);
308        result
309    }
310
311    pub(crate) fn from_local_committed(
312        expected: impl IntoIterator<Item = (RegionId, LocalRegionAuthority)>,
313        regions: impl IntoIterator<Item = (RegionId, CommittedRegion)>,
314    ) -> Result<Self, BatchError> {
315        let mut expected = expected.into_iter().collect::<Vec<_>>();
316        if expected.len() > 16 {
317            return Err(BatchError::CommitMismatch);
318        }
319        expected.sort_unstable_by_key(|(id, _)| *id);
320        if let Some(duplicate) = expected
321            .windows(2)
322            .find(|pair| pair[0].0 == pair[1].0)
323            .map(|pair| pair[0].0)
324        {
325            return Err(BatchError::DuplicateRegionId(duplicate));
326        }
327        let mut keyed = BTreeMap::new();
328        for (id, region) in regions {
329            if keyed.insert(id, region).is_some() {
330                return Err(BatchError::DuplicateRegionId(id));
331            }
332        }
333        if keyed.len() != expected.len() || keyed.len() > 16 {
334            return Err(BatchError::CommitMismatch);
335        }
336        for (id, authority) in expected {
337            match (authority, keyed.get(&id)) {
338                (LocalRegionAuthority::Writer, Some(CommittedRegion::Writer(_)))
339                | (LocalRegionAuthority::Reader, Some(CommittedRegion::Reader(_))) => {}
340                _ => return Err(BatchError::CommitMismatch),
341            }
342        }
343        Ok(Self { regions: keyed })
344    }
345
346    /// Number of active regions still retained by this set.
347    pub fn len(&self) -> usize {
348        self.regions.len()
349    }
350
351    /// Whether every region has been removed.
352    pub fn is_empty(&self) -> bool {
353        self.regions.is_empty()
354    }
355
356    /// Removes the endpoint-local writable mapping for `id`.
357    pub fn take_writer(&mut self, id: RegionId) -> Result<ActiveWriter, BatchError> {
358        match self.regions.get(&id) {
359            None => return Err(BatchError::UnknownRegion(id)),
360            Some(CommittedRegion::Reader(_)) => return Err(BatchError::WrongDirection(id)),
361            Some(CommittedRegion::Writer(_)) => {}
362        }
363        match self.regions.remove(&id) {
364            Some(CommittedRegion::Writer(writer)) => Ok(writer),
365            Some(CommittedRegion::Reader(_)) => Err(BatchError::WrongDirection(id)),
366            None => Err(BatchError::UnknownRegion(id)),
367        }
368    }
369
370    /// Removes the endpoint-local read-only mapping for `id`.
371    pub fn take_reader(&mut self, id: RegionId) -> Result<ActiveReader, BatchError> {
372        match self.regions.get(&id) {
373            None => return Err(BatchError::UnknownRegion(id)),
374            Some(CommittedRegion::Writer(_)) => return Err(BatchError::WrongDirection(id)),
375            Some(CommittedRegion::Reader(_)) => {}
376        }
377        match self.regions.remove(&id) {
378            Some(CommittedRegion::Reader(reader)) => Ok(reader),
379            Some(CommittedRegion::Writer(_)) => Err(BatchError::WrongDirection(id)),
380            None => Err(BatchError::UnknownRegion(id)),
381        }
382    }
383}
384
385#[cfg(test)]
386#[path = "batch_test.rs"]
387mod tests;
native_ipc/batch.rs

native_ipc/
batch.rs
