aven_core/sync/
encrypted_tail.rs

1//! Internal ordinary encrypted task stream. Membership remains chain-owned.
2pub mod attachments;
3pub mod batch;
4mod client;
5pub use client::Preflight;
6mod codec;
7pub(crate) mod dependencies;
8mod domain;
9#[cfg(feature = "test-support")]
10pub(crate) mod fuzz;
11pub(crate) mod graphs;
12mod labels;
13mod moves;
14mod notes;
15mod recurrence;
16mod server;
17
18use super::{LocalSharedStatePackageKey, seed_claim::peer};
19use anyhow::{Result, ensure};
20use serde::{Deserialize, Serialize};
21use sha2::Sha256;
22
23pub const RECORD_LIMIT: usize = 135640;
24/// Bodies without a record; also the framing allowance around records.
25pub const CONTROL_LIMIT: usize = 16384;
26/// One record: an append request or a lookup response.
27pub const APPEND_LIMIT: usize = super::base64_bytes::encoded_len(RECORD_LIMIT) + CONTROL_LIMIT;
28/// Serialized size of the records in one pull page.
29pub const PAGE_BYTES: usize = 2 * 1048576;
30pub const PAGE_COUNT: usize = 256;
31pub const RESPONSE_LIMIT: usize = PAGE_BYTES + CONTROL_LIMIT;
32pub const BATCH_COUNT: usize = 128;
33pub const BATCH_BYTES: usize = 1048576;
34pub const BATCH_APPEND_LIMIT: usize = super::base64_bytes::encoded_len(BATCH_BYTES) + CONTROL_LIMIT;
35pub const BATCH_CONTROL_LIMIT: usize = 256 * 1024;
36
37/// An appended record's identity collides with a published prefix record.
38#[derive(Debug)]
39pub struct PrefixIdentityCollision;
40
41impl std::fmt::Display for PrefixIdentityCollision {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        f.write_str("error encrypted-tail-prefix-identity-collision")
44    }
45}
46
47impl std::error::Error for PrefixIdentityCollision {}
48
49#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
50#[serde(deny_unknown_fields)]
51pub struct Context {
52    pub vault: [u8; 32],
53    pub genesis: [u8; 32],
54    pub device: [u8; 32],
55    pub head: [u8; 32],
56    pub stream: [u8; 32],
57    pub descriptor: [u8; 32],
58}
59impl Context {
60    pub fn authentication<'a>(
61        &self,
62        bearer: &'a super::seed_claim::Secret,
63    ) -> peer::Authentication<'a> {
64        peer::Authentication {
65            vault: self.vault,
66            genesis: self.genesis,
67            device: self.device,
68            head: self.head,
69            bearer,
70        }
71    }
72}
73/// Protected host inputs. The host retains installation and authority exclusion.
74pub struct Authority {
75    pub context: Context,
76    pub membership: super::seed_claim::membership::Membership,
77    pub keys: super::seed_claim::membership::VerifiedKeys,
78    pub prefix: i64,
79    pub association: String,
80    pub sync_generation: i64,
81}
82/// Context-bound lookup absence. Only atomic replacement consumes this observation.
83pub struct AbsentOperation {
84    context: Context,
85    record: Vec<u8>,
86}
87/// Exact frozen bytes for one dispatch. Image heads include their staged upload.
88pub struct Push {
89    pub record: Vec<u8>,
90    pub upload: Option<attachments::Upload>,
91}
92/// Local round state read in one validated transaction.
93pub struct RoundState {
94    pub cursor: i64,
95    /// Finite page target captured while initial image catch-up is pending.
96    pub initial_watermark: Option<i64>,
97    /// No frozen or unacknowledged local metadata.
98    pub idle: bool,
99    pub upload_pending: bool,
100    /// Image demand, known only after initial image catch-up.
101    pub downloads: Option<Downloads>,
102}
103#[derive(Clone, Copy)]
104pub struct Downloads {
105    pub pending: bool,
106    pub unavailable: bool,
107}
108impl Authority {
109    pub fn generation(&self) -> [u8; 32] {
110        self.membership.current_generation().id
111    }
112    pub fn rotation_pending(&self) -> bool {
113        self.membership.rotation_pending()
114    }
115    pub fn key(&self, generation: [u8; 32]) -> Result<&LocalSharedStatePackageKey> {
116        self.keys.key(generation)
117    }
118    pub fn record_is_closed(&self, record: &[u8]) -> Result<bool> {
119        let e = codec::parse(record)?;
120        self.key(e.generation)?;
121        valid(e.vault == self.context.vault && e.stream == self.context.stream)?;
122        Ok(e.generation != self.generation())
123    }
124    /// Caller binds the lookup response to this authority and the parsed operation ID.
125    pub fn confirm_absent(&self, record: &[u8], response: &Reply) -> Result<AbsentOperation> {
126        self.validate()?;
127        valid(matches!(response, Reply::Absent))?;
128        codec::open(self, record)?;
129        Ok(AbsentOperation {
130            context: self.context.clone(),
131            record: record.to_vec(),
132        })
133    }
134    fn validate(&self) -> Result<()> {
135        self.keys.validate(&self.membership)?;
136        let b = self.membership.publication().binding();
137        valid(
138            self.context.vault == b.vault_id
139                && self.context.genesis == self.membership.genesis().commitment()
140                && self.context.head == self.membership.head()
141                && self.context.stream == b.stream_id
142                && self.context.descriptor == b.descriptor_commitment
143                && self.prefix == i64::try_from(b.prefix_count)?,
144        )
145    }
146}
147#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
148#[serde(deny_unknown_fields)]
149pub struct Mapping {
150    pub operation_id: String,
151    pub sequence: i64,
152    pub commitment: [u8; 32],
153}
154#[derive(Clone, Serialize, Deserialize)]
155#[serde(deny_unknown_fields)]
156pub struct Accepted {
157    pub mapping: Mapping,
158    #[serde(with = "crate::sync::base64_bytes")]
159    pub record: Vec<u8>,
160}
161#[derive(Clone, Serialize, Deserialize)]
162#[serde(deny_unknown_fields)]
163pub struct BatchFeatures {
164    pub count: usize,
165    pub bytes: usize,
166}
167
168#[derive(Clone, Serialize, Deserialize)]
169#[serde(deny_unknown_fields)]
170pub enum Operation {
171    Features,
172    Append {
173        ticket: Option<attachments::Ticket>,
174        #[serde(with = "crate::sync::base64_bytes")]
175        record: Vec<u8>,
176    },
177    Lookup {
178        operation_id: String,
179        expected: Option<Mapping>,
180    },
181    Pull {
182        after: i64,
183        limit: usize,
184        watermark: Option<i64>,
185    },
186}
187#[derive(Clone, Serialize, Deserialize)]
188#[serde(deny_unknown_fields)]
189pub struct Page {
190    pub after: i64,
191    pub watermark: i64,
192    pub cursor: i64,
193    pub has_more: bool,
194    pub records: Vec<Accepted>,
195}
196#[derive(Clone, Serialize, Deserialize)]
197#[serde(transparent)]
198pub struct BatchRecord(
199    #[serde(
200        serialize_with = "crate::sync::base64_bytes::serialize",
201        deserialize_with = "crate::sync::base64_bytes::bounded::<_, RECORD_LIMIT>"
202    )]
203    pub Vec<u8>,
204);
205
206#[derive(Clone, Serialize, Deserialize)]
207#[serde(deny_unknown_fields)]
208pub enum BatchOperation {
209    Append {
210        #[serde(deserialize_with = "batch::bounded_items")]
211        records: Vec<BatchRecord>,
212    },
213    Resolve {
214        #[serde(deserialize_with = "batch::bounded_items")]
215        operation_ids: Vec<String>,
216    },
217}
218
219#[derive(Clone, Serialize, Deserialize)]
220#[serde(deny_unknown_fields)]
221pub enum Resolution {
222    Found(batch::CompactMapping),
223    Absent { operation_id: String },
224    Bootstrap { operation_id: String },
225}
226
227#[derive(Clone, Serialize, Deserialize)]
228#[serde(deny_unknown_fields)]
229pub enum BatchReply {
230    Appended(#[serde(deserialize_with = "batch::bounded_items")] Vec<batch::CompactMapping>),
231    Resolved(#[serde(deserialize_with = "batch::bounded_items")] Vec<Resolution>),
232}
233
234#[derive(Clone, Serialize, Deserialize)]
235#[serde(deny_unknown_fields)]
236pub enum Reply {
237    Features(BatchFeatures),
238    Appended(Mapping),
239    Found(Accepted),
240    Absent,
241    Bootstrap,
242    Page(Page),
243}
244pub(crate) use crate::sync::codec::hash;
245pub(crate) fn valid(ok: bool) -> Result<()> {
246    ensure!(ok, "error encrypted-tail-invalid");
247    Ok(())
248}
249
250#[cfg(test)]
251mod tests;
252
253/// An authority over the fixed publication fixture and its initial key.
254#[cfg(any(test, feature = "test-support"))]
255pub(crate) fn fixture_authority() -> Authority {
256    let (membership, keys) = crate::sync::seed_claim::membership::test_support::content_authority();
257    let b = membership.publication().binding();
258    Authority {
259        context: Context {
260            vault: b.vault_id,
261            genesis: membership.genesis().commitment(),
262            device: [3; 32],
263            head: membership.head(),
264            stream: b.stream_id,
265            descriptor: b.descriptor_commitment,
266        },
267        prefix: b.prefix_count as i64,
268        membership,
269        keys,
270        association: "test".into(),
271        sync_generation: 1,
272    }
273}
274
275#[cfg(any(test, feature = "test-support"))]
276pub fn attachment_integrity_scan_metrics() -> (u64, u64) {
277    attachments::client::integrity_scan_metrics()
278}
279
280#[cfg(any(test, feature = "test-support"))]
281pub(crate) fn open_record(authority: &Authority, record: &[u8]) -> Result<super::wire::ChangeWire> {
282    codec::open(authority, record)
283}
284
285/// Re-seals an accepted record after `edit`, keeping its sequence.
286#[cfg(any(test, feature = "test-support"))]
287pub(crate) fn reseal(
288    authority: &Authority,
289    accepted: &Accepted,
290    edit: impl FnOnce(&mut super::wire::ChangeWire),
291) -> Result<Accepted> {
292    let mut change = codec::open(authority, &accepted.record)?;
293    edit(&mut change);
294    let record = codec::seal(authority, &change)?;
295    Ok(Accepted {
296        mapping: Mapping {
297            operation_id: change.change_id,
298            sequence: accepted.mapping.sequence,
299            commitment: hash(&record),
300        },
301        record,
302    })
303}
aven_core/sync/encrypted_tail.rs

aven_core/sync/
encrypted_tail.rs
