behavior_actors/atomic/keyed_pool/
binding.rs

1//! Binding capacity and generation-exact application evidence.
2
3use core::cmp::Ordering;
4use core::num::NonZeroU64;
5use std::sync::Arc;
6
7use super::super::RoleName;
8use super::super::capacity::{PositiveCapacity, ZeroCapacity};
9
10/// Positive maximum number of retained key bindings.
11#[derive(Clone, Copy, Debug, Eq, PartialEq)]
12pub struct BindingCapacity {
13    maximum: PositiveCapacity,
14}
15
16impl BindingCapacity {
17    /// Validate the maximum number of retained bindings.
18    ///
19    /// # Errors
20    ///
21    /// Returns [`ZeroCapacity`] when `maximum` is zero.
22    pub fn new(maximum: usize) -> Result<Self, ZeroCapacity> {
23        PositiveCapacity::new(maximum).map(|maximum| Self { maximum })
24    }
25
26    const fn maximum(self) -> usize {
27        self.maximum.get()
28    }
29}
30
31/// Opaque non-reused identity of one retained key binding.
32///
33/// Applications receive generations from pool outcomes and cannot mint them:
34///
35/// ```compile_fail,E0599
36/// let _ = behavior_actors::atomic::BindingGeneration::new(1);
37/// ```
38#[derive(Clone)]
39pub struct BindingGeneration {
40    ordinal: NonZeroU64,
41    token: Arc<()>,
42}
43
44impl BindingGeneration {
45    fn issued(ordinal: NonZeroU64, token: Arc<()>) -> Self {
46        Self { ordinal, token }
47    }
48
49    /// Inspect the pool-local ordinal without gaining construction authority.
50    #[must_use]
51    pub const fn get(&self) -> u64 {
52        self.ordinal.get()
53    }
54}
55
56impl core::fmt::Debug for BindingGeneration {
57    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
58        formatter
59            .debug_tuple("BindingGeneration")
60            .field(&self.ordinal)
61            .finish()
62    }
63}
64
65impl PartialEq for BindingGeneration {
66    fn eq(&self, other: &Self) -> bool {
67        self.ordinal == other.ordinal && Arc::ptr_eq(&self.token, &other.token)
68    }
69}
70
71impl Eq for BindingGeneration {}
72
73/// Exact binding state expected by one rebalance or unbind command.
74#[derive(Clone, Debug, Eq, PartialEq)]
75pub enum BindingExpectation {
76    /// The key must have no retained binding.
77    Absent,
78    /// The key must retain this exact generation.
79    Exact(BindingGeneration),
80}
81
82/// Non-authorizing role and generation evidence carried by accepted work.
83///
84/// The value intentionally contains no key. Cloning it cannot create a binding
85/// or extend key retention.
86pub struct BindingEvidence<Role> {
87    generation: BindingGeneration,
88    role: RoleName<Role>,
89}
90
91impl<Role> BindingEvidence<Role> {
92    fn issued(generation: BindingGeneration, role: RoleName<Role>) -> Self {
93        Self { generation, role }
94    }
95
96    /// Borrow the opaque generation accepted for this work.
97    #[must_use]
98    pub const fn generation(&self) -> &BindingGeneration {
99        &self.generation
100    }
101
102    /// Borrow the immutable semantic role selected at admission.
103    #[must_use]
104    pub fn role(&self) -> &Role {
105        self.role.role()
106    }
107}
108
109impl<Role> Clone for BindingEvidence<Role> {
110    fn clone(&self) -> Self {
111        Self {
112            generation: self.generation.clone(),
113            role: self.role.clone(),
114        }
115    }
116}
117
118impl<Role> core::fmt::Debug for BindingEvidence<Role>
119where
120    Role: core::fmt::Debug,
121{
122    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
123        formatter
124            .debug_struct("BindingEvidence")
125            .field("generation", &self.generation)
126            .field("role", self.role())
127            .finish()
128    }
129}
130
131/// Caller-authored correlation echoed by one binding-management reply.
132#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
133pub struct BindingRequestId(u64);
134
135impl BindingRequestId {
136    /// Name one rebalance or unbind command.
137    #[must_use]
138    pub const fn new(value: u64) -> Self {
139        Self(value)
140    }
141
142    /// Inspect the caller-authored value.
143    #[must_use]
144    pub const fn get(self) -> u64 {
145        self.0
146    }
147}
148
149struct Binding<Key, Role> {
150    key: Key,
151    evidence: BindingEvidence<Role>,
152}
153
154impl<Key, Role> Binding<Key, Role> {
155    fn into_parts(self) -> (Key, BindingEvidence<Role>) {
156        (self.key, self.evidence)
157    }
158}
159
160struct GenerationSequence {
161    next: Option<NonZeroU64>,
162    token: Arc<()>,
163}
164
165impl GenerationSequence {
166    fn new() -> Self {
167        Self {
168            next: Some(NonZeroU64::MIN),
169            token: Arc::new(()),
170        }
171    }
172
173    fn issue(&mut self) -> Option<BindingGeneration> {
174        let ordinal = self.next?;
175        self.next = ordinal.checked_add(1);
176        Some(BindingGeneration::issued(ordinal, Arc::clone(&self.token)))
177    }
178}
179
180pub(super) enum BindingReservationRejected<Key> {
181    AlreadyBound(Key),
182    CapacityExhausted(Key),
183    GenerationsExhausted(Key),
184}
185
186pub(super) struct BindingTable<Key, Role> {
187    capacity: BindingCapacity,
188    bindings: Vec<Binding<Key, Role>>,
189    generations: GenerationSequence,
190}
191
192impl<Key, Role> BindingTable<Key, Role>
193where
194    Key: Ord,
195{
196    pub(super) fn new(capacity: BindingCapacity) -> Self {
197        Self {
198            capacity,
199            bindings: Vec::new(),
200            generations: GenerationSequence::new(),
201        }
202    }
203
204    pub(super) fn binding(&self, key: &Key) -> Option<&BindingEvidence<Role>> {
205        self.position(key)
206            .ok()
207            .map(|position| &self.bindings[position].evidence)
208    }
209
210    pub(super) fn occupied(&mut self, key: &Key) -> Option<OccupiedBinding<'_, Key, Role>> {
211        let position = self.position(key).ok()?;
212        Some(OccupiedBinding {
213            binding: &mut self.bindings[position],
214            generations: &mut self.generations,
215        })
216    }
217
218    pub(super) fn reserve(
219        &mut self,
220        key: Key,
221        role: RoleName<Role>,
222    ) -> Result<ReservedBinding<'_, Key, Role>, BindingReservationRejected<Key>> {
223        let position = match self.position(&key) {
224            Ok(_) => return Err(BindingReservationRejected::AlreadyBound(key)),
225            Err(position) => position,
226        };
227        match self.bindings.len().cmp(&self.capacity.maximum()) {
228            Ordering::Less => {}
229            Ordering::Equal | Ordering::Greater => {
230                return Err(BindingReservationRejected::CapacityExhausted(key));
231            }
232        }
233        let generation = match self.generations.issue() {
234            Some(generation) => generation,
235            None => return Err(BindingReservationRejected::GenerationsExhausted(key)),
236        };
237        Ok(ReservedBinding {
238            bindings: &mut self.bindings,
239            position,
240            key,
241            generation,
242            role,
243        })
244    }
245
246    pub(super) fn remove(&mut self, key: &Key) -> Option<(Key, BindingEvidence<Role>)> {
247        self.position(key)
248            .ok()
249            .map(|position| self.bindings.remove(position).into_parts())
250    }
251
252    pub(super) fn retire_role(&mut self, role: &Role) -> Vec<(Key, BindingEvidence<Role>)>
253    where
254        Role: Eq,
255    {
256        self.bindings
257            .extract_if(.., |binding| binding.evidence.role() == role)
258            .map(Binding::into_parts)
259            .collect()
260    }
261
262    pub(super) fn drain(&mut self) -> Vec<(Key, BindingEvidence<Role>)> {
263        self.bindings.drain(..).map(Binding::into_parts).collect()
264    }
265
266    fn position(&self, key: &Key) -> Result<usize, usize> {
267        self.bindings
268            .binary_search_by(|binding| binding.key.cmp(key))
269    }
270}
271
272#[must_use = "a reserved binding must commit or return its key"]
273pub(super) struct ReservedBinding<'table, Key, Role> {
274    bindings: &'table mut Vec<Binding<Key, Role>>,
275    position: usize,
276    key: Key,
277    generation: BindingGeneration,
278    role: RoleName<Role>,
279}
280
281pub(super) struct OccupiedBinding<'table, Key, Role> {
282    binding: &'table mut Binding<Key, Role>,
283    generations: &'table mut GenerationSequence,
284}
285
286impl<Key, Role> OccupiedBinding<'_, Key, Role> {
287    pub(super) fn rebind(
288        self,
289        role: RoleName<Role>,
290    ) -> Option<(BindingEvidence<Role>, BindingEvidence<Role>)> {
291        let generation = self.generations.issue()?;
292        let current = BindingEvidence::issued(generation, role);
293        let admitted = current.clone();
294        let prior = core::mem::replace(&mut self.binding.evidence, current);
295        Some((prior, admitted))
296    }
297}
298
299impl<Key, Role> ReservedBinding<'_, Key, Role> {
300    pub(super) fn commit(self) -> BindingEvidence<Role> {
301        let evidence = BindingEvidence::issued(self.generation, self.role);
302        let admitted = evidence.clone();
303        self.bindings.insert(
304            self.position,
305            Binding {
306                key: self.key,
307                evidence,
308            },
309        );
310        admitted
311    }
312
313    pub(super) fn reject(self) -> Key {
314        self.key
315    }
316}
317
318#[cfg(test)]
319mod tests {
320    use core::num::NonZeroU64;
321    use std::sync::Arc;
322
323    use super::{
324        BindingCapacity, BindingEvidence, BindingGeneration, BindingReservationRejected,
325        BindingTable, GenerationSequence,
326    };
327    use crate::atomic::RoleName;
328
329    #[derive(Debug, Eq, Ord, PartialEq, PartialOrd)]
330    struct Account(u8);
331
332    #[derive(Debug, Eq, PartialEq)]
333    enum SearchRole {
334        Primary,
335        Replica,
336    }
337
338    fn bindings(maximum: usize) -> BindingTable<Account, SearchRole> {
339        let capacity = BindingCapacity::new(maximum)
340            .unwrap_or_else(|_| panic!("test binding capacity must be positive"));
341        BindingTable::new(capacity)
342    }
343
344    fn commit(
345        table: &mut BindingTable<Account, SearchRole>,
346        account: Account,
347        role: SearchRole,
348    ) -> BindingEvidence<SearchRole> {
349        table
350            .reserve(account, RoleName::new(role))
351            .unwrap_or_else(|_| panic!("test binding reservation must succeed"))
352            .commit()
353    }
354
355    fn first_generation() -> BindingGeneration {
356        BindingGeneration {
357            ordinal: NonZeroU64::MIN,
358            token: Arc::new(()),
359        }
360    }
361
362    #[test]
363    fn independent_pools_cannot_share_a_numeric_generation() {
364        let left = first_generation();
365        let right = first_generation();
366
367        assert_eq!(left.get(), right.get());
368        assert_ne!(left, right);
369    }
370
371    #[test]
372    fn evidence_clones_identity_without_owning_a_key() {
373        let evidence = BindingEvidence {
374            generation: first_generation(),
375            role: RoleName::new(String::from("search")),
376        };
377        let cloned = evidence.clone();
378
379        assert_eq!(evidence.generation(), cloned.generation());
380        assert_eq!(evidence.role(), "search");
381        assert_eq!(cloned.role(), "search");
382    }
383
384    #[test]
385    fn table_preserves_order_and_returns_rejected_keys_without_clone() {
386        let mut table = bindings(2);
387        let second = commit(&mut table, Account(2), SearchRole::Replica);
388        let first = commit(&mut table, Account(1), SearchRole::Primary);
389
390        let found_first = table
391            .binding(&Account(1))
392            .unwrap_or_else(|| panic!("first binding must remain"));
393        let found_second = table
394            .binding(&Account(2))
395            .unwrap_or_else(|| panic!("second binding must remain"));
396        assert_eq!(found_first.generation(), first.generation());
397        assert_eq!(found_first.role(), first.role());
398        assert_eq!(found_second.generation(), second.generation());
399        assert_eq!(found_second.role(), second.role());
400
401        let duplicate = table
402            .reserve(Account(1), RoleName::new(SearchRole::Replica))
403            .err()
404            .unwrap_or_else(|| panic!("duplicate key must be rejected"));
405        assert!(matches!(
406            duplicate,
407            BindingReservationRejected::AlreadyBound(Account(1))
408        ));
409
410        let full = table
411            .reserve(Account(3), RoleName::new(SearchRole::Primary))
412            .err()
413            .unwrap_or_else(|| panic!("full table must reject the owned key"));
414        assert!(matches!(
415            full,
416            BindingReservationRejected::CapacityExhausted(Account(3))
417        ));
418
419        let drained = table.drain();
420        assert_eq!(
421            drained
422                .iter()
423                .map(|(account, _)| account.0)
424                .collect::<Vec<_>>(),
425            vec![1, 2]
426        );
427    }
428
429    #[test]
430    fn rejected_admission_burns_its_generation_without_retaining_the_key() {
431        let mut table = bindings(2);
432        let rejected_key = table
433            .reserve(Account(1), RoleName::new(SearchRole::Primary))
434            .unwrap_or_else(|_| panic!("first reservation must succeed"))
435            .reject();
436        assert_eq!(rejected_key, Account(1));
437        assert!(matches!(table.binding(&Account(1)), None));
438
439        let evidence = commit(&mut table, Account(1), SearchRole::Primary);
440        assert_eq!(evidence.generation().get(), 2);
441    }
442
443    #[test]
444    fn generation_exhaustion_never_wraps_or_loses_the_key() {
445        let maximum = NonZeroU64::new(u64::MAX)
446            .unwrap_or_else(|| panic!("the maximum unsigned value is non-zero"));
447        let mut sequence = GenerationSequence {
448            next: Some(maximum),
449            token: Arc::new(()),
450        };
451        let final_generation = sequence
452            .issue()
453            .unwrap_or_else(|| panic!("the last generation remains issuable"));
454        assert_eq!(final_generation.get(), u64::MAX);
455        let exhausted_generation = sequence.issue();
456        assert!(matches!(exhausted_generation, None));
457
458        let mut table = bindings(1);
459        table.generations.next = None;
460        let rejection = table
461            .reserve(Account(9), RoleName::new(SearchRole::Primary))
462            .err()
463            .unwrap_or_else(|| panic!("exhaustion must reject the reservation"));
464        assert!(matches!(
465            rejection,
466            BindingReservationRejected::GenerationsExhausted(Account(9))
467        ));
468    }
469
470    #[test]
471    fn role_retirement_and_unbind_return_complete_owned_bindings() {
472        let mut table = bindings(3);
473        commit(&mut table, Account(3), SearchRole::Primary);
474        commit(&mut table, Account(1), SearchRole::Replica);
475        commit(&mut table, Account(2), SearchRole::Primary);
476
477        let retired = table.retire_role(&SearchRole::Primary);
478        assert_eq!(
479            retired
480                .iter()
481                .map(|(account, _)| account.0)
482                .collect::<Vec<_>>(),
483            vec![2, 3]
484        );
485        assert!(matches!(table.binding(&Account(2)), None));
486        assert!(matches!(table.binding(&Account(3)), None));
487
488        let (account, evidence) = table
489            .remove(&Account(1))
490            .unwrap_or_else(|| panic!("remaining binding must be removable"));
491        assert_eq!(account, Account(1));
492        assert_eq!(evidence.role(), &SearchRole::Replica);
493        assert!(matches!(table.binding(&Account(1)), None));
494    }
495
496    #[test]
497    fn role_change_commits_once_without_removing_the_binding() {
498        let mut table = bindings(1);
499        let original = commit(&mut table, Account(1), SearchRole::Primary);
500
501        let (prior, current) = table
502            .occupied(&Account(1))
503            .unwrap_or_else(|| panic!("the binding must remain occupied"))
504            .rebind(RoleName::new(SearchRole::Replica))
505            .unwrap_or_else(|| panic!("a fresh generation must remain"));
506
507        assert_eq!(prior.generation(), original.generation());
508        assert_eq!(prior.role(), &SearchRole::Primary);
509        assert_eq!(current.generation().get(), 2);
510        assert_eq!(current.role(), &SearchRole::Replica);
511        let retained = table
512            .binding(&Account(1))
513            .unwrap_or_else(|| panic!("replacement cannot expose absence"));
514        assert_eq!(retained.generation(), current.generation());
515        assert_eq!(retained.role(), current.role());
516    }
517
518    #[test]
519    fn exhausted_role_change_preserves_the_current_binding() {
520        let mut table = bindings(1);
521        let original = commit(&mut table, Account(1), SearchRole::Primary);
522        table.generations.next = None;
523
524        let replacement = table
525            .occupied(&Account(1))
526            .unwrap_or_else(|| panic!("the binding must remain occupied"))
527            .rebind(RoleName::new(SearchRole::Replica));
528        assert!(matches!(replacement, None));
529
530        let retained = table
531            .binding(&Account(1))
532            .unwrap_or_else(|| panic!("exhaustion cannot remove the binding"));
533        assert_eq!(retained.generation(), original.generation());
534        assert_eq!(retained.role(), &SearchRole::Primary);
535    }
536}
behavior_actors/atomic/keyed_pool/binding.rs

behavior_actors/atomic/keyed_pool/
binding.rs
