durable-actors 0.7.10

Standalone regional durable-actors control plane, host, and durability runtime
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

use super::SnapshotStore;
use crate::{
    state_log::{SqliteSnapshot, StateSnapshot},
    storage::SnapshotRef,
};
use anyhow::{Context, Result, ensure};
use async_trait::async_trait;
use bytes::Bytes;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
pub(crate) struct Checkpoint {
    pub source: SnapshotRef,
    pub sqlite: SqliteSnapshot,
}

#[async_trait]
pub(crate) trait SnapshotHistory: Send {
    async fn read(&mut self, object: &str) -> Result<Bytes>;
    async fn checkpoint(&mut self, object: &str) -> Result<Option<Bytes>>;
}

pub(crate) struct ResolvedSqlite {
    pub sqlite: SqliteSnapshot,
    pub checkpoint_version: u64,
    pub checkpoint: bool,
    pub parents: usize,
}

pub(crate) async fn resolve(
    history: &mut dyn SnapshotHistory,
    object: &str,
    bytes: Bytes,
) -> Result<ResolvedSqlite> {
    let mut current = StateSnapshot::decode(&bytes)?;
    current.validate_object(object)?;
    let txid = current.sqlite.txid;
    let mut reference = SnapshotRef::new(object.into(), &current, &bytes);
    let mut segments = Vec::new();
    let mut checkpoint = false;
    let mut parents = 0;
    loop {
        if current.sqlite.parent.is_some()
            && let Some(bytes) = history.checkpoint(&reference.object).await?
        {
            let sqlite = Checkpoint::decode(&bytes, &reference, current.sqlite.txid)
                .with_context(|| format!("validate checkpoint for {}", reference.object))?;
            checkpoint = true;
            segments.push(sqlite.files);
            break;
        }
        let first = current.sqlite.files[0].first;
        segments.push(current.sqlite.files);
        let Some(parent) = current.sqlite.parent else {
            break;
        };
        let bytes = history
            .read(&parent.object)
            .await
            .with_context(|| format!("read SQLite dependency {}", parent.object))?;
        parent
            .verify(&bytes)
            .with_context(|| format!("verify SQLite dependency {}", parent.object))?;
        current = StateSnapshot::decode(&bytes)?;
        current.validate_object(&parent.object)?;
        ensure!(
            current.state_version == parent.state_version
                && current.request_id == parent.request_id,
            "SQLite dependency identity mismatch"
        );
        ensure!(
            current.sqlite.txid.checked_add(1) == Some(first),
            "SQLite dependency transaction gap"
        );
        reference = parent;
        parents += 1;
    }
    Ok(ResolvedSqlite {
        checkpoint_version: reference.state_version,
        checkpoint,
        parents,
        sqlite: SqliteSnapshot {
            txid,
            parent: None,
            files: segments.into_iter().rev().flatten().collect(),
        },
    })
}

impl Checkpoint {
    pub fn decode(bytes: &[u8], source: &SnapshotRef, txid: u64) -> Result<SqliteSnapshot> {
        let checkpoint: Self = serde_json::from_slice(bytes)?;
        ensure!(&checkpoint.source == source, "checkpoint source mismatch");
        checkpoint.sqlite.validate(source.state_version)?;
        ensure!(
            checkpoint.sqlite.parent.is_none() && checkpoint.sqlite.txid == txid,
            "checkpoint transaction mismatch"
        );
        Ok(checkpoint.sqlite)
    }
}

pub(super) struct StoredHistory<'a, T: ?Sized>(pub &'a T);

#[async_trait]
impl<T: SnapshotStore + ?Sized> SnapshotHistory for StoredHistory<'_, T> {
    async fn read(&mut self, object: &str) -> Result<Bytes> {
        self.0
            .get(object)
            .await?
            .context("SQLite dependency missing")
    }
    async fn checkpoint(&mut self, object: &str) -> Result<Option<Bytes>> {
        self.0.get(&format!("{object}.checkpoint")).await
    }
}