cloud_sdk/operation/permit/
shared.rs

1//! Explicitly shareable permits backed by caller-owned atomic state.
2
3use core::sync::atomic::{AtomicU32, Ordering};
4
5use super::state::{AttemptPhase, PermitAttempt};
6use super::{
7    ExecutionPermitError, PermitDisposition, PermitIdempotencyKey, PermitScope, PermitState,
8    PermitTimestamp, PlanSubject, ReconciliationToken, RecoveryToken, ReplayPolicy,
9};
10
11const STATE_MASK: u32 = 0b111;
12const GENERATION_SHIFT: u32 = 3;
13const GENERATION_MASK: u32 = 0x1fff;
14const REMAINING_SHIFT: u32 = 16;
15
16const READY: u32 = 0;
17const IN_FLIGHT: u32 = 1;
18const RECOVERABLE: u32 = 2;
19const PENDING: u32 = 3;
20const SPENT: u32 = 4;
21
22/// Caller-owned atomic authority shared by all explicit permit-handle clones.
23///
24/// The state is initialized only through a shared permit constructor taking
25/// `&mut SharedPermitState`. That exclusive borrow prevents safe code from
26/// binding two independent plans to the same atomic state concurrently.
27pub struct SharedPermitState {
28    packed: AtomicU32,
29    last_offset: AtomicU32,
30}
31
32impl SharedPermitState {
33    /// Creates unbound spent state ready for one exclusive plan binding.
34    #[must_use]
35    pub fn new() -> Self {
36        Self {
37            packed: AtomicU32::new(pack(SPENT, 0, 0)),
38            last_offset: AtomicU32::new(0),
39        }
40    }
41
42    /// Returns the current atomic lifecycle state.
43    #[must_use]
44    pub fn state(&self) -> PermitState {
45        unpack_state(self.packed.load(Ordering::Acquire))
46    }
47
48    fn initialize(
49        &mut self,
50        subject: PlanSubject<'_, '_>,
51        now: PermitTimestamp,
52    ) -> Result<(), ExecutionPermitError> {
53        let offset = subject.validity().offset(now)?;
54        *self.packed.get_mut() = pack(READY, 0, subject.attempt_budget().get());
55        *self.last_offset.get_mut() = offset;
56        Ok(())
57    }
58
59    fn begin(
60        &self,
61        subject: PlanSubject<'_, '_>,
62        now: PermitTimestamp,
63    ) -> Result<u16, ExecutionPermitError> {
64        self.observe(subject, now)?;
65        loop {
66            let current = self.packed.load(Ordering::Acquire);
67            let (state, generation, remaining) = unpack(current);
68            match state {
69                READY if remaining != 0 => {}
70                READY | SPENT => return Err(ExecutionPermitError::Spent),
71                IN_FLIGHT => return Err(ExecutionPermitError::AttemptInFlight),
72                RECOVERABLE => return Err(ExecutionPermitError::RecoveryRequired),
73                PENDING => return Err(ExecutionPermitError::ReconciliationRequired),
74                _ => return Err(ExecutionPermitError::Spent),
75            }
76            let next_remaining = remaining
77                .checked_sub(1)
78                .ok_or(ExecutionPermitError::Spent)?;
79            let next = pack(IN_FLIGHT, generation, next_remaining);
80            if self
81                .packed
82                .compare_exchange(current, next, Ordering::AcqRel, Ordering::Acquire)
83                .is_ok()
84            {
85                return Ok(generation);
86            }
87        }
88    }
89
90    pub(super) fn complete(
91        &self,
92        expected_generation: u16,
93        phase: AttemptPhase,
94    ) -> PermitDisposition {
95        loop {
96            let current = self.packed.load(Ordering::Acquire);
97            let (state, generation, remaining) = unpack(current);
98            if state != IN_FLIGHT || generation != expected_generation {
99                return PermitDisposition::Spent;
100            }
101            let (next_state, disposition) = match phase {
102                AttemptPhase::Applied | AttemptPhase::Rejected => (SPENT, PermitDisposition::Spent),
103                AttemptPhase::NotSent if remaining == 0 => (SPENT, PermitDisposition::Spent),
104                AttemptPhase::NotSent => (
105                    RECOVERABLE,
106                    PermitDisposition::Recoverable(RecoveryToken(generation)),
107                ),
108                AttemptPhase::Uncertain => (
109                    PENDING,
110                    PermitDisposition::PendingReconciliation(ReconciliationToken(generation)),
111                ),
112            };
113            let next = pack(next_state, generation, remaining);
114            if self
115                .packed
116                .compare_exchange(current, next, Ordering::AcqRel, Ordering::Acquire)
117                .is_ok()
118            {
119                return disposition;
120            }
121        }
122    }
123
124    fn recover_not_sent(
125        &self,
126        subject: PlanSubject<'_, '_>,
127        token: RecoveryToken,
128        now: PermitTimestamp,
129    ) -> Result<(), ExecutionPermitError> {
130        self.observe(subject, now)?;
131        if subject.replay_policy() == ReplayPolicy::SingleAttempt {
132            return Err(ExecutionPermitError::ReplayForbidden);
133        }
134        self.rearm(RECOVERABLE, token.0)
135    }
136
137    fn reconcile_not_applied(
138        &self,
139        bound: PlanSubject<'_, '_>,
140        candidate: PlanSubject<'_, '_>,
141        token: ReconciliationToken,
142        idempotency: PermitIdempotencyKey<'_>,
143        now: PermitTimestamp,
144    ) -> Result<(), ExecutionPermitError> {
145        self.observe(bound, now)?;
146        if bound.replay_policy() != ReplayPolicy::ReconcileThenRetry {
147            return Err(ExecutionPermitError::ReplayForbidden);
148        }
149        if !bound.fingerprint().matches(candidate.fingerprint()) {
150            return Err(ExecutionPermitError::FingerprintMismatch);
151        }
152        if !bound
153            .idempotency()
154            .is_some_and(|expected| expected.matches(idempotency))
155        {
156            return Err(ExecutionPermitError::IdempotencyMismatch);
157        }
158        self.rearm(PENDING, token.0)
159    }
160
161    fn rearm(
162        &self,
163        required_state: u32,
164        expected_generation: u16,
165    ) -> Result<(), ExecutionPermitError> {
166        loop {
167            let current = self.packed.load(Ordering::Acquire);
168            let (state, generation, remaining) = unpack(current);
169            if state != required_state || generation != expected_generation {
170                return Err(ExecutionPermitError::StaleGeneration);
171            }
172            if remaining == 0 {
173                let _ = self.packed.compare_exchange(
174                    current,
175                    pack(SPENT, generation, 0),
176                    Ordering::AcqRel,
177                    Ordering::Acquire,
178                );
179                return Err(ExecutionPermitError::Spent);
180            }
181            let Some(next_generation) = generation
182                .checked_add(1)
183                .filter(|value| u32::from(*value) <= GENERATION_MASK)
184            else {
185                if self
186                    .packed
187                    .compare_exchange(
188                        current,
189                        pack(SPENT, generation, 0),
190                        Ordering::AcqRel,
191                        Ordering::Acquire,
192                    )
193                    .is_ok()
194                {
195                    return Err(ExecutionPermitError::GenerationExhausted);
196                }
197                continue;
198            };
199            let next = pack(READY, next_generation, remaining);
200            if self
201                .packed
202                .compare_exchange(current, next, Ordering::AcqRel, Ordering::Acquire)
203                .is_ok()
204            {
205                return Ok(());
206            }
207        }
208    }
209
210    // `try_update` is unavailable on Rust 1.90; retain its predecessor through MSRV.
211    #[allow(deprecated)]
212    pub(super) fn observe(
213        &self,
214        subject: PlanSubject<'_, '_>,
215        now: PermitTimestamp,
216    ) -> Result<(), ExecutionPermitError> {
217        let offset = match subject.validity().offset(now) {
218            Ok(offset) => offset,
219            Err(error) => {
220                self.spend();
221                return Err(error);
222            }
223        };
224        let result = self
225            .last_offset
226            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |previous| {
227                (offset >= previous).then_some(offset)
228            })
229            .map(|_| ())
230            .map_err(|_| ExecutionPermitError::ClockRollback);
231        if result.is_err() {
232            self.spend();
233        }
234        result
235    }
236
237    pub(super) fn observe_attempt(
238        &self,
239        subject: PlanSubject<'_, '_>,
240        expected_generation: u16,
241        now: PermitTimestamp,
242    ) -> Result<(), ExecutionPermitError> {
243        self.observe(subject, now)?;
244        let (state, generation, _) = unpack(self.packed.load(Ordering::Acquire));
245        if state != IN_FLIGHT || generation != expected_generation {
246            return Err(ExecutionPermitError::Spent);
247        }
248        Ok(())
249    }
250
251    fn spend(&self) {
252        loop {
253            let current = self.packed.load(Ordering::Acquire);
254            let (_, generation, _) = unpack(current);
255            if current == pack(SPENT, generation, 0) {
256                return;
257            }
258            if self
259                .packed
260                .compare_exchange(
261                    current,
262                    pack(SPENT, generation, 0),
263                    Ordering::AcqRel,
264                    Ordering::Acquire,
265                )
266                .is_ok()
267            {
268                return;
269            }
270        }
271    }
272}
273
274impl Default for SharedPermitState {
275    fn default() -> Self {
276        Self::new()
277    }
278}
279
280impl core::fmt::Debug for SharedPermitState {
281    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
282        formatter
283            .debug_struct("SharedPermitState")
284            .field("state", &self.state())
285            .field("authority", &"[redacted]")
286            .finish()
287    }
288}
289
290macro_rules! shared_permit {
291    ($name:ident, $scope:expr, $description:literal) => {
292        #[doc = $description]
293        ///
294        /// Every clone references the same caller-owned atomic state. Cloning
295        /// never creates new budget or an independent recovery generation.
296        pub struct $name<'state, 'request, 'fingerprint> {
297            state: &'state SharedPermitState,
298            subject: PlanSubject<'request, 'fingerprint>,
299        }
300
301        impl<'state, 'request, 'fingerprint> $name<'state, 'request, 'fingerprint> {
302            /// Exclusively binds fresh shared state to one confirmed plan.
303            pub fn new(
304                state: &'state mut SharedPermitState,
305                subject: PlanSubject<'request, 'fingerprint>,
306                now: PermitTimestamp,
307            ) -> Result<Self, ExecutionPermitError> {
308                if subject.scope() != $scope {
309                    return Err(ExecutionPermitError::ScopeMismatch);
310                }
311                state.initialize(subject, now)?;
312                Ok(Self { state, subject })
313            }
314
315            /// Returns the shared lifecycle state.
316            #[must_use]
317            pub fn state(&self) -> PermitState {
318                self.state.state()
319            }
320
321            /// Atomically starts one attempt for the bound plan.
322            pub fn begin(
323                &self,
324                now: PermitTimestamp,
325            ) -> Result<PermitAttempt<'_, 'request, 'fingerprint>, ExecutionPermitError> {
326                let generation = self.state.begin(self.subject, now)?;
327                Ok(PermitAttempt::shared(self.state, self.subject, generation))
328            }
329
330            /// Starts only if the candidate fingerprint matches the bound plan.
331            pub fn begin_for(
332                &self,
333                candidate: PlanSubject<'_, '_>,
334                now: PermitTimestamp,
335            ) -> Result<PermitAttempt<'_, 'request, 'fingerprint>, ExecutionPermitError> {
336                if !self.subject.fingerprint().matches(candidate.fingerprint()) {
337                    return Err(ExecutionPermitError::FingerprintMismatch);
338                }
339                self.begin(now)
340            }
341
342            /// Atomically recovers a generation-matched `NotSent` attempt.
343            pub fn recover_not_sent(
344                &self,
345                token: RecoveryToken,
346                now: PermitTimestamp,
347            ) -> Result<(), ExecutionPermitError> {
348                self.state.recover_not_sent(self.subject, token, now)
349            }
350
351            /// Rearms after caller-performed operation-specific reconciliation.
352            pub fn reconcile_not_applied(
353                &self,
354                token: ReconciliationToken,
355                candidate: PlanSubject<'_, '_>,
356                idempotency: PermitIdempotencyKey<'_>,
357                now: PermitTimestamp,
358            ) -> Result<(), ExecutionPermitError> {
359                self.state
360                    .reconcile_not_applied(self.subject, candidate, token, idempotency, now)
361            }
362        }
363
364        impl Clone for $name<'_, '_, '_> {
365            fn clone(&self) -> Self {
366                Self {
367                    state: self.state,
368                    subject: self.subject,
369                }
370            }
371        }
372
373        impl core::fmt::Debug for $name<'_, '_, '_> {
374            fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
375                formatter
376                    .debug_struct(stringify!($name))
377                    .field("state", &self.state())
378                    .field("plan", &"[redacted]")
379                    .finish()
380            }
381        }
382    };
383}
384
385shared_permit!(
386    SharedMutationPermit,
387    PermitScope::Mutation,
388    "Shared atomic mutation authority."
389);
390shared_permit!(
391    SharedDestructivePermit,
392    PermitScope::Destructive,
393    "Shared atomic destructive authority."
394);
395shared_permit!(
396    SharedCostPermit,
397    PermitScope::Cost,
398    "Shared atomic price-bounded authority."
399);
400
401fn pack(state: u32, generation: u16, remaining: u16) -> u32 {
402    state | (u32::from(generation) << GENERATION_SHIFT) | (u32::from(remaining) << REMAINING_SHIFT)
403}
404
405fn unpack(value: u32) -> (u32, u16, u16) {
406    let generation = u16::try_from((value >> GENERATION_SHIFT) & GENERATION_MASK).unwrap_or(0);
407    let remaining = u16::try_from(value >> REMAINING_SHIFT).unwrap_or(0);
408    (value & STATE_MASK, generation, remaining)
409}
410
411fn unpack_state(value: u32) -> PermitState {
412    match value & STATE_MASK {
413        READY => PermitState::Ready,
414        IN_FLIGHT => PermitState::InFlight,
415        RECOVERABLE => PermitState::Recoverable,
416        PENDING => PermitState::PendingReconciliation,
417        _ => PermitState::Spent,
418    }
419}
cloud_sdk/operation/permit/shared.rs

cloud_sdk/operation/permit/
shared.rs
