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;
19use anyhow::{Result, ensure};
20use serde::Serialize;
21use sha2::Sha256;
22
23pub use aven_protocol::wire::tail::{
24    APPEND_LIMIT, BATCH_APPEND_LIMIT, BATCH_BYTES, BATCH_CONTROL_LIMIT, BATCH_COUNT, CONTROL_LIMIT,
25    PAGE_BYTES, PAGE_COUNT, RECORD_LIMIT, RESPONSE_LIMIT,
26};
27
28/// An appended record's identity collides with a published prefix record.
29#[derive(Debug)]
30pub struct PrefixIdentityCollision;
31
32impl std::fmt::Display for PrefixIdentityCollision {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        f.write_str("error encrypted-tail-prefix-identity-collision")
35    }
36}
37
38impl std::error::Error for PrefixIdentityCollision {}
39
40pub use aven_protocol::wire::tail::Context;
41/// Protected host inputs. The host retains installation and authority exclusion.
42pub struct Authority {
43    pub context: Context,
44    pub membership: super::seed_claim::membership::Membership,
45    pub keys: super::seed_claim::membership::VerifiedKeys,
46    pub prefix: i64,
47    pub association: String,
48    pub sync_generation: i64,
49}
50/// Context-bound lookup absence. Only atomic replacement consumes this observation.
51pub struct AbsentOperation {
52    context: Context,
53    record: Vec<u8>,
54}
55/// Exact frozen bytes for one dispatch. Image heads include their staged upload.
56pub struct Push {
57    pub record: Vec<u8>,
58    pub upload: Option<attachments::Upload>,
59}
60/// Local round state read in one validated transaction.
61pub struct RoundState {
62    pub cursor: i64,
63    /// Finite page target captured while initial image catch-up is pending.
64    pub initial_watermark: Option<i64>,
65    /// No frozen or unacknowledged local metadata.
66    pub idle: bool,
67    pub upload_pending: bool,
68    /// Image demand, known only after initial image catch-up.
69    pub downloads: Option<Downloads>,
70}
71#[derive(Clone, Copy)]
72pub struct Downloads {
73    pub pending: bool,
74    pub unavailable: bool,
75}
76impl Authority {
77    pub fn generation(&self) -> [u8; 32] {
78        self.membership.current_generation().id
79    }
80    pub fn rotation_pending(&self) -> bool {
81        self.membership.rotation_pending()
82    }
83    pub fn key(&self, generation: [u8; 32]) -> Result<&LocalSharedStatePackageKey> {
84        self.keys.key(generation)
85    }
86    pub fn record_is_closed(&self, record: &[u8]) -> Result<bool> {
87        let e = codec::parse(record)?;
88        self.key(e.generation)?;
89        valid(e.vault == self.context.vault && e.stream == self.context.stream)?;
90        Ok(e.generation != self.generation())
91    }
92    /// Caller binds the lookup response to this authority and the parsed operation ID.
93    pub fn confirm_absent(&self, record: &[u8], response: &Reply) -> Result<AbsentOperation> {
94        self.validate()?;
95        valid(matches!(response, Reply::Absent))?;
96        codec::open(self, record)?;
97        Ok(AbsentOperation {
98            context: self.context.clone(),
99            record: record.to_vec(),
100        })
101    }
102    fn validate(&self) -> Result<()> {
103        self.keys.validate(&self.membership)?;
104        let b = self.membership.publication().binding();
105        valid(
106            self.context.vault == b.vault_id
107                && self.context.genesis == self.membership.genesis().commitment()
108                && self.context.head == self.membership.head()
109                && self.context.stream == b.stream_id
110                && self.context.descriptor == b.descriptor_commitment
111                && self.prefix == i64::try_from(b.prefix_count)?,
112        )
113    }
114}
115pub(crate) use crate::sync::codec::hash;
116pub use aven_protocol::wire::tail::{
117    Accepted, BatchFeatures, BatchOperation, BatchRecord, BatchReply, Mapping, Operation, Page,
118    Reply, Resolution,
119};
120pub(crate) fn valid(ok: bool) -> Result<()> {
121    ensure!(ok, "error encrypted-tail-invalid");
122    Ok(())
123}
124
125#[cfg(test)]
126mod tests;
127
128/// An authority over the fixed publication fixture and its initial key.
129#[cfg(any(test, feature = "test-support"))]
130pub(crate) fn fixture_authority() -> Authority {
131    let (membership, keys) = crate::sync::seed_claim::membership::test_support::content_authority();
132    let b = membership.publication().binding();
133    Authority {
134        context: Context {
135            vault: b.vault_id,
136            genesis: membership.genesis().commitment(),
137            device: [3; 32],
138            head: membership.head(),
139            stream: b.stream_id,
140            descriptor: b.descriptor_commitment,
141        },
142        prefix: b.prefix_count as i64,
143        membership,
144        keys,
145        association: "test".into(),
146        sync_generation: 1,
147    }
148}
149
150#[cfg(any(test, feature = "test-support"))]
151pub fn attachment_integrity_scan_metrics() -> (u64, u64) {
152    attachments::client::integrity_scan_metrics()
153}
154
155#[cfg(any(test, feature = "test-support"))]
156pub(crate) fn open_record(authority: &Authority, record: &[u8]) -> Result<super::wire::ChangeWire> {
157    codec::open(authority, record)
158}
159
160/// Re-seals an accepted record after `edit`, keeping its sequence.
161#[cfg(any(test, feature = "test-support"))]
162pub(crate) fn reseal(
163    authority: &Authority,
164    accepted: &Accepted,
165    edit: impl FnOnce(&mut super::wire::ChangeWire),
166) -> Result<Accepted> {
167    let mut change = codec::open(authority, &accepted.record)?;
168    edit(&mut change);
169    let record = codec::seal(authority, &change)?;
170    Ok(Accepted {
171        mapping: Mapping {
172            operation_id: change.change_id,
173            sequence: accepted.mapping.sequence,
174            commitment: hash(&record),
175        },
176        record,
177    })
178}
aven_core/sync/encrypted_tail.rs

aven_core/sync/
encrypted_tail.rs
