use std::time::Duration;
use serde::{Deserialize, Deserializer, Serialize};
use thiserror::Error;
use crate::hook::{
SettlementFailureClass, SettlementFailureCode, SettlementFailureReason, SettlementSkipReason,
};
use crate::outcome_store::SettlementRoutingInput;
pub const SETTLE_DEAD_LETTER_SCHEMA: &str = "chio.settle.dead-letter.v1";
fn deserialize_dead_letter_schema<'de, D>(deserializer: D) -> Result<String, D::Error>
where
D: Deserializer<'de>,
{
let schema = String::deserialize(deserializer)?;
if schema == SETTLE_DEAD_LETTER_SCHEMA {
Ok(schema)
} else {
Err(serde::de::Error::custom(
"unsupported settlement dead-letter schema",
))
}
}
pub const DEFAULT_MAX_RETRIES: u32 = 5;
pub const DEFAULT_INITIAL_BACKOFF_MS: u64 = 250;
pub const DEFAULT_BACKOFF_MULTIPLIER: u32 = 2;
pub const DEFAULT_BACKOFF_CAP_MS: u64 = 60_000;
const MAX_RETRIES: u32 = 32;
const MAX_BACKOFF_CAP_MS: u64 = 86_400_000;
const MAX_BACKOFF_MULTIPLIER: u32 = 16;
#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
pub enum RetryPolicyError {
#[error("max_retries exceeds 32: {max_retries}")]
MaxRetriesTooHigh { max_retries: u32 },
#[error("initial_backoff_ms must be nonzero")]
InitialBackoffZero,
#[error("backoff_cap_ms must be nonzero")]
BackoffCapZero,
#[error("initial_backoff_ms {initial_backoff_ms} exceeds backoff_cap_ms {backoff_cap_ms}")]
InitialBackoffExceedsCap {
initial_backoff_ms: u64,
backoff_cap_ms: u64,
},
#[error("backoff_cap_ms exceeds 86400000: {backoff_cap_ms}")]
BackoffCapTooHigh { backoff_cap_ms: u64 },
#[error("backoff_multiplier must be in 1..=16: {backoff_multiplier}")]
BackoffMultiplierOutOfRange { backoff_multiplier: u32 },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RetryPolicy {
pub max_retries: u32,
pub initial_backoff_ms: u64,
pub backoff_multiplier: u32,
pub backoff_cap_ms: u64,
}
impl Default for RetryPolicy {
fn default() -> Self {
Self {
max_retries: DEFAULT_MAX_RETRIES,
initial_backoff_ms: DEFAULT_INITIAL_BACKOFF_MS,
backoff_multiplier: DEFAULT_BACKOFF_MULTIPLIER,
backoff_cap_ms: DEFAULT_BACKOFF_CAP_MS,
}
}
}
impl RetryPolicy {
pub const fn validate(&self) -> Result<(), RetryPolicyError> {
if self.max_retries > MAX_RETRIES {
return Err(RetryPolicyError::MaxRetriesTooHigh {
max_retries: self.max_retries,
});
}
if self.initial_backoff_ms == 0 {
return Err(RetryPolicyError::InitialBackoffZero);
}
if self.backoff_cap_ms == 0 {
return Err(RetryPolicyError::BackoffCapZero);
}
if self.backoff_cap_ms > MAX_BACKOFF_CAP_MS {
return Err(RetryPolicyError::BackoffCapTooHigh {
backoff_cap_ms: self.backoff_cap_ms,
});
}
if self.initial_backoff_ms > self.backoff_cap_ms {
return Err(RetryPolicyError::InitialBackoffExceedsCap {
initial_backoff_ms: self.initial_backoff_ms,
backoff_cap_ms: self.backoff_cap_ms,
});
}
if self.backoff_multiplier == 0 || self.backoff_multiplier > MAX_BACKOFF_MULTIPLIER {
return Err(RetryPolicyError::BackoffMultiplierOutOfRange {
backoff_multiplier: self.backoff_multiplier,
});
}
Ok(())
}
#[must_use]
pub fn backoff_for(&self, attempt: u32) -> Duration {
let factor = u64::from(self.backoff_multiplier)
.max(1)
.saturating_pow(attempt);
Duration::from_millis(
self.initial_backoff_ms
.saturating_mul(factor)
.min(self.backoff_cap_ms),
)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RetryDecision {
Accepted,
Skip {
reason: SettlementSkipReason,
},
Retry {
attempt: u32,
backoff: Duration,
reason: SettlementFailureReason,
},
DeadLetter {
reason: SettlementFailureReason,
},
}
#[must_use]
pub fn classify_attempt(
policy: &RetryPolicy,
attempt: u32,
outcome: &SettlementRoutingInput,
) -> RetryDecision {
match outcome {
SettlementRoutingInput::Accepted => RetryDecision::Accepted,
SettlementRoutingInput::Skipped { reason } => RetryDecision::Skip { reason: *reason },
SettlementRoutingInput::Permanent { reason } => RetryDecision::DeadLetter {
reason: reason.clone(),
},
SettlementRoutingInput::Retryable { reason } => {
if reason.effective_class(SettlementFailureClass::Retryable)
== SettlementFailureClass::Permanent
|| attempt >= policy.max_retries
{
RetryDecision::DeadLetter {
reason: reason.clone(),
}
} else {
RetryDecision::Retry {
attempt: attempt + 1,
backoff: policy.backoff_for(attempt),
reason: reason.clone(),
}
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct DeadLetterRecord {
pub schema: String,
pub receipt_id: String,
pub finalized_at: u64,
pub attempts: u32,
pub reason: SettlementFailureReason,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TypedDeadLetterRecord {
#[serde(deserialize_with = "deserialize_dead_letter_schema")]
schema: String,
receipt_id: String,
finalized_at: u64,
attempts: u32,
reason: SettlementFailureReason,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct LegacyDeadLetterRecord {
#[serde(deserialize_with = "deserialize_dead_letter_schema")]
schema: String,
receipt_id: String,
finalized_at: u64,
attempts: u32,
reason: String,
#[serde(default)]
pipeline_error: Option<String>,
}
#[derive(Deserialize)]
#[serde(untagged)]
enum DeadLetterRecordWire {
Typed(TypedDeadLetterRecord),
Legacy(LegacyDeadLetterRecord),
}
impl<'de> Deserialize<'de> for DeadLetterRecord {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
match DeadLetterRecordWire::deserialize(deserializer)? {
DeadLetterRecordWire::Typed(record) => Ok(Self {
schema: record.schema,
receipt_id: record.receipt_id,
finalized_at: record.finalized_at,
attempts: record.attempts,
reason: record.reason,
}),
DeadLetterRecordWire::Legacy(record) => {
let detail = record.pipeline_error.as_deref().unwrap_or(&record.reason);
Ok(Self {
schema: record.schema,
receipt_id: record.receipt_id,
finalized_at: record.finalized_at,
attempts: record.attempts,
reason: SettlementFailureReason::from_detail(
SettlementFailureCode::Backend,
detail,
),
})
}
}
}
}
impl DeadLetterRecord {
#[must_use]
pub fn has_supported_schema(&self) -> bool {
self.schema == SETTLE_DEAD_LETTER_SCHEMA
}
#[must_use]
pub fn new(
receipt_id: impl Into<String>,
finalized_at: u64,
attempts: u32,
reason: SettlementFailureReason,
) -> Self {
Self {
schema: SETTLE_DEAD_LETTER_SCHEMA.to_string(),
receipt_id: receipt_id.into(),
finalized_at,
attempts: attempts.max(1),
reason,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn failure(detail: &str) -> SettlementFailureReason {
SettlementFailureReason::from_detail(SettlementFailureCode::Rpc, detail)
}
fn serialize<T: Serialize>(value: &T) -> String {
match serde_json::to_string(value) {
Ok(encoded) => encoded,
Err(error) => panic!("value must serialize: {error}"),
}
}
#[test]
fn schema_is_stable() {
assert_eq!(SETTLE_DEAD_LETTER_SCHEMA, "chio.settle.dead-letter.v1");
}
#[test]
fn string_reason_v1_schema_decodes_to_a_bounded_reason() {
let decoded = serde_json::from_value::<DeadLetterRecord>(serde_json::json!({
"schema": "chio.settle.dead-letter.v1",
"receipt_id": "receipt-1",
"finalized_at": 1,
"attempts": 1,
"reason": "rpc unavailable",
"pipeline_error": "settlement pipeline error: rpc unavailable",
}));
let record = match decoded {
Ok(record) => record,
Err(error) => panic!("legacy dead-letter record must decode: {error}"),
};
let expected = SettlementFailureReason::from_detail(
SettlementFailureCode::Backend,
"settlement pipeline error: rpc unavailable",
);
assert_eq!(record.reason, expected);
}
#[test]
fn dead_letter_deserialization_rejects_an_unsupported_schema() {
let result = serde_json::from_value::<DeadLetterRecord>(serde_json::json!({
"schema": "chio.settle.dead-letter.v99",
"receipt_id": "receipt-1",
"finalized_at": 1,
"attempts": 1,
"reason": {
"code": "backend",
"detail_sha256": vec![0_u8; 32],
},
}));
assert!(result.is_err());
}
#[test]
fn default_policy_matches_documented_bounds() {
let policy = RetryPolicy::default();
assert_eq!(policy.max_retries, DEFAULT_MAX_RETRIES);
assert_eq!(policy.initial_backoff_ms, DEFAULT_INITIAL_BACKOFF_MS);
assert_eq!(policy.backoff_multiplier, DEFAULT_BACKOFF_MULTIPLIER);
assert_eq!(policy.backoff_cap_ms, DEFAULT_BACKOFF_CAP_MS);
}
#[test]
fn backoff_grows_exponentially_until_cap() {
let policy = RetryPolicy {
max_retries: 8,
initial_backoff_ms: 100,
backoff_multiplier: 2,
backoff_cap_ms: 1000,
};
assert_eq!(policy.backoff_for(0), Duration::from_millis(100));
assert_eq!(policy.backoff_for(1), Duration::from_millis(200));
assert_eq!(policy.backoff_for(2), Duration::from_millis(400));
assert_eq!(policy.backoff_for(3), Duration::from_millis(800));
assert_eq!(policy.backoff_for(4), Duration::from_millis(1000));
assert_eq!(policy.backoff_for(50), Duration::from_millis(1000));
}
#[test]
fn permanent_outcomes_skip_the_retry_envelope() {
let policy = RetryPolicy::default();
let reason = failure("policy denied");
let outcome = SettlementRoutingInput::Permanent {
reason: reason.clone(),
};
match classify_attempt(&policy, 0, &outcome) {
RetryDecision::DeadLetter { reason: actual } => assert_eq!(actual, reason),
other => panic!("expected dead letter, got {other:?}"),
}
}
#[test]
fn skipped_outcomes_pass_through() {
let policy = RetryPolicy::default();
let outcome = SettlementRoutingInput::Skipped {
reason: SettlementSkipReason::ZeroCharge,
};
assert_eq!(
classify_attempt(&policy, 0, &outcome),
RetryDecision::Skip {
reason: SettlementSkipReason::ZeroCharge,
}
);
}
#[test]
fn accepted_outcomes_pass_through() {
let policy = RetryPolicy::default();
assert_eq!(
classify_attempt(&policy, 0, &SettlementRoutingInput::Accepted),
RetryDecision::Accepted
);
}
#[test]
fn retryable_outcomes_consume_the_envelope_then_dead_letter() {
let policy = RetryPolicy {
max_retries: 2,
initial_backoff_ms: 10,
backoff_multiplier: 2,
backoff_cap_ms: 100,
};
let reason = failure("rpc lag");
let outcome = SettlementRoutingInput::Retryable {
reason: reason.clone(),
};
match classify_attempt(&policy, 0, &outcome) {
RetryDecision::Retry {
attempt,
backoff,
reason: actual,
} => {
assert_eq!(attempt, 1);
assert_eq!(backoff, Duration::from_millis(10));
assert_eq!(actual, reason);
}
other => panic!("expected retry, got {other:?}"),
}
match classify_attempt(&policy, 1, &outcome) {
RetryDecision::Retry { attempt, .. } => assert_eq!(attempt, 2),
other => panic!("expected retry, got {other:?}"),
}
match classify_attempt(&policy, 2, &outcome) {
RetryDecision::DeadLetter { reason: actual } => assert_eq!(actual, reason),
other => panic!("expected dead letter, got {other:?}"),
}
}
#[test]
fn dead_letter_record_contains_only_bounded_failure_detail() {
let record = DeadLetterRecord::new("rcpt-1", 100, 3, failure("connection refused"));
let encoded = serialize(&record);
assert_eq!(record.attempts, 3);
assert_eq!(record.schema, SETTLE_DEAD_LETTER_SCHEMA);
assert_eq!(record.reason.code(), SettlementFailureCode::Rpc);
assert!(!encoded.contains("connection refused"));
assert!(!encoded.contains("pipeline_error"));
}
#[test]
fn dead_letter_record_attempts_floor_is_one() {
let record = DeadLetterRecord::new("rcpt-x", 0, 0, failure("permanent"));
assert_eq!(record.attempts, 1);
}
#[test]
fn retry_policy_accepts_every_boundary() {
let default = RetryPolicy::default();
let valid = [
RetryPolicy {
max_retries: 0,
..default
},
RetryPolicy {
max_retries: 32,
..default
},
RetryPolicy {
initial_backoff_ms: 1,
..default
},
RetryPolicy {
initial_backoff_ms: 1,
backoff_cap_ms: 1,
..default
},
RetryPolicy {
backoff_cap_ms: 86_400_000,
..default
},
RetryPolicy {
backoff_multiplier: 1,
..default
},
RetryPolicy {
backoff_multiplier: 16,
..default
},
];
for policy in valid {
assert_eq!(policy.validate(), Ok(()));
}
}
#[test]
fn retry_policy_rejects_every_out_of_bounds_value() {
let default = RetryPolicy::default();
let invalid = [
(
RetryPolicy {
max_retries: 33,
..default
},
RetryPolicyError::MaxRetriesTooHigh { max_retries: 33 },
),
(
RetryPolicy {
initial_backoff_ms: 0,
..default
},
RetryPolicyError::InitialBackoffZero,
),
(
RetryPolicy {
backoff_cap_ms: 0,
..default
},
RetryPolicyError::BackoffCapZero,
),
(
RetryPolicy {
initial_backoff_ms: default.backoff_cap_ms + 1,
..default
},
RetryPolicyError::InitialBackoffExceedsCap {
initial_backoff_ms: default.backoff_cap_ms + 1,
backoff_cap_ms: default.backoff_cap_ms,
},
),
(
RetryPolicy {
backoff_cap_ms: 86_400_001,
..default
},
RetryPolicyError::BackoffCapTooHigh {
backoff_cap_ms: 86_400_001,
},
),
(
RetryPolicy {
backoff_multiplier: 0,
..default
},
RetryPolicyError::BackoffMultiplierOutOfRange {
backoff_multiplier: 0,
},
),
(
RetryPolicy {
backoff_multiplier: 17,
..default
},
RetryPolicyError::BackoffMultiplierOutOfRange {
backoff_multiplier: 17,
},
),
];
for (policy, expected) in invalid {
assert_eq!(policy.validate(), Err(expected));
}
}
#[test]
fn typed_retry_reason_survives_exhaustion() {
let reason = failure("upstream unavailable");
let input = SettlementRoutingInput::Retryable {
reason: reason.clone(),
};
let policy = RetryPolicy {
max_retries: 0,
..RetryPolicy::default()
};
assert!(matches!(
classify_attempt(&policy, 0, &input),
RetryDecision::DeadLetter { reason: actual } if actual == reason
));
}
#[test]
fn known_permanent_code_never_enters_the_retry_envelope() {
let reason = SettlementFailureReason::from_detail(
SettlementFailureCode::InvalidReceiptSignature,
"invalid signature",
);
let input = SettlementRoutingInput::Retryable {
reason: reason.clone(),
};
assert_eq!(
classify_attempt(&RetryPolicy::default(), 0, &input),
RetryDecision::DeadLetter { reason }
);
}
#[test]
fn backoff_with_unit_multiplier_is_constant_for_any_attempt() {
let policy = RetryPolicy {
backoff_multiplier: 1,
..RetryPolicy::default()
};
assert_eq!(
policy.backoff_for(u32::MAX),
Duration::from_millis(policy.initial_backoff_ms)
);
}
}