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

harmos_storage/
receipts.rs
