ledgence_orchestration_api/
retention.rs

1//! Operator-triggered retention; no execution is expired merely by reading it.
2use crate::{ContractError, ContractFuture, Result, Scope};
3use serde::{Deserialize, Serialize};
4use std::time::Instant;
5
6pub const MIN_RETENTION_MS: u64 = 90 * 24 * 60 * 60 * 1000;
7pub const MAX_RETENTION_BATCH: u32 = 256;
8
9/// Applies to terminal executions and the latest terminal callback activity.
10/// Existing retiring records continue physical collection under any later policy.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
12#[serde(deny_unknown_fields)]
13pub struct RetentionPolicy {
14    pub retain_for_ms: u64,
15    /// Dependent-row work budget, not an execution count. Small task ledgers
16    /// may share this budget across tables within their single-target transaction.
17    pub batch_size: u32,
18}
19impl Default for RetentionPolicy {
20    fn default() -> Self {
21        Self {
22            retain_for_ms: MIN_RETENTION_MS,
23            batch_size: 128,
24        }
25    }
26}
27impl RetentionPolicy {
28    pub fn validate(&self) -> Result<()> {
29        if !(MIN_RETENTION_MS..=253_402_300_799_999).contains(&self.retain_for_ms)
30            || !(1..=MAX_RETENTION_BATCH).contains(&self.batch_size)
31        {
32            return Err(ContractError::InvalidInput(
33                "retention requires at least 90 days and a batch size from 1 through 256".into(),
34            ));
35        }
36        Ok(())
37    }
38}
39
40/// One bounded transaction. Zero removed rows does not mean the database has no
41/// retained records: protection, cursor rotation, or a new retirement can occur.
42#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(deny_unknown_fields)]
44pub struct RetentionProgress {
45    pub examined: u32,
46    pub retired: u32,
47    pub deleted_rows: u32,
48    pub deleted_executions: u32,
49    pub deleted_sessions: u32,
50}
51
52/// Bounded age candidates only; protective references can defer collection.
53#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
54#[serde(deny_unknown_fields)]
55pub struct RetentionPreview {
56    pub task_candidates: Vec<String>,
57    pub workflow_candidates: Vec<String>,
58    pub retiring_tasks: Vec<String>,
59    pub retiring_workflows: Vec<String>,
60    pub expired_session_candidates: Vec<String>,
61}
62
63pub trait RetentionStore: Send + Sync {
64    fn retention_preview<'a>(
65        &'a self,
66        scope: &'a Scope,
67        policy: &'a RetentionPolicy,
68        deadline: Instant,
69    ) -> ContractFuture<'a, RetentionPreview>;
70
71    /// Cooperating calls must preserve live cursors, active execution trees and
72    /// unfinished delivery/reconciliation obligations. Crash/retry can repeat a
73    /// batch; progress counts only the transaction acknowledged by this call.
74    fn retain_batch<'a>(
75        &'a self,
76        scope: &'a Scope,
77        policy: &'a RetentionPolicy,
78        deadline: Instant,
79    ) -> ContractFuture<'a, RetentionProgress>;
80}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85    #[test]
86    fn retention_cannot_shorten_the_documented_receipt_lifetime() {
87        assert!(RetentionPolicy::default().validate().is_ok());
88        for retain_for_ms in [0, MIN_RETENTION_MS - 1, u64::MAX] {
89            assert!(
90                RetentionPolicy {
91                    retain_for_ms,
92                    ..Default::default()
93                }
94                .validate()
95                .is_err()
96            );
97        }
98        for batch_size in [0, MAX_RETENTION_BATCH + 1, u32::MAX] {
99            assert!(
100                RetentionPolicy {
101                    batch_size,
102                    ..Default::default()
103                }
104                .validate()
105                .is_err()
106            );
107        }
108    }
109}
ledgence_orchestration_api/retention.rs

ledgence_orchestration_api/
retention.rs
