ledgence_orchestration_api/storage.rs
1//! Atomic persistence operations used by the application service.
2//!
3//! Implementations own their transaction boundaries, authoritative clock, and
4//! generated identities. A successful mutation reply follows durable commit of
5//! all affected records and history. A database failure is never an empty queue
6//! or proof that ownership was lost; an uncertain commit requires reconciliation
7//! using the original operation identity.
8
9use crate::*;
10use ledgence_worker_api::ProgramDescriptor;
11
12/// Persistence boundary for complete single-task lifecycle operations.
13///
14/// Adapters load related records consistently, invoke the lifecycle core under
15/// the required locks, and atomically persist its complete transition. These
16/// operations must not perform external program resolution inside transactions.
17pub trait TaskStore: Send + Sync {
18 /// Claim one exact external dispatch. Successful replies follow durable
19 /// acceptance and bind the complete command. Unsupported implementations
20 /// reject explicitly; they must never fall back to an unrestricted queue scan.
21 fn claim_dispatch<'a>(&'a self, _command: &'a ClaimCommand) -> ContractFuture<'a, ClaimReply> {
22 Box::pin(async {
23 Err(ContractError::InvalidInput(
24 "targeted dispatch claims are unsupported".into(),
25 ))
26 })
27 }
28
29 fn open_session<'a>(
30 &'a self,
31 scope: &'a Scope,
32 queue: &'a str,
33 concurrency: u32,
34 ) -> ContractFuture<'a, WorkerSession>;
35 fn extend_session<'a>(
36 &'a self,
37 worker_session_id: &'a str,
38 ) -> ContractFuture<'a, WorkerSession>;
39 /// Read an already accepted submission before contacting the program store.
40 fn lookup_submission<'a>(
41 &'a self,
42 scope: &'a Scope,
43 idempotency_key: &'a str,
44 ) -> ContractFuture<'a, Option<TaskSnapshot>>;
45 /// Atomically accept this binding or replay the concurrently accepted winner.
46 ///
47 /// Scoped submission-key uniqueness is authoritative. A matching winner's
48 /// input, descriptor, and origin context remain unchanged; different
49 /// normalized input conflicts. The descriptor supplied by a losing caller
50 /// must never replace the accepted one.
51 fn accept_resolved_submission<'a>(
52 &'a self,
53 command: &'a SubmitCommand,
54 descriptor: &'a ProgramDescriptor,
55 ) -> ContractFuture<'a, TaskSnapshot>;
56 /// Read one bounded page of matching committed task statuses in descending
57 /// submission-time/task-ID order. Each page has its own read snapshot.
58 fn list_tasks<'a>(
59 &'a self,
60 scope: &'a Scope,
61 query: &'a TaskListQuery,
62 ) -> ContractFuture<'a, TaskPage>;
63 /// Read compact scheduling metadata without application payloads.
64 fn status<'a>(&'a self, scope: &'a Scope, task_id: &'a str) -> ContractFuture<'a, TaskStatus>;
65 /// Read task metadata and its logical outcome from one consistent snapshot.
66 fn result<'a>(&'a self, scope: &'a Scope, task_id: &'a str) -> ContractFuture<'a, TaskResult>;
67 fn inspect<'a>(
68 &'a self,
69 scope: &'a Scope,
70 task_id: &'a str,
71 ) -> ContractFuture<'a, TaskSnapshot>;
72 fn inspect_attempt<'a>(
73 &'a self,
74 scope: &'a Scope,
75 task_id: &'a str,
76 attempt_id: &'a str,
77 ) -> ContractFuture<'a, AttemptSnapshot>;
78 /// Read at most 100 ordered history records after the supplied sequence.
79 fn history<'a>(
80 &'a self,
81 scope: &'a Scope,
82 task_id: &'a str,
83 after_sequence: u64,
84 ) -> ContractFuture<'a, Vec<RecordedHistoryEvent>>;
85 /// Probe under short atomic storage locks. Pending must roll back every
86 /// mutation and release its connection before returning. The deadline bounds
87 /// connection admission, contention, retries and commit acknowledgement.
88 fn probe_acquisition<'a>(
89 &'a self,
90 command: &'a AcquireCommand,
91 finish_empty: bool,
92 deadline: std::time::Instant,
93 ) -> ContractFuture<'a, AcquisitionProbe>;
94
95 /// Immediate completion convenience for storage consumers and adapter tests.
96 fn acquire<'a>(&'a self, command: &'a AcquireCommand) -> ContractFuture<'a, AcquireReply> {
97 Box::pin(async move {
98 let deadline = std::time::Instant::now()
99 + std::time::Duration::from_millis(CONTROL_REQUEST_TIMEOUT_MS);
100 match self.probe_acquisition(command, true, deadline).await? {
101 AcquisitionProbe::Completed { reply, .. } => Ok(reply),
102 AcquisitionProbe::Pending { .. } => Err(ContractError::Unavailable(
103 "store returned Pending from final acquisition probe".into(),
104 )),
105 }
106 })
107 }
108 fn renew<'a>(&'a self, command: &'a RenewCommand) -> ContractFuture<'a, Authority>;
109 fn settle<'a>(&'a self, command: &'a SettleCommand) -> ContractFuture<'a, SettleReply>;
110 fn confirm_quiescence<'a>(&'a self, owner: &'a LeaseOwner) -> ContractFuture<'a, TaskState>;
111 fn cancel<'a>(&'a self, scope: &'a Scope, task_id: &'a str) -> ContractFuture<'a, TaskState>;
112}
113
114/// Maximum number of task candidates shortlisted by one recovery operation.
115pub const MAX_RECOVERY_BATCH: u32 = 100;
116
117/// Committed progress from one bounded recovery operation.
118///
119/// Counts do not imply the expired queue is exhausted: other candidates may
120/// remain, including locked tasks skipped by this operation.
121#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
122pub struct RecoveryProgress {
123 /// Shortlisted candidates examined, including candidates subsequently skipped.
124 pub examined: u32,
125 /// Expiry transitions successfully committed.
126 pub expired: u32,
127}
128
129/// Internal maintenance boundary, independent of worker/client delivery calls.
130pub trait RecoveryStore: Send + Sync {
131 /// Recover at most `limit` candidates, where `1 <= limit <= MAX_RECOVERY_BATCH`.
132 ///
133 /// Each candidate is rechecked with authoritative time under its task lock.
134 /// Repeated/concurrent scans must not duplicate finalization or history.
135 /// Query duration is bounded separately from the shortlist size. A later
136 /// error may follow already committed task transitions; retries are safe.
137 fn expire_batch(&self, limit: u32) -> ContractFuture<'_, RecoveryProgress>;
138}