durable-actors 0.7.1

Standalone regional durable-actors control plane, host, and durability runtime
Documentation 
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

use anyhow::{Context, Result, ensure};
use aws_lc_rs::digest::{SHA256, digest};
use base64::{Engine, engine::general_purpose::STANDARD};
use serde::{Deserialize, Serialize};

#[derive(Clone, Debug, Serialize, Deserialize)]
pub(super) struct Record {
    pub version: u64,
    pub digest: String,
    pub base: Option<String>,
    pub length: usize,
    pub data: String,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub(super) struct Batch {
    pub prefix: String,
    pub records: Vec<Record>,
}

impl Record {
    pub fn encode(version: u64, bytes: &[u8], previous: Option<&[u8]>) -> Result<Self> {
        let full = zstd::bulk::compress(bytes, 1)?;
        let delta = previous
            .map(|previous| zstd::bulk::Compressor::with_dictionary(1, previous)?.compress(bytes))
            .transpose()?;
        let (data, base) = match (previous, delta) {
            (Some(previous), Some(delta)) if delta.len() < full.len() => {
                (delta, Some(checksum(previous)))
            }
            _ => (full, None),
        };
        Ok(Self {
            version,
            digest: checksum(bytes),
            base,
            length: bytes.len(),
            data: STANDARD.encode(data),
        })
    }

    pub fn decode(&self, previous: Option<&[u8]>) -> Result<Vec<u8>> {
        let dictionary = if let Some(base) = &self.base {
            let previous = previous.context("change log base is missing")?;
            ensure!(
                checksum(previous) == *base,
                "change log base checksum mismatch"
            );
            previous
        } else {
            &[]
        };
        let bytes = zstd::bulk::Decompressor::with_dictionary(dictionary)?
            .decompress(&STANDARD.decode(&self.data)?, self.length)?;
        ensure!(
            bytes.len() == self.length && checksum(&bytes) == self.digest,
            "change log checksum mismatch"
        );
        let snapshot = crate::state_log::StateSnapshot::decode(&bytes)?;
        ensure!(
            snapshot.state_version == self.version,
            "change log version mismatch"
        );
        Ok(bytes)
    }
}

impl Batch {
    pub fn decode(&self) -> Result<Vec<(u64, Vec<u8>)>> {
        ensure!(!self.records.is_empty(), "empty archive batch");
        let mut decoded: Vec<(u64, Vec<u8>)> = Vec::new();
        for record in &self.records {
            if let Some((version, _)) = decoded.last() {
                ensure!(
                    version.checked_add(1) == Some(record.version),
                    "archive log has a gap"
                );
            }
            let bytes = record.decode(decoded.last().map(|(_, bytes)| bytes.as_slice()))?;
            decoded.push((record.version, bytes));
        }
        Ok(decoded)
    }

    pub fn key(&self) -> Result<String> {
        let first = self.records.first().context("empty archive batch")?.version;
        let last = self.records.last().unwrap().version;
        Ok(format!(
            "{}{first:020}-{last:020}-{}.json",
            archive_prefix(&self.prefix)?,
            checksum(&serde_json::to_vec(self)?)
        ))
    }
}

pub(super) fn archive_prefix(prefix: &str) -> Result<String> {
    let root = crate::storage_paths::ROOT;
    Ok(format!(
        "{root}archive/{}",
        prefix
            .strip_prefix(&format!("{root}snapshots/"))
            .context("invalid snapshot prefix")?
    ))
}

pub(super) fn checksum(bytes: &[u8]) -> String {
    use base64::engine::general_purpose::URL_SAFE_NO_PAD;
    URL_SAFE_NO_PAD.encode(digest(&SHA256, bytes).as_ref())
}