riffdb_client_rust/
command.rs

1//! Immutable same-input command submissions and bounded retry state.
2
3use std::error::Error;
4use std::fmt;
5use std::num::NonZeroU32;
6use std::time::Duration;
7
8use riffdb_errors::{ApplicationErrorCode, PublicErrorKind};
9use riffdb_proto::MAX_PROTOCOL_NAME_BYTES;
10use riffdb_proto::v1;
11use riffdb_types::RequestId;
12
13use crate::status::{ClientError, OutcomeUnknown, carries_uncertainty, is_retryable};
14
15/// A caller-selected nonzero bound on total transport submissions.
16///
17/// RiffDB defines no default attempt count in the POC. Same-input recovery is
18/// budgeted here; Overloaded retries apply bounded exponential backoff with
19/// jitter (via a bounded async backoff) to avoid retry storms.
20#[derive(Clone, Copy, Debug, Eq, PartialEq)]
21pub struct AttemptBudget(NonZeroU32);
22
23/// Base delay for Overloaded exponential backoff (doubles each retry).
24const OVERLOADED_BACKOFF_BASE: Duration = Duration::from_millis(25);
25/// Hard cap on Overloaded backoff delay.
26const OVERLOADED_BACKOFF_CAP: Duration = Duration::from_millis(400);
27
28impl AttemptBudget {
29    /// Constructs a total submission bound.
30    #[must_use]
31    pub const fn new(maximum_submissions: u32) -> Option<Self> {
32        match NonZeroU32::new(maximum_submissions) {
33            Some(value) => Some(Self(value)),
34            None => None,
35        }
36    }
37
38    /// Returns the maximum number of transport submissions.
39    #[must_use]
40    pub const fn maximum_submissions(self) -> u32 {
41        self.0.get()
42    }
43}
44
45/// A safe local error while constructing an immutable command submission.
46#[derive(Clone, Copy, Debug, Eq, PartialEq)]
47pub enum CommandShapeError {
48    /// The command name is empty or exceeds the protocol-name bound.
49    InvalidCommandName,
50    /// An explicitly selected contract version is zero.
51    InvalidContractVersion,
52    /// The command input is not a structurally valid record value.
53    InvalidCommandInput,
54}
55
56impl fmt::Display for CommandShapeError {
57    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
58        formatter.write_str(match self {
59            Self::InvalidCommandName => "command name is invalid",
60            Self::InvalidContractVersion => "contract version is invalid",
61            Self::InvalidCommandInput => "command input is invalid",
62        })
63    }
64}
65
66impl Error for CommandShapeError {}
67
68/// One immutable idempotent command used by every automatic retry attempt.
69///
70/// Generic transport code retains the opaque input byte-for-byte but cannot
71/// identify or validate the contract field carrying an idempotency key.
72/// Generated command modules bind their typed key to both execution input and
73/// explicit outcome recovery. The server remains authoritative for schema and
74/// idempotency validation.
75#[derive(Clone, PartialEq)]
76pub struct IdempotentCommand {
77    command_name: String,
78    expected_contract_version: Option<u64>,
79    input: v1::Value,
80}
81
82impl fmt::Debug for IdempotentCommand {
83    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
84        formatter.write_str("IdempotentCommand([REDACTED])")
85    }
86}
87
88impl IdempotentCommand {
89    /// Constructs one immutable generic command submission.
90    pub fn new(
91        command_name: impl Into<String>,
92        expected_contract_version: Option<u64>,
93        input: v1::Value,
94    ) -> Result<Self, CommandShapeError> {
95        let command_name = command_name.into();
96        if command_name.is_empty() || command_name.len() > MAX_PROTOCOL_NAME_BYTES {
97            return Err(CommandShapeError::InvalidCommandName);
98        }
99        if expected_contract_version == Some(0) {
100            return Err(CommandShapeError::InvalidContractVersion);
101        }
102        if riffdb_proto::validate_command_input_value(&input).is_err()
103            || !matches!(input.kind.as_ref(), Some(v1::value::Kind::RecordValue(_)))
104        {
105            return Err(CommandShapeError::InvalidCommandInput);
106        }
107        Ok(Self {
108            command_name,
109            expected_contract_version,
110            input,
111        })
112    }
113
114    /// Borrows the exact command name reused by every attempt.
115    #[must_use]
116    pub fn command_name(&self) -> &str {
117        &self.command_name
118    }
119
120    /// Returns the explicit contract version, when one was selected.
121    #[must_use]
122    pub const fn expected_contract_version(&self) -> Option<u64> {
123        self.expected_contract_version
124    }
125
126    /// Borrows the exact input cloned for every attempt.
127    #[must_use]
128    pub const fn input(&self) -> &v1::Value {
129        &self.input
130    }
131
132    pub(crate) fn request(&self, request_id: RequestId) -> v1::ExecuteCommandRequest {
133        v1::ExecuteCommandRequest {
134            request_id: request_id.into_bytes().to_vec(),
135            command_name: self.command_name.clone(),
136            expected_contract_version: self.expected_contract_version,
137            input: Some(self.input.clone()),
138        }
139    }
140}
141
142pub(crate) struct RetryState {
143    remaining_submissions: u32,
144    unresolved_uncertainty: bool,
145    /// Zero-based index of the next retry delay (after the first failure).
146    overloaded_retry_index: u32,
147    /// Per-request decorrelation seed from the request id's first 8 LE bytes.
148    jitter_seed: u64,
149}
150
151impl RetryState {
152    pub(crate) const fn new(budget: AttemptBudget) -> Self {
153        Self {
154            remaining_submissions: budget.maximum_submissions(),
155            unresolved_uncertainty: false,
156            overloaded_retry_index: 0,
157            jitter_seed: 0,
158        }
159    }
160
161    /// Seeds decorrelated jitter from the current attempt's request identity.
162    pub(crate) fn note_request_id(&mut self, request_id: &RequestId) {
163        let bytes = request_id.as_bytes();
164        self.jitter_seed = u64::from_le_bytes([
165            bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
166        ]);
167    }
168
169    pub(crate) fn begin_submission(&mut self) -> bool {
170        let Some(remaining) = self.remaining_submissions.checked_sub(1) else {
171            return false;
172        };
173        self.remaining_submissions = remaining;
174        true
175    }
176
177    pub(crate) fn request_id_failure(
178        &self,
179        error: crate::IdentifierGenerationError,
180    ) -> ClientError {
181        if self.unresolved_uncertainty {
182            ClientError::OutcomeUnknown(OutcomeUnknown)
183        } else {
184            ClientError::IdentifierGeneration(error)
185        }
186    }
187
188    pub(crate) fn handle_failure(&mut self, error: ClientError) -> RetryDecision {
189        self.unresolved_uncertainty |= carries_uncertainty(&error);
190        if !is_retryable(&error) {
191            return RetryDecision::Return(if self.unresolved_uncertainty {
192                ClientError::OutcomeUnknown(OutcomeUnknown)
193            } else {
194                error
195            });
196        }
197        if self.remaining_submissions > 0 {
198            if is_overloaded(&error) {
199                let delay = overloaded_backoff(self.jitter_seed, self.overloaded_retry_index);
200                self.overloaded_retry_index = self.overloaded_retry_index.saturating_add(1);
201                return RetryDecision::RetryAfter(delay);
202            }
203            return RetryDecision::Retry;
204        }
205        if self.unresolved_uncertainty {
206            RetryDecision::Return(ClientError::OutcomeUnknown(OutcomeUnknown))
207        } else {
208            RetryDecision::Return(error)
209        }
210    }
211}
212
213/// Applies the Overloaded backoff delay without blocking the runtime thread.
214///
215/// Kept out of `client.rs` so architecture boundaries continue to forbid generic
216/// retry sleep authority there. Uses `tokio::time` (already transitive via tonic).
217pub(crate) async fn apply_overloaded_backoff(delay: Duration) {
218    tokio::time::sleep(delay).await;
219}
220
221/// Bounded exponential backoff with decorrelated jitter for Overloaded retries.
222///
223/// Ladder is `min(cap, base * 2^index)` with cap 400 ms. Delay is drawn uniform
224/// in `[capped/4, capped]` via splitmix64 over `(seed, index)` so concurrent
225/// clients at the same retry index do not sleep identically. No process-global
226/// RNG is used; the seed is the request id's first eight little-endian bytes.
227#[must_use]
228pub(crate) fn overloaded_backoff(jitter_seed: u64, retry_index: u32) -> Duration {
229    let shift = retry_index.min(4);
230    let base_ms = OVERLOADED_BACKOFF_BASE
231        .as_millis()
232        .saturating_mul(1u128 << shift);
233    let capped_ms = base_ms.min(OVERLOADED_BACKOFF_CAP.as_millis());
234    if capped_ms == 0 {
235        return Duration::from_millis(0);
236    }
237    let floor_ms = (capped_ms / 4).max(1);
238    let span = capped_ms.saturating_sub(floor_ms).saturating_add(1);
239    let mixed = splitmix64(
240        jitter_seed
241            ^ 0x9E37_79B9_7F4A_7C15u64.wrapping_mul(u64::from(retry_index.saturating_add(1))),
242    );
243    let offset = u128::from(mixed) % span;
244    let ms = floor_ms.saturating_add(offset).min(capped_ms);
245    Duration::from_millis(u64::try_from(ms).unwrap_or(u64::MAX))
246}
247
248/// Deterministic 64-bit mix used only for client Overloaded jitter.
249const fn splitmix64(mut value: u64) -> u64 {
250    value = value.wrapping_add(0x9E37_79B9_7F4A_7C15);
251    let mut z = value;
252    z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
253    z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
254    z ^ (z >> 31)
255}
256
257const fn is_overloaded(error: &ClientError) -> bool {
258    matches!(
259        error,
260        ClientError::Public(error) if matches!(error.kind(), PublicErrorKind::Overloaded)
261    ) || matches!(
262        error,
263        ClientError::Application(error)
264            if matches!(error.code(), ApplicationErrorCode::Overloaded)
265    )
266}
267
268#[derive(Debug)]
269pub(crate) enum RetryDecision {
270    /// Immediate same-input resubmission (non-Overloaded retryables).
271    Retry,
272    /// Overloaded: wait then resubmit within the remaining attempt budget.
273    RetryAfter(Duration),
274    Return(ClientError),
275}
276
277#[cfg(test)]
278mod tests {
279    use riffdb_errors::{PublicError, RecoveryAction};
280
281    use super::*;
282    use crate::status::DetailsFreeStatus;
283
284    fn record_input() -> v1::Value {
285        v1::Value {
286            kind: Some(v1::value::Kind::RecordValue(v1::ValueRecord {
287                fields: vec![v1::ValueField {
288                    field_id: Some(1),
289                    name: String::new(),
290                    value: Some(v1::Value {
291                        kind: Some(v1::value::Kind::StringValue("same-key".to_owned())),
292                    }),
293                }],
294            })),
295        }
296    }
297
298    #[test]
299    fn every_request_changes_only_the_outer_request_id() {
300        let command =
301            IdempotentCommand::new("CreateBudget", Some(7), record_input()).expect("command");
302        let first_id = RequestId::from_unix_milliseconds_and_random(1, [1; 10]).expect("ID");
303        let second_id = RequestId::from_unix_milliseconds_and_random(2, [2; 10]).expect("ID");
304        let first = command.request(first_id);
305        let second = command.request(second_id);
306
307        assert_ne!(first.request_id, second.request_id);
308        assert_eq!(first.command_name, second.command_name);
309        assert_eq!(
310            first.expected_contract_version,
311            second.expected_contract_version
312        );
313        assert_eq!(first.input, second.input);
314    }
315
316    #[test]
317    fn transport_uncertainty_exhaustion_is_explicit_outcome_unknown() {
318        let mut state = RetryState::new(AttemptBudget::new(2).expect("budget"));
319        assert!(state.begin_submission());
320        assert!(matches!(
321            state.handle_failure(ClientError::DetailsFree(
322                DetailsFreeStatus::TransportUnavailable
323            )),
324            RetryDecision::Retry
325        ));
326        assert!(state.begin_submission());
327        assert!(matches!(
328            state.handle_failure(ClientError::DetailsFree(
329                DetailsFreeStatus::TransportUnavailable
330            )),
331            RetryDecision::Return(ClientError::OutcomeUnknown(_))
332        ));
333    }
334
335    #[test]
336    fn exhausted_known_retry_error_preserves_the_checked_public_error() {
337        let error = PublicError::concurrency_deadline_exceeded();
338        assert_eq!(error.recovery_action(), RecoveryAction::Retry);
339        let mut state = RetryState::new(AttemptBudget::new(1).expect("budget"));
340        assert!(state.begin_submission());
341        assert!(matches!(
342            state.handle_failure(ClientError::Public(error)),
343            RetryDecision::Return(ClientError::Public(returned))
344                if returned == PublicError::concurrency_deadline_exceeded()
345        ));
346    }
347
348    #[test]
349    fn identifier_failure_after_uncertainty_does_not_hide_possible_commit() {
350        let mut state = RetryState::new(AttemptBudget::new(2).expect("budget"));
351        assert!(state.begin_submission());
352        assert!(matches!(
353            state.handle_failure(ClientError::DetailsFree(
354                DetailsFreeStatus::TransportUnavailable
355            )),
356            RetryDecision::Retry
357        ));
358        assert!(state.begin_submission());
359        assert!(matches!(
360            state.request_id_failure(crate::IdentifierGenerationError::EntropyUnavailable),
361            ClientError::OutcomeUnknown(_)
362        ));
363    }
364
365    #[test]
366    fn later_terminal_error_does_not_hide_an_earlier_uncertain_submission() {
367        let mut state = RetryState::new(AttemptBudget::new(2).expect("budget"));
368        assert!(state.begin_submission());
369        assert!(matches!(
370            state.handle_failure(ClientError::DetailsFree(
371                DetailsFreeStatus::TransportUnavailable
372            )),
373            RetryDecision::Retry
374        ));
375        assert!(state.begin_submission());
376        assert!(matches!(
377            state.handle_failure(ClientError::Public(PublicError::authorization_denied())),
378            RetryDecision::Return(ClientError::OutcomeUnknown(_))
379        ));
380    }
381
382    #[test]
383    fn overloaded_retry_uses_bounded_backoff_within_attempt_budget() {
384        let mut state = RetryState::new(AttemptBudget::new(3).expect("budget"));
385        assert!(state.begin_submission());
386        let first = state.handle_failure(ClientError::Public(PublicError::overloaded()));
387        assert!(matches!(first, RetryDecision::RetryAfter(delay) if !delay.is_zero()));
388        assert!(state.begin_submission());
389        let second = state.handle_failure(ClientError::Public(PublicError::overloaded()));
390        match (first, second) {
391            (RetryDecision::RetryAfter(a), RetryDecision::RetryAfter(b)) => {
392                assert!(b.as_millis() >= a.as_millis() / 2, "backoff stays bounded");
393                assert!(b <= OVERLOADED_BACKOFF_CAP);
394            }
395            other => panic!("expected RetryAfter pair, got {other:?}"),
396        }
397        assert!(state.begin_submission());
398        assert!(matches!(
399            state.handle_failure(ClientError::Public(PublicError::overloaded())),
400            RetryDecision::Return(ClientError::Public(error))
401                if error.kind() == riffdb_errors::PublicErrorKind::Overloaded
402        ));
403    }
404
405    #[test]
406    fn overloaded_backoff_is_capped_and_deterministic_for_seed_and_index() {
407        let a = overloaded_backoff(0xA1B2_C3D4_E5F6_7788, 0);
408        let b = overloaded_backoff(0xA1B2_C3D4_E5F6_7788, 0);
409        assert_eq!(a, b);
410        let capped = overloaded_backoff(0x1111_2222_3333_4444, 10);
411        assert!(capped <= OVERLOADED_BACKOFF_CAP);
412        assert!(capped.as_millis() * 4 >= OVERLOADED_BACKOFF_CAP.as_millis());
413    }
414
415    #[test]
416    fn overloaded_backoff_floor_is_at_least_one_quarter_of_cap() {
417        for index in 0..8 {
418            let delay = overloaded_backoff(0xDEAD_BEEF_CAFE_F00D, index);
419            let shift = index.min(4);
420            let base_ms = OVERLOADED_BACKOFF_BASE
421                .as_millis()
422                .saturating_mul(1u128 << shift);
423            let capped_ms = base_ms.min(OVERLOADED_BACKOFF_CAP.as_millis());
424            let floor_ms = (capped_ms / 4).max(1);
425            assert!(delay.as_millis() >= floor_ms);
426            assert!(delay.as_millis() <= capped_ms);
427        }
428    }
429
430    #[test]
431    fn overloaded_backoff_diverges_across_seeds_at_same_index() {
432        let left = overloaded_backoff(1, 2);
433        let right = overloaded_backoff(2, 2);
434        assert_ne!(left, right);
435    }
436}
riffdb_client_rust/command.rs

riffdb_client_rust/
command.rs
