harmos_storage/
receipts.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::collections::{BTreeMap, VecDeque};
18use std::num::NonZeroUsize;
19use std::sync::Arc;
20use std::time::{Duration, SystemTime};
21
22// Third Party Imports
23use serde::{Deserialize, Serialize};
24use tokio::time::Instant;
25
26// Local Imports
27use super::config::MUTATION_RETENTION;
28use super::{FileProducer, MutationId, MutationWindow, StorageManager, StoreHistory};
29use crate::{Error, Identifier, MemorySize, Result};
30
31/* === Constants === */
32
33const ID_COPIES: usize = 5;
34const COMPLETION_BYTES: usize = 2 * std::mem::size_of::<usize>();
35const NODE_SHARE: usize = node_bytes().div_ceil(4);
36// The Map And Order Headers Plus The Root Node, Charged While Any Receipt Is Held.
37const LEDGER_BASE: usize = std::mem::size_of::<BTreeMap<MutationId, Entry>>() + std::mem::size_of::<VecDeque<MutationId>>() + node_bytes();
38// Smallest Possible Receipt Charge, Bounding Recovered Row Counts.
39const ENTRY_MINIMUM: usize =
40    ID_COPIES * std::mem::size_of::<MutationId>() + std::mem::size_of::<MutationValue>() + NODE_SHARE + COMPLETION_BYTES;
41
42/* === Macros === */
43
44/* === Type Aliases === */
45
46/* === Structs === */
47
48#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
49#[serde(deny_unknown_fields)]
50pub struct MutationReceipt {
51    pub mutation: MutationId,
52    pub fingerprint: [u8; 32],
53    pub expires_at: SystemTime,
54    pub value: MutationValue,
55}
56
57pub(super) struct Receipts {
58    execution: crate::execution::Execution,
59    window: Identifier,
60    deadline: Instant,
61    expires_at: SystemTime,
62    entries: BTreeMap<MutationId, Entry>,
63    // Admission Order, Oldest First, For Expiry And Eviction.
64    order: VecDeque<MutationId>,
65    bytes: usize,
66    bound: NonZeroUsize,
67}
68
69struct Entry {
70    fingerprint: [u8; 32],
71    deadline: Instant,
72    expires_at: SystemTime,
73    value: Option<MutationValue>,
74    bytes: usize,
75    completion: Arc<()>,
76}
77
78pub(super) struct Completion(Arc<()>);
79
80/* === Enums === */
81
82#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
83#[serde(tag = "kind", content = "value", rename_all = "snake_case", deny_unknown_fields)]
84pub enum MutationValue {
85    Cursor(super::StorageCursor),
86    Deleted(super::StagedFile),
87    Producer(FileProducer),
88    Reference(Box<super::FileReference>),
89    Progress(super::FileProgress),
90    Rejected,
91}
92
93pub(super) enum ReceiptState {
94    Missing,
95    Pending,
96    Complete(MutationValue),
97}
98
99/* === Traits === */
100
101/* === Implementations === */
102
103impl Entry {
104    fn reclaimable(&self) -> bool {
105        self.value.is_some() && Arc::strong_count(&self.completion) == 1
106    }
107}
108
109impl Receipts {
110    pub(super) fn new(retention: Duration, bound: NonZeroUsize, execution: crate::execution::Execution) -> Result<Self> {
111        let now = execution.now();
112        let wall = SystemTime::now();
113        Ok(Self {
114            execution,
115            window: epoch()?,
116            deadline: now + retention,
117            expires_at: wall + retention,
118            entries: BTreeMap::new(),
119            order: VecDeque::new(),
120            bytes: 0,
121            bound,
122        })
123    }
124
125    pub(super) fn recovery_limits(&self) -> super::ReadLimits {
126        super::ReadLimits {
127            rows: NonZeroUsize::new(self.bound.get() / ENTRY_MINIMUM).unwrap_or(NonZeroUsize::MIN),
128            bytes: self.bound.saturating_add(std::mem::size_of::<Vec<MutationReceipt>>()),
129        }
130    }
131
132    pub(super) fn window(&mut self, history: &StoreHistory, retention: Duration) -> Result<MutationWindow> {
133        let now = self.execution.now();
134        self.purge(now);
135        if now >= self.deadline {
136            self.window = epoch()?;
137            self.deadline = now + retention;
138            self.expires_at = SystemTime::now().checked_add(retention).ok_or_else(expired)?;
139        }
140        Ok(MutationWindow { history: history.clone(), id: self.window.clone(), expires_at: self.expires_at })
141    }
142
143    pub(super) fn lookup(&self, id: &MutationId, fingerprint: &[u8; 32], completion: Option<&Completion>) -> Result<ReceiptState> {
144        let now = self.execution.now();
145        if let Some(entry) = self.entries.get(id) {
146            let admitted = completion.is_some_and(|completion| Arc::ptr_eq(&completion.0, &entry.completion));
147            if !admitted && now >= entry.deadline {
148                tracing::error!(window = %id.window, mutation = %id.key, reason = "retry window expired", "storage mutation rejected");
149                return Err(expired());
150            }
151            if entry.fingerprint != *fingerprint {
152                tracing::error!(window = %id.window, mutation = %id.key, reason = "identity reused with different request", "storage mutation rejected");
153                return Err(Error::InvalidInput { field: "storage mutation", reason: "identity reused with different request" });
154            }
155            return Ok(match &entry.value {
156                Some(value) => ReceiptState::Complete(value.clone()),
157                None => ReceiptState::Pending,
158            });
159        }
160        if id.window != self.window || now >= self.deadline {
161            tracing::error!(window = %id.window, mutation = %id.key, reason = "retry window expired or unknown", "storage mutation rejected");
162            return Err(expired());
163        }
164        Ok(ReceiptState::Missing)
165    }
166
167    pub(super) fn completion(&self, id: &MutationId) -> Completion {
168        Completion(self.entries.get(id).expect("admitted receipt").completion.clone())
169    }
170
171    pub(super) fn insert(&mut self, id: &MutationId, fingerprint: [u8; 32], outcome_bytes: usize) -> Result<()> {
172        self.purge(self.execution.now());
173        if !matches!(self.lookup(id, &fingerprint, None)?, ReceiptState::Missing) {
174            return Err(Error::AlreadyRegistered { domain: "storage mutation", name: id.key.to_string() });
175        }
176        let bytes = charge(id, outcome_bytes)?;
177        while self.bytes.saturating_add(LEDGER_BASE).saturating_add(bytes) > self.bound.get() {
178            self.evict_oldest()?;
179        }
180        self.order.push_back(id.clone());
181        self.entries.insert(
182            id.clone(),
183            Entry { fingerprint, deadline: self.deadline, expires_at: self.expires_at, value: None, bytes, completion: Arc::new(()) },
184        );
185        self.bytes += bytes;
186        Ok(())
187    }
188
189    pub(super) fn prepared(&self, id: &MutationId, value: MutationValue) -> MutationReceipt {
190        let entry = self.entries.get(id).expect("admitted receipt");
191        MutationReceipt { mutation: id.clone(), fingerprint: entry.fingerprint, expires_at: entry.expires_at, value }
192    }
193
194    pub(super) fn complete(&mut self, id: &MutationId, value: MutationValue) {
195        let entry = self.entries.get_mut(id).expect("pending receipt retained through completion");
196        if entry.value.is_some() {
197            return;
198        }
199        // Admission Reserved The Largest Possible Outcome Before Mutating.
200        assert!(charge(id, value.memory_size()).is_ok_and(|bytes| bytes <= entry.bytes));
201        entry.value = Some(value);
202    }
203
204    pub(super) fn snapshot_bytes(&self) -> usize {
205        if self.entries.is_empty() { 0 } else { self.bytes + LEDGER_BASE }
206    }
207
208    pub(super) fn clear(&mut self) {
209        self.entries = BTreeMap::new();
210        self.order = VecDeque::new();
211        self.bytes = 0;
212    }
213
214    // Drop The Oldest Completed Receipt; Pending Receipts Still Await Their Outcome.
215    fn evict_oldest(&mut self) -> Result<()> {
216        let index = self.order.iter().position(|id| self.entries.get(id).is_some_and(Entry::reclaimable)).ok_or_else(exhausted)?;
217        let id = self.order.remove(index).expect("ordered receipt");
218        let entry = self.entries.remove(&id).expect("ordered receipt");
219        self.bytes -= entry.bytes;
220        Ok(())
221    }
222
223    pub(super) fn snapshot(&self, through: Option<&super::StorageCursor>) -> Vec<MutationReceipt> {
224        let now = self.execution.now();
225        let count = self.entries.values().filter(|entry| entry.value.is_some() && now < entry.deadline).count();
226        let mut receipts = Vec::with_capacity(count);
227        for (mutation, entry) in &self.entries {
228            if let Some(value) = &entry.value
229                && now < entry.deadline
230                && match value {
231                    MutationValue::Reference(reference) => {
232                        through.is_none_or(|through| through.covers(&reference.position).unwrap_or(false))
233                    }
234                    MutationValue::Cursor(cursor) => through.is_none_or(|through| through.covers(cursor).unwrap_or(false)),
235                    _ => true,
236                }
237            {
238                receipts.push(MutationReceipt {
239                    mutation: mutation.clone(),
240                    fingerprint: entry.fingerprint,
241                    expires_at: entry.expires_at,
242                    value: value.clone(),
243                });
244            }
245        }
246        receipts
247    }
248
249    pub(super) fn restore(&mut self, mut receipts: Vec<MutationReceipt>) -> Result<()> {
250        let now = self.execution.now();
251        let wall = SystemTime::now();
252        receipts.sort_by_key(|receipt| receipt.expires_at);
253        let mut restored = BTreeMap::new();
254        let mut order = VecDeque::with_capacity(receipts.len());
255        let mut total = 0usize;
256        for receipt in receipts {
257            let Ok(remaining) = receipt.expires_at.duration_since(wall) else {
258                continue;
259            };
260            let remaining = remaining.min(MUTATION_RETENTION);
261            if remaining.is_zero() {
262                continue;
263            }
264            let bytes = charge(&receipt.mutation, receipt.value.memory_size())?;
265            total = total.checked_add(bytes).ok_or_else(exhausted)?;
266            let entry = Entry {
267                fingerprint: receipt.fingerprint,
268                deadline: now + remaining,
269                expires_at: receipt.expires_at,
270                value: Some(receipt.value),
271                bytes,
272                completion: Arc::new(()),
273            };
274            order.push_back(receipt.mutation.clone());
275            if restored.insert(receipt.mutation, entry).is_some() {
276                return Err(Error::InvalidInput { field: "mutation recovery", reason: "duplicate receipt identity" });
277            }
278        }
279        self.entries = restored;
280        self.order = order;
281        self.bytes = total;
282        while self.snapshot_bytes() > self.bound.get() {
283            self.evict_oldest()?;
284        }
285        Ok(())
286    }
287
288    // Pending Work And Attached Callers Retain Their Charged Outcome Independently
289    // Of Retry Eligibility. A Pinned Front Must Not Hold Later Expired Receipts.
290    pub(super) fn purge(&mut self, now: Instant) {
291        self.order.retain(|id| {
292            let remove = self.entries.get(id).is_none_or(|entry| now >= entry.deadline && entry.reclaimable());
293            if remove && let Some(entry) = self.entries.remove(id) {
294                self.bytes -= entry.bytes;
295            }
296            !remove
297        });
298        if self.entries.is_empty() {
299            self.clear();
300        }
301    }
302}
303
304impl<C> StorageManager<C> {
305    #[tracing::instrument(name = "storage.receipts.recover_receipts", level = "debug", skip_all)]
306    pub(super) async fn recover_receipts(&self) -> Result<()> {
307        if !self.inner.primary.supports_mutation_receipts() {
308            return Ok(());
309        }
310        let limits = self.inner.state.lock().unwrap().receipts.recovery_limits();
311        let bytes = limits.bytes.get();
312        let memory = super::capacity::Reservation::lifecycle(&self.inner, super::capacity::Pool::Read, bytes)?.retain();
313        let receipts =
314            self.inner.bounded("primary.recover_mutations", self.inner.primary.recover_mutations(limits, memory.clone())).await?;
315        if receipts.len() > limits.rows.get() || receipts.memory_size() > bytes {
316            return Err(Error::InvalidInput { field: "mutation recovery", reason: "backend exceeded recovery bounds" });
317        }
318        for receipt in &receipts {
319            if let MutationValue::Deleted(file) = &receipt.value {
320                self.inner.primary.history().check(&file.history)?;
321            }
322            if let MutationValue::Cursor(cursor) = &receipt.value {
323                let state = self.inner.state.lock().unwrap();
324                if !state.status.saved.covers(cursor)? {
325                    return Err(Error::InvalidInput { field: "mutation recovery", reason: "receipt exceeds saved data frontier" });
326                }
327            }
328            if let MutationValue::Producer(producer) = &receipt.value {
329                self.inner.primary.history().check(&producer.session.history)?;
330                if producer.session.generation == 0 {
331                    return Err(Error::InvalidInput { field: "mutation recovery", reason: "zero producer generation" });
332                }
333            }
334        }
335        for receipt in &receipts {
336            if let MutationValue::Reference(reference) = &receipt.value {
337                self.inner.primary.history().check(&reference.position.history)?;
338                reference
339                    .check_native()
340                    .map_err(|_| Error::InvalidInput { field: "mutation recovery", reason: "invalid completed file reference" })?;
341            }
342        }
343        let mut state = self.inner.state.lock().unwrap();
344        state.receipts.restore(receipts)?;
345        state.collect();
346        Ok(())
347    }
348
349    pub fn info(&self) -> Result<crate::storage::StorageInfo> {
350        let mut state = self.inner.state.lock().unwrap();
351        if !state.ready || state.status.closing {
352            return Err(Error::NotReady { domain: "storage" });
353        }
354        let mutations = if self.inner.primary.supports_mutation_receipts() {
355            Some(state.receipts.window(self.inner.primary.history(), MUTATION_RETENTION)?)
356        } else {
357            None
358        };
359        Ok(crate::storage::StorageInfo {
360            accepted: state.status.accepted.clone(),
361            saved: state.status.saved.clone(),
362            mutations,
363            file_batch_bytes: self.inner.settings.response.batch_bytes.get() as u64,
364        })
365    }
366
367    pub fn mutation_window(&self) -> Result<MutationWindow> {
368        if !self.inner.primary.supports_mutation_receipts() {
369            return Err(Error::InvalidInput { field: "storage mutations", reason: "backend does not support mutation receipts" });
370        }
371        let mut state = self.inner.state.lock().unwrap();
372        if !state.ready || state.status.closing {
373            return Err(Error::NotReady { domain: "storage mutations" });
374        }
375        let window = state.receipts.window(self.inner.primary.history(), MUTATION_RETENTION)?;
376        state.collect();
377        Ok(window)
378    }
379}
380
381impl MutationValue {
382    pub(super) fn deleted(self) -> Result<super::StagedFile> {
383        match self {
384            Self::Deleted(file) => Ok(file),
385            _ => Err(rejected()),
386        }
387    }
388
389    pub(super) fn cursor(self) -> Result<super::StorageCursor> {
390        match self {
391            Self::Cursor(value) => Ok(value),
392            _ => Err(rejected()),
393        }
394    }
395
396    pub(super) fn producer(self) -> Result<FileProducer> {
397        match self {
398            Self::Producer(value) => Ok(value),
399            _ => Err(rejected()),
400        }
401    }
402
403    pub(super) fn reference(self) -> Result<super::FileReference> {
404        match self {
405            Self::Reference(value) => Ok(*value),
406            _ => Err(rejected()),
407        }
408    }
409
410    pub(super) fn progress(self) -> Result<super::FileProgress> {
411        match self {
412            Self::Progress(value) => Ok(value),
413            _ => Err(rejected()),
414        }
415    }
416}
417
418impl MemorySize for MutationValue {
419    fn memory_size(&self) -> usize {
420        std::mem::size_of::<Self>()
421            + match self {
422                Self::Deleted(value) => value.memory_size() - std::mem::size_of::<super::StagedFile>(),
423                Self::Cursor(value) => value.memory_size() - std::mem::size_of::<super::StorageCursor>(),
424                Self::Producer(value) => value.memory_size() - std::mem::size_of::<FileProducer>(),
425                Self::Reference(value) => value.as_ref().memory_size(),
426                Self::Progress(_) | Self::Rejected => 0,
427            }
428    }
429}
430impl MemorySize for MutationReceipt {
431    fn memory_size(&self) -> usize {
432        std::mem::size_of::<Self>() + self.mutation.memory_size() - std::mem::size_of::<MutationId>() + self.value.memory_size()
433            - std::mem::size_of::<MutationValue>()
434    }
435}
436
437/* === Public Functions === */
438
439/* === Private Functions === */
440
441fn epoch() -> Result<Identifier> {
442    let mut bytes = [0u8; 16];
443    getrandom::fill(&mut bytes)
444        .map_err(|source| Error::OperationFailed { operation: "storage retry window".into(), source: Arc::new(source) })?;
445    let name = bytes.iter().map(|byte| format!("{byte:02x}")).collect::<String>();
446    Identifier::new(name).map_err(|_| Error::InvalidInput { field: "storage retry window", reason: "invalid generated identity" })
447}
448
449// One Entry's Amortized Share: Std B-Tree Nodes Below The Root Hold At Least Five
450// Entries And Internal Nodes Add At Most A Quarter More, So A Quarter Node Per
451// Entry Covers Every Node But The Root, Which `LEDGER_BASE` Charges Once. The
452// Identity Is Copied Into The Map Key, The Admission Order, Snapshots And Saves.
453fn charge(id: &MutationId, outcome_bytes: usize) -> Result<usize> {
454    id.memory_size()
455        .checked_mul(ID_COPIES)
456        .and_then(|bytes| bytes.checked_add(outcome_bytes))
457        .and_then(|bytes| bytes.checked_add(NODE_SHARE))
458        .and_then(|bytes| bytes.checked_add(COMPLETION_BYTES))
459        .ok_or_else(exhausted)
460}
461
462const fn node_bytes() -> usize {
463    let alignment = max(max(std::mem::align_of::<MutationId>(), std::mem::align_of::<Entry>()), std::mem::align_of::<usize>());
464    (std::mem::size_of::<MutationId>() + std::mem::size_of::<Entry>()) * 11
465        + 13 * std::mem::size_of::<usize>()
466        + 2 * std::mem::size_of::<u16>()
467        + 5 * alignment
468}
469
470const fn max(left: usize, right: usize) -> usize {
471    if left > right { left } else { right }
472}
473
474fn rejected() -> Error {
475    Error::InvalidInput { field: "file mutation", reason: "owned production ended without the requested result" }
476}
477fn exhausted() -> Error {
478    Error::ResourceExhausted { resource: "storage mutation receipts" }
479}
480fn expired() -> Error {
481    Error::InvalidInput { field: "storage mutation", reason: "retry window expired or unknown; reconcile state before issuing new work" }
482}
483
484/* === Tests === */
harmos_storage/receipts.rs

harmos_storage/
receipts.rs
