aven_core/sync/shared_state/
adoption.rs

1//! Original-installation adoption. Never installs the captured domain snapshot.
2use anyhow::{Context, Result, ensure};
3use serde::{Deserialize, Serialize};
4use sha2::{Digest, Sha256};
5use sqlx::SqliteConnection;
6use std::borrow::Cow;
7
8use super::validated::{FreezeBinding, ProofCache, VALIDATION_VERSION, ValidatedSeed};
9use super::{NeverDispatchedLocalSharedCapture, package};
10use crate::db::{self, Database};
11use crate::sync::LocalSharedStatePackageKey;
12use crate::sync::seed_claim::{Genesis, Publication, PublicationOutcome, SeedAuthority};
13
14/// Host-persisted source identity, independent of replaceable SQLite state.
15#[derive(Clone, Debug, PartialEq, Eq)]
16pub struct SeedSourceAuthority(Vec<u8>);
17
18impl SeedSourceAuthority {
19    pub fn generate(account: [u8; 32], genesis: &Genesis) -> Result<Self> {
20        let mut bytes = b"AVENSRC1".to_vec();
21        bytes.extend(account);
22        let mut incarnation = [0; 32];
23        getrandom::fill(&mut incarnation).context("error seed-source-entropy")?;
24        bytes.extend(incarnation);
25        bytes.extend(genesis.commitment());
26        Ok(Self(bytes))
27    }
28
29    pub fn from_protected_storage(
30        bytes: &[u8],
31        account: [u8; 32],
32        genesis: &Genesis,
33    ) -> Result<Self> {
34        ensure!(
35            bytes.len() == 104
36                && &bytes[..8] == b"AVENSRC1"
37                && bytes[8..40] == account
38                && bytes[72..] == genesis.commitment(),
39            "error seed-source-authority-mismatch"
40        );
41        Ok(Self(bytes.to_vec()))
42    }
43
44    pub fn protected_storage_bytes(&self) -> &[u8] {
45        &self.0
46    }
47}
48
49#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
50#[serde(deny_unknown_fields)]
51struct IntentData {
52    source: Vec<u8>,
53    client: String,
54    generation: i64,
55    candidate: String,
56    descriptor: Vec<u8>,
57    publication: Vec<u8>,
58    history: [u8; 32],
59    /// Absent only in intents written before freezes were bound; those fail
60    /// closed rather than being trusted or re-signed.
61    #[serde(default, skip_serializing_if = "Option::is_none")]
62    freeze: Option<FreezeBinding>,
63}
64
65/// Exact local intent, not permission to dispatch or evidence of server acceptance.
66#[derive(Clone, Debug)]
67pub struct SeedPublicationIntent {
68    bytes: Vec<u8>,
69    data: IntentData,
70}
71
72impl SeedPublicationIntent {
73    pub fn from_protected_storage(
74        bytes: &[u8],
75        source: &SeedSourceAuthority,
76        genesis: &Genesis,
77    ) -> Result<Self> {
78        ensure!(bytes.len() <= 65536, "error seed-intent-too-large");
79        let data: IntentData =
80            serde_json::from_slice(bytes).context("error seed-intent-corrupt")?;
81        ensure!(
82            source.0[72..] == genesis.commitment()
83                && data.source == source.0
84                && serde_json::to_vec(&data)? == bytes,
85            "error seed-intent-source-mismatch"
86        );
87        let publication = Publication::from_record(genesis, &data.descriptor, &data.publication)?;
88        ensure!(
89            hex::encode(publication.binding().bootstrap_id) == data.candidate,
90            "error seed-intent-candidate-mismatch"
91        );
92        Ok(Self {
93            bytes: bytes.to_vec(),
94            data,
95        })
96    }
97
98    pub fn protected_storage_bytes(&self) -> &[u8] {
99        &self.bytes
100    }
101    pub(crate) fn freeze_binding(&self) -> Option<FreezeBinding> {
102        self.data.freeze
103    }
104
105    /// Fails unless `proof` is the proof this intent's binding records.
106    fn check_proof(&self, proof: &ValidatedSeed) -> Result<()> {
107        let binding = self.data.freeze.context("error seed-freeze-unsupported")?;
108        ensure!(
109            binding.validation_version == VALIDATION_VERSION,
110            "error seed-intent-receipt-unsupported"
111        );
112        ensure!(
113            proof.identity().candidate() == self.data.candidate
114                && proof.identity().history() == self.data.history
115                && proof.descriptor() == self.data.descriptor
116                && *proof.binding() == binding,
117            "error seed-package-mismatch"
118        );
119        Ok(())
120    }
121
122    pub fn descriptor(&self) -> &[u8] {
123        &self.data.descriptor
124    }
125
126    pub fn publication(&self, genesis: &Genesis) -> Result<Publication> {
127        Publication::from_record(genesis, &self.data.descriptor, &self.data.publication)
128    }
129}
130
131pub(crate) async fn ensure_unbound(conn: &mut SqliteConnection) -> Result<()> {
132    let bound: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_source) OR EXISTS(SELECT 1 FROM local_seed_publication_intent)").fetch_one(&mut *conn).await?;
133    let peer_bound: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_enrollment)")
134        .fetch_one(&mut *conn)
135        .await?;
136    ensure!(
137        !bound && !peer_bound,
138        "error e2ee-installation-fenced encrypted-tail-and-replacement-unavailable"
139    );
140    Ok(())
141}
142
143pub(super) async fn ensure_no_intent(conn: &mut SqliteConnection) -> Result<()> {
144    let exists: bool =
145        sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_publication_intent) OR EXISTS(SELECT 1 FROM local_shared_capture_journal WHERE publication_owned = 1)")
146            .fetch_one(conn)
147            .await?;
148    ensure!(
149        !exists,
150        "error seed-publication-intent-owned local-cancellation-unavailable"
151    );
152    Ok(())
153}
154
155async fn source_matches(
156    conn: &mut SqliteConnection,
157    source: &SeedSourceAuthority,
158) -> Result<String> {
159    let (bytes, client): (Vec<u8>, String) =
160        sqlx::query_as("SELECT authority, client_id FROM local_seed_source WHERE singleton = 1")
161            .fetch_optional(&mut *conn)
162            .await?
163            .context("error seed-source-missing")?;
164    ensure!(
165        bytes == source.0 && db::get_meta(conn, "client_id").await?.as_deref() == Some(&client),
166        "error seed-source-mismatch"
167    );
168    Ok(client)
169}
170
171async fn generation(conn: &mut SqliteConnection) -> Result<i64> {
172    db::get_meta(conn, "sync_generation")
173        .await?
174        .context("error seed-generation-missing")?
175        .parse()
176        .context("error seed-generation-invalid")
177}
178
179/// Packs the source history map: changes sorted by ID, each payload embedded
180/// as its JSON text. Readers compare payloads semantically, so historical
181/// meaning does not depend on JSON whitespace.
182pub(super) fn pack_history(
183    rows: &[crate::data_safety::export_types::ChangeRow],
184) -> Result<Vec<u8>> {
185    /// A source history row as stored, embedding its payload JSON verbatim.
186    #[derive(Serialize)]
187    struct Stored<'a> {
188        change_id: &'a str,
189        client_id: &'a str,
190        local_seq: i64,
191        entity_type: &'a str,
192        entity_id: &'a str,
193        field: Option<&'a str>,
194        op_type: &'a str,
195        payload: &'a serde_json::value::RawValue,
196        base_version: Option<&'a str>,
197        created_at: &'a str,
198        server_seq: Option<i64>,
199    }
200    let mut stored = rows
201        .iter()
202        .map(|row| {
203            Ok(Stored {
204                change_id: &row.change_id,
205                client_id: &row.client_id,
206                local_seq: row.local_seq,
207                entity_type: &row.entity_type,
208                entity_id: &row.entity_id,
209                field: row.field.as_deref(),
210                op_type: &row.op_type,
211                payload: serde_json::from_str(&row.payload)?,
212                base_version: row.base_version.as_deref(),
213                created_at: &row.created_at,
214                server_seq: row.server_seq,
215            })
216        })
217        .collect::<Result<Vec<_>>>()?;
218    stored.sort_by(|a, b| a.change_id.cmp(b.change_id));
219    super::pack_document(&stored)
220}
221
222/// One row of the stored source history map, borrowing its text.
223#[derive(Deserialize)]
224#[serde(deny_unknown_fields)]
225struct SourceChange<'a> {
226    #[serde(borrow)]
227    change_id: Cow<'a, str>,
228    #[serde(borrow)]
229    client_id: Cow<'a, str>,
230    local_seq: i64,
231    #[serde(borrow)]
232    entity_type: Cow<'a, str>,
233    #[serde(borrow)]
234    entity_id: Cow<'a, str>,
235    #[serde(borrow)]
236    field: Option<Cow<'a, str>>,
237    #[serde(borrow)]
238    op_type: Cow<'a, str>,
239    #[serde(borrow)]
240    payload: &'a serde_json::value::RawValue,
241    #[serde(borrow)]
242    base_version: Option<Cow<'a, str>>,
243    #[serde(borrow)]
244    created_at: Cow<'a, str>,
245    server_seq: Option<i64>,
246}
247
248/// One live `changes` row, decoded from a JSON array in column order.
249#[derive(Deserialize)]
250struct LiveChange<'a> {
251    #[serde(borrow)]
252    change_id: Cow<'a, str>,
253    #[serde(borrow)]
254    client_id: Cow<'a, str>,
255    local_seq: i64,
256    #[serde(borrow)]
257    entity_type: Cow<'a, str>,
258    #[serde(borrow)]
259    entity_id: Cow<'a, str>,
260    #[serde(borrow)]
261    field: Option<Cow<'a, str>>,
262    #[serde(borrow)]
263    op_type: Cow<'a, str>,
264    #[serde(borrow)]
265    payload: Cow<'a, str>,
266    #[serde(borrow)]
267    base_version: Option<Cow<'a, str>>,
268    #[serde(borrow)]
269    created_at: Cow<'a, str>,
270    server_seq: Option<i64>,
271}
272
273impl<'a> From<&'a crate::data_safety::ChangeRow> for LiveChange<'a> {
274    fn from(row: &'a crate::data_safety::ChangeRow) -> Self {
275        Self {
276            change_id: Cow::Borrowed(&row.change_id),
277            client_id: Cow::Borrowed(&row.client_id),
278            local_seq: row.local_seq,
279            entity_type: Cow::Borrowed(&row.entity_type),
280            entity_id: Cow::Borrowed(&row.entity_id),
281            field: row.field.as_deref().map(Cow::Borrowed),
282            op_type: Cow::Borrowed(&row.op_type),
283            payload: Cow::Borrowed(&row.payload),
284            base_version: row.base_version.as_deref().map(Cow::Borrowed),
285            created_at: Cow::Borrowed(&row.created_at),
286            server_seq: row.server_seq,
287        }
288    }
289}
290
291impl SourceChange<'_> {
292    /// Compares every field, the original server sequence only when asked.
293    /// Payloads match byte-for-byte first and semantically otherwise.
294    fn matches(&self, row: &LiveChange<'_>, with_server_seq: bool) -> Result<bool> {
295        if self.change_id != row.change_id
296            || self.client_id != row.client_id
297            || self.local_seq != row.local_seq
298            || self.entity_type != row.entity_type
299            || self.entity_id != row.entity_id
300            || self.field.as_deref() != row.field.as_deref()
301            || self.op_type != row.op_type
302            || self.base_version.as_deref() != row.base_version.as_deref()
303            || self.created_at != row.created_at
304            || (with_server_seq && self.server_seq != row.server_seq)
305        {
306            return Ok(false);
307        }
308        let source = self.payload.get();
309        if source == row.payload {
310            return Ok(true);
311        }
312        Ok(serde_json::from_str::<serde_json::Value>(source)?
313            == serde_json::from_str::<serde_json::Value>(&row.payload)?)
314    }
315}
316
317const HISTORY_PAGE: i64 = 1024;
318
319async fn validate_history(
320    conn: &mut SqliteConnection,
321    candidate: &str,
322    source: &SeedSourceAuthority,
323) -> Result<[u8; 32]> {
324    // Source authority, history, provenance and sync generation.
325    type CaptureRow = (Option<Vec<u8>>, Option<Vec<u8>>, Option<Vec<u8>>, i64);
326    let (capture_source, history, stored_provenance, captured_generation): CaptureRow =
327        sqlx::query_as(
328            "SELECT j.source_authority, d.source_history, d.source_provenance, j.sync_generation
329         FROM local_shared_capture_journal j
330         JOIN local_shared_capture_documents d USING (candidate_id)
331         WHERE j.candidate_id = ?",
332        )
333        .bind(candidate)
334        .fetch_optional(&mut *conn)
335        .await?
336        .context("error seed-capture-missing")?;
337    ensure!(
338        capture_source.as_deref() == Some(source.0.as_slice())
339            && captured_generation == generation(conn).await?,
340        "error seed-capture-source-changed"
341    );
342    let (Some(stored_history), Some(stored_provenance)) = (history, stored_provenance) else {
343        anyhow::bail!("error seed-capture-incompatible recapture-never-dispatched");
344    };
345    let history = super::unpack_document(&stored_history)?;
346    let provenance = super::unpack_document(&stored_provenance)?;
347    let expected: Vec<SourceChange<'_>> =
348        serde_json::from_str(&history).context("error seed-history-invalid")?;
349    ensure!(
350        expected
351            .windows(2)
352            .all(|pair| pair[0].change_id < pair[1].change_id),
353        "error seed-history-invalid"
354    );
355
356    let captured: i64 = sqlx::query_scalar(
357        "SELECT count(*) FROM local_shared_capture_changes WHERE candidate_id = ?",
358    )
359    .bind(candidate)
360    .fetch_one(&mut *conn)
361    .await?;
362    ensure!(
363        usize::try_from(captured)? == expected.len(),
364        "error seed-captured-history-map-mismatch"
365    );
366    // Captured IDs are exactly the source IDs, so an ordered walk of the live
367    // rows joined to them must meet every source row once, in the same order.
368    let mut pending = expected.iter();
369    let mut after = String::new();
370    let mut accepted = 0_i64;
371    loop {
372        // One JSON array per page: decoding thousands of rows one by one
373        // through the driver costs several times the query itself.
374        let page: String = sqlx::query_scalar(
375            "SELECT json_group_array(json_array(
376                        change_id, client_id, local_seq, entity_type, entity_id, field,
377                        op_type, payload, base_version, created_at, server_seq))
378             FROM (SELECT ch.* FROM local_shared_capture_changes c
379                   JOIN changes ch ON ch.change_id = c.change_id
380                   WHERE c.candidate_id = ? AND c.change_id > ?
381                   ORDER BY c.change_id LIMIT ?)",
382        )
383        .bind(candidate)
384        .bind(&after)
385        .bind(HISTORY_PAGE)
386        .fetch_one(&mut *conn)
387        .await?;
388        let page: Vec<LiveChange<'_>> =
389            serde_json::from_str(&page).context("error seed-captured-history-changed")?;
390        let Some(last) = page.last() else {
391            break;
392        };
393        after = last.change_id.to_string();
394        for row in &page {
395            let source_row = pending
396                .next()
397                .context("error seed-captured-history-changed")?;
398            ensure!(
399                source_row.matches(row, true)?,
400                "error seed-captured-history-changed"
401            );
402            accepted += i64::from(row.server_seq.is_some());
403        }
404    }
405    ensure!(
406        pending.next().is_none(),
407        "error seed-captured-history-changed"
408    );
409
410    // Parsed once and compared in memory: joining SQL against json_each
411    // re-parses the whole document for every row.
412    #[derive(Deserialize)]
413    struct Provenance {
414        change_id: String,
415        source_server_seq: Option<i64>,
416        source_pending_rank: Option<i64>,
417    }
418    let source: Vec<Provenance> =
419        serde_json::from_str(&provenance).context("error seed-history-invalid")?;
420    let current: String = sqlx::query_scalar(
421        "SELECT json_group_array(json_array(
422                    p.change_id, p.source_server_seq, p.source_pending_rank))
423         FROM shared_history_provenance p
424         JOIN local_shared_capture_changes c
425           ON c.candidate_id = ? AND c.change_id = p.change_id",
426    )
427    .bind(candidate)
428    .fetch_one(&mut *conn)
429    .await?;
430    let current: Vec<(Cow<'_, str>, Option<i64>, Option<i64>)> =
431        serde_json::from_str(&current).context("error seed-source-provenance-changed")?;
432    let current: std::collections::HashMap<&str, (Option<i64>, Option<i64>)> = current
433        .iter()
434        .map(|(id, seq, rank)| (id.as_ref(), (*seq, *rank)))
435        .collect();
436    // Every source row must match a current row of a captured change, and
437    // no captured change may carry provenance the source lacks. Provenance of
438    // uncaptured changes appears in neither set.
439    let unique: std::collections::HashSet<&str> =
440        source.iter().map(|row| row.change_id.as_str()).collect();
441    let provenance_changed = unique.len() != source.len()
442        || source.len() != current.len()
443        || source.iter().any(|row| {
444            current.get(row.change_id.as_str())
445                != Some(&(row.source_server_seq, row.source_pending_rank))
446        });
447    ensure!(!provenance_changed, "error seed-source-provenance-changed");
448
449    // The walk met every captured row once, so any other accepted row is
450    // uncaptured. Counting uses the server_seq index instead of scanning
451    // every change with its payload.
452    let all_accepted: i64 =
453        sqlx::query_scalar("SELECT count(*) FROM changes WHERE server_seq IS NOT NULL")
454            .fetch_one(&mut *conn)
455            .await?;
456    ensure!(
457        all_accepted == accepted,
458        "error seed-uncaptured-accepted-history"
459    );
460
461    Ok(history_digest(&stored_history, &stored_provenance))
462}
463
464/// Commits to the stored history and provenance documents as bytes, so any
465/// rewrite of them changes it, even one that decodes to the same text.
466pub(super) fn history_digest(history: &[u8], provenance: &[u8]) -> [u8; 32] {
467    let mut digest = Sha256::new();
468    digest.update(b"aven-local-source-history-v1");
469    digest.update((history.len() as u64).to_be_bytes());
470    digest.update(history);
471    digest.update(provenance);
472    digest.finalize().into()
473}
474
475/// Checks the stored source history map against the frozen capture, ignoring
476/// original server sequences, and returns the history commitment.
477pub(super) async fn check_frozen_history(
478    conn: &mut SqliteConnection,
479    capture: &NeverDispatchedLocalSharedCapture,
480) -> Result<[u8; 32]> {
481    let (history, provenance): (Option<Vec<u8>>, Option<Vec<u8>>) = sqlx::query_as(
482        "SELECT source_history, source_provenance
483         FROM local_shared_capture_documents WHERE candidate_id = ?",
484    )
485    .bind(capture.candidate_id())
486    .fetch_optional(&mut *conn)
487    .await?
488    .context("error seed-capture-missing")?;
489    let (Some(history), Some(provenance)) = (history, provenance) else {
490        anyhow::bail!("error seed-capture-incompatible recapture-never-dispatched");
491    };
492    let text = super::unpack_document(&history)?;
493    let expected: Vec<SourceChange<'_>> =
494        serde_json::from_str(&text).context("error seed-history-invalid")?;
495    let mut frozen = capture
496        .capture
497        .snapshot
498        .tables
499        .changes
500        .iter()
501        .collect::<Vec<_>>();
502    frozen.sort_by(|a, b| a.change_id.cmp(&b.change_id));
503    ensure!(
504        frozen.len() == expected.len(),
505        "error seed-captured-history-map-mismatch"
506    );
507    for (source_row, frozen_row) in expected.iter().zip(&frozen) {
508        ensure!(
509            source_row.matches(&LiveChange::from(*frozen_row), false)?,
510            "error seed-captured-history-map-mismatch"
511        );
512    }
513    Ok(history_digest(&history, &provenance))
514}
515
516/// Records captured provenance and moves captured rows to their prefix
517/// ranks. Rows outside the capture keep their state.
518async fn adopt_captured_history(
519    conn: &mut SqliteConnection,
520    candidate: &str,
521    prefix_count: u64,
522) -> Result<()> {
523    let captured: i64 = sqlx::query_scalar(
524        "SELECT count(*) FROM local_shared_capture_changes WHERE candidate_id = ?",
525    )
526    .bind(candidate)
527    .fetch_one(&mut *conn)
528    .await?;
529    ensure!(
530        u64::try_from(captured)? == prefix_count,
531        "error seed-prefix-count-mismatch"
532    );
533    let provenance_conflict: bool = sqlx::query_scalar(
534        "SELECT EXISTS(
535             SELECT 1 FROM local_shared_capture_changes c
536             JOIN shared_history_provenance p USING (change_id)
537             WHERE c.candidate_id = ?
538               AND (c.source_server_seq IS NOT p.source_server_seq
539                    OR c.source_pending_rank IS NOT p.source_pending_rank)
540         )",
541    )
542    .bind(candidate)
543    .fetch_one(&mut *conn)
544    .await?;
545    ensure!(!provenance_conflict, "error seed-provenance-changed");
546    sqlx::query(
547        "INSERT OR IGNORE INTO shared_history_provenance(
548             change_id, source_server_seq, source_pending_rank
549         )
550         SELECT change_id, source_server_seq, source_pending_rank
551         FROM local_shared_capture_changes WHERE candidate_id = ?",
552    )
553    .bind(candidate)
554    .execute(&mut *conn)
555    .await?;
556    // The unique server_seq index requires clearing old ranks before
557    // assigning the new ones. Rows already at their rank, usually most of
558    // them, are left alone: rewriting a change rewrites its whole payload.
559    sqlx::query(
560        "UPDATE changes SET server_seq = NULL
561         WHERE server_seq IS NOT NULL AND EXISTS(
562             SELECT 1 FROM local_shared_capture_changes c
563             WHERE c.candidate_id = ? AND c.change_id = changes.change_id
564               AND c.prefix_rank != changes.server_seq
565         )",
566    )
567    .bind(candidate)
568    .execute(&mut *conn)
569    .await?;
570    sqlx::query(
571        "UPDATE changes SET server_seq = (
572             SELECT prefix_rank FROM local_shared_capture_changes c
573             WHERE c.candidate_id = ?1 AND c.change_id = changes.change_id
574         )
575         WHERE server_seq IS NULL AND change_id IN
576         (SELECT change_id FROM local_shared_capture_changes WHERE candidate_id = ?1)",
577    )
578    .bind(candidate)
579    .execute(&mut *conn)
580    .await?;
581    let ranked: i64 = sqlx::query_scalar(
582        "SELECT count(*) FROM local_shared_capture_changes c
583         JOIN changes ch ON ch.change_id = c.change_id
584         WHERE c.candidate_id = ? AND ch.server_seq = c.prefix_rank",
585    )
586    .bind(candidate)
587    .fetch_one(&mut *conn)
588    .await?;
589    ensure!(ranked == captured, "error seed-history-coverage-changed");
590    Ok(())
591}
592
593impl Database {
594    /// Loads exact upload bytes only for a sealed, protected publication intent.
595    /// Adopted installations need only their retained intent and remote outcome.
596    pub(crate) async fn seed_publication_upload(
597        &self,
598        source: &SeedSourceAuthority,
599        intent: &SeedPublicationIntent,
600        seed: &SeedAuthority,
601        key: &LocalSharedStatePackageKey,
602        proofs: &mut ProofCache,
603    ) -> Result<Option<package::upload::FrozenUpload>> {
604        let proof = match self.seed_publication_intent_bytes().await? {
605            Some((_, state)) if state == "adopted" => None,
606            // Failures surface only where the writer would have validated.
607            _ => Some(proofs.get(self, key, seed.genesis().commitment()).await),
608        };
609        let mut conn = self.acquire_writer().await?;
610        let mut tx = db::begin_immediate(&mut conn).await?;
611        source_matches(&mut tx, source).await?;
612        let (stored, state): (Vec<u8>, String) = sqlx::query_as(
613            "SELECT intent, state FROM local_seed_publication_intent WHERE singleton = 1",
614        )
615        .fetch_optional(&mut *tx)
616        .await?
617        .context("error seed-intent-missing")?;
618        ensure!(stored == intent.bytes, "error seed-intent-mismatch");
619        if state == "adopted" {
620            return Ok(None);
621        }
622        ensure!(
623            state == "sealed" && intent.data.generation == generation(&mut tx).await?,
624            "error seed-intent-not-sealed"
625        );
626        ensure!(
627            validate_history(&mut tx, &intent.data.candidate, source).await? == intent.data.history,
628            "error seed-history-commitment-mismatch"
629        );
630        let proof = proof.context("error seed-intent-changed")??;
631        proof.identity().assert_matches(&mut tx).await?;
632        intent.check_proof(proof)?;
633        let upload = package::upload::FrozenUpload::open(&mut tx, &intent.data.candidate).await?;
634        ensure!(
635            upload.descriptor() == proof.descriptor()
636                && upload.catalogs() == &proof.metadata().catalogs,
637            "error seed-package-mismatch"
638        );
639        tx.commit().await?;
640        // Records stay in SQLite; each is verified against these catalogs as
641        // it is read for upload.
642        Ok(Some(upload))
643    }
644
645    pub async fn seed_source_pin(&self) -> Result<Option<Vec<u8>>> {
646        let mut conn = self.acquire_reader().await?;
647        Ok(
648            sqlx::query_scalar("SELECT authority FROM local_seed_source WHERE singleton = 1")
649                .fetch_optional(&mut *conn)
650                .await?,
651        )
652    }
653
654    /// Called under the installation interlock after protected persistence.
655    pub async fn bind_seed_source(
656        &self,
657        source: &SeedSourceAuthority,
658        installation: &db::installation::InstallationGuard,
659    ) -> Result<()> {
660        ensure!(
661            self.file_identity() == Some(installation.identity()),
662            "error seed-source-installation-mismatch"
663        );
664        installation.fence()?;
665        let mut conn = self.acquire_writer().await?;
666        let mut tx = db::begin_immediate(&mut conn).await?;
667        let existing: Option<Vec<u8>> =
668            sqlx::query_scalar("SELECT authority FROM local_seed_source WHERE singleton = 1")
669                .fetch_optional(&mut *tx)
670                .await?;
671        if let Some(existing) = existing {
672            ensure!(existing == source.0, "error seed-source-mismatch");
673            source_matches(&mut tx, source).await?;
674        } else {
675            super::ensure_no_active_local_shared_capture(&mut tx).await?;
676            let client = db::get_meta(&mut tx, "client_id")
677                .await?
678                .context("missing client identity")?;
679            sqlx::query("INSERT INTO local_seed_source VALUES (1, ?, ?)")
680                .bind(&source.0)
681                .bind(client)
682                .execute(&mut *tx)
683                .await?;
684        }
685        tx.commit().await?;
686        Ok(())
687    }
688
689    pub async fn seed_publication_intent_bytes(&self) -> Result<Option<(Vec<u8>, String)>> {
690        let mut conn = self.acquire_reader().await?;
691        Ok(sqlx::query_as(
692            "SELECT intent, state FROM local_seed_publication_intent WHERE singleton = 1",
693        )
694        .fetch_optional(&mut *conn)
695        .await?)
696    }
697
698    /// Commits cancellation exclusion before any protected intent is exposed.
699    /// Retrying returns exact persisted bytes, never a newly signed context.
700    pub async fn prepare_seed_publication_intent(
701        &self,
702        source: &SeedSourceAuthority,
703        seed: &SeedAuthority,
704        key: &LocalSharedStatePackageKey,
705    ) -> Result<SeedPublicationIntent> {
706        self.prepare_seed_publication_intent_with(source, seed, key, &mut ProofCache::default())
707            .await
708    }
709
710    pub(crate) async fn prepare_seed_publication_intent_with(
711        &self,
712        source: &SeedSourceAuthority,
713        seed: &SeedAuthority,
714        key: &LocalSharedStatePackageKey,
715        proofs: &mut ProofCache,
716    ) -> Result<SeedPublicationIntent> {
717        let proof = match self.seed_publication_intent_bytes().await? {
718            Some((_, state)) if state == "adopted" => None,
719            // Failures surface only where the writer would have validated.
720            _ => Some(proofs.get(self, key, seed.genesis().commitment()).await),
721        };
722        let mut conn = self.acquire_writer().await?;
723        let mut tx = db::begin_immediate(&mut conn).await?;
724        let client = source_matches(&mut tx, source).await?;
725        let pin: Vec<u8> =
726            sqlx::query_scalar("SELECT commitment FROM local_seed_genesis_pin WHERE singleton = 1")
727                .fetch_optional(&mut *tx)
728                .await?
729                .context("error seed-genesis-pin-missing")?;
730        ensure!(
731            pin == seed.genesis().commitment(),
732            "error seed-genesis-pin-mismatch"
733        );
734        let existing: Option<(Vec<u8>, String)> = sqlx::query_as(
735            "SELECT intent, state FROM local_seed_publication_intent WHERE singleton = 1",
736        )
737        .fetch_optional(&mut *tx)
738        .await?;
739        if let Some((bytes, state)) = existing {
740            let intent =
741                SeedPublicationIntent::from_protected_storage(&bytes, source, seed.genesis())?;
742            if state != "adopted" {
743                ensure!(
744                    intent.data.generation == generation(&mut tx).await?
745                        && intent.data.history
746                            == validate_history(&mut tx, &intent.data.candidate, source).await?,
747                    "error seed-intent-source-changed"
748                );
749                let proof = proof.context("error seed-intent-changed")??;
750                proof.identity().assert_matches(&mut tx).await?;
751                intent.check_proof(proof)?;
752                tx.commit().await?;
753            }
754            return Ok(intent);
755        }
756        ensure_no_intent(&mut tx).await?;
757        let proof = proof.context("error seed-intent-changed")??;
758        let candidate = proof.identity().candidate().to_string();
759        let history = validate_history(&mut tx, &candidate, source).await?;
760        proof.identity().assert_matches(&mut tx).await?;
761        ensure!(
762            history == proof.identity().history(),
763            "error seed-capture-changed"
764        );
765        // The package was authenticated under `key` against the capture.
766        let publication = seed.sign_authenticated_publication(proof.descriptor(), key)?;
767        let data = IntentData {
768            source: source.0.clone(),
769            client,
770            generation: generation(&mut tx).await?,
771            candidate,
772            descriptor: proof.descriptor().to_vec(),
773            publication: publication.record().to_vec(),
774            history,
775            freeze: Some(*proof.binding()),
776        };
777        let bytes = serde_json::to_vec(&data)?;
778        sqlx::query("INSERT INTO local_seed_publication_intent(singleton, candidate_id, intent, state) VALUES (1, ?, ?, 'preparing')").bind(&data.candidate).bind(&bytes).execute(&mut *tx).await?;
779        sqlx::query(
780            "UPDATE local_shared_capture_journal SET publication_owned = 1 WHERE candidate_id = ?",
781        )
782        .bind(&data.candidate)
783        .execute(&mut *tx)
784        .await?;
785        #[cfg(test)]
786        wait_intent_boundary(&data.candidate).await;
787        tx.commit().await?;
788        SeedPublicationIntent::from_protected_storage(&bytes, source, seed.genesis())
789    }
790
791    /// Hosts supply an intent reloaded from protected storage, never a boolean.
792    pub async fn seal_seed_publication_intent(
793        &self,
794        source: &SeedSourceAuthority,
795        intent: &SeedPublicationIntent,
796    ) -> Result<()> {
797        let mut conn = self.acquire_writer().await?;
798        let mut tx = db::begin_immediate(&mut conn).await?;
799        source_matches(&mut tx, source).await?;
800        ensure!(
801            intent.data.source == source.0
802                && intent.data.generation == generation(&mut tx).await?
803                && intent.data.history
804                    == validate_history(&mut tx, &intent.data.candidate, source).await?,
805            "error seed-intent-source-changed"
806        );
807        let commitment: Vec<u8> = sqlx::query_scalar("SELECT frozen_descriptor_commitment FROM local_shared_capture_journal WHERE candidate_id = ?").bind(&intent.data.candidate).fetch_one(&mut *tx).await?;
808        ensure!(
809            commitment == Sha256::digest(&intent.data.descriptor).as_slice(),
810            "error seed-frozen-descriptor-changed"
811        );
812        let changed = sqlx::query("UPDATE local_seed_publication_intent SET state = 'sealed' WHERE singleton = 1 AND intent = ? AND state IN ('preparing', 'sealed')").bind(&intent.bytes).execute(&mut *tx).await?.rows_affected();
813        ensure!(changed == 1, "error seed-intent-cas-failed");
814        tx.commit().await?;
815        Ok(())
816    }
817
818    /// Atomically adopts the original seed's history, never its old domain image.
819    /// Returns false for an already committed matching adoption without rewinding.
820    pub(crate) async fn adopt_seed_publication(
821        &self,
822        source: &SeedSourceAuthority,
823        intent: &SeedPublicationIntent,
824        seed: &SeedAuthority,
825        key: &LocalSharedStatePackageKey,
826        outcome: &PublicationOutcome,
827        proofs: &mut ProofCache,
828    ) -> Result<bool> {
829        let verified =
830            SeedPublicationIntent::from_protected_storage(&intent.bytes, source, seed.genesis())?;
831        outcome.validate_expected(seed.genesis(), &verified.data.descriptor)?;
832        ensure!(
833            outcome.publication().record().as_slice() == verified.data.publication,
834            "error seed-publication-outcome-mismatch"
835        );
836        let binding = outcome.publication().binding();
837        let association = format!(
838            "{}:{}:{}",
839            hex::encode(binding.vault_id),
840            hex::encode(binding.stream_id),
841            hex::encode(binding.bootstrap_id)
842        );
843        let validated = match self.seed_publication_intent_bytes().await? {
844            Some((_, state)) if state == "adopted" => None,
845            // Failures surface only where the writer would have validated.
846            _ => Some(proofs.get(self, key, seed.genesis().commitment()).await),
847        };
848        let mut conn = self.acquire_writer().await?;
849        let mut tx = db::begin_immediate(&mut conn).await?;
850        let pin: Vec<u8> =
851            sqlx::query_scalar("SELECT commitment FROM local_seed_genesis_pin WHERE singleton = 1")
852                .fetch_optional(&mut *tx)
853                .await?
854                .context("error seed-genesis-pin-missing")?;
855        ensure!(
856            pin == seed.genesis().commitment(),
857            "error seed-genesis-pin-mismatch"
858        );
859        ensure!(
860            source_matches(&mut tx, source).await? == intent.data.client,
861            "error seed-source-client-changed"
862        );
863        let (stored, state, adopted_generation): (Vec<u8>, String, Option<i64>) = sqlx::query_as("SELECT intent, state, association_generation FROM local_seed_publication_intent WHERE singleton = 1").fetch_optional(&mut *tx).await?.context("error seed-intent-missing")?;
864        ensure!(stored == intent.bytes, "error seed-intent-mismatch");
865        if state == "adopted" {
866            ensure!(
867                db::get_meta(&mut tx, "e2ee_association").await?.as_deref() == Some(&association)
868                    && adopted_generation == Some(generation(&mut tx).await?)
869                    && db::get_meta(&mut tx, "sync_cursor")
870                        .await?
871                        .context("missing cursor")?
872                        .parse::<u64>()?
873                        >= binding.prefix_count,
874                "error seed-adopted-association-changed"
875            );
876            crate::sync::encrypted_tail::dependencies::validate(
877                &mut tx,
878                &association,
879                i64::try_from(binding.prefix_count)?,
880            )
881            .await?;
882            crate::sync::encrypted_tail::attachments::client::validate(
883                &mut tx,
884                &association,
885                i64::try_from(binding.prefix_count)?,
886                &binding.descriptor_commitment,
887            )
888            .await?;
889            tx.commit().await?;
890            return Ok(false);
891        }
892        ensure!(
893            state == "sealed" && intent.data.generation == generation(&mut tx).await?,
894            "error seed-intent-not-sealed"
895        );
896        ensure!(
897            validate_history(&mut tx, &intent.data.candidate, source).await? == intent.data.history,
898            "error seed-history-commitment-mismatch"
899        );
900        // The authenticated outcome proves publication of the intent's exact
901        // descriptor. Adoption checks only its bound capture and history; local
902        // ciphertext records are not adoption inputs.
903        let proof = validated.context("error seed-intent-changed")??;
904        proof.identity().assert_matches(&mut tx).await?;
905        intent.check_proof(proof)?;
906        adopt_captured_history(&mut tx, &intent.data.candidate, binding.prefix_count).await?;
907        crate::epic_membership::recover(&mut tx, true).await?;
908        let next = intent
909            .data
910            .generation
911            .checked_add(1)
912            .context("sync generation overflow")?;
913        crate::sync::encrypted_tail::dependencies::initialize(
914            &mut tx,
915            &association,
916            next,
917            i64::try_from(binding.prefix_count)?,
918            &proof.capture().capture.snapshot.tables.task_dependencies,
919        )
920        .await?;
921        crate::sync::encrypted_tail::graphs::initialize(
922            &mut tx,
923            &proof.capture().capture.snapshot.tables.task_epic_links,
924        )
925        .await?;
926        crate::sync::encrypted_tail::attachments::client::initialize_index(
927            &mut tx,
928            &association,
929            next,
930            i64::try_from(binding.prefix_count)?,
931            &binding.descriptor_commitment,
932            proof.attachments().clone(),
933            true,
934        )
935        .await?;
936        db::set_meta(&mut tx, "sync_generation", &next.to_string()).await?;
937        db::set_meta(&mut tx, "sync_cursor", &binding.prefix_count.to_string()).await?;
938        db::set_meta(&mut tx, "e2ee_association", &association).await?;
939        sqlx::query("DELETE FROM meta WHERE key = 'sync_server_url'")
940            .execute(&mut *tx)
941            .await?;
942        sqlx::query("UPDATE local_seed_publication_intent SET state = 'adopted', association_generation = ?, association = ? WHERE singleton = 1").bind(next).bind(&association).execute(&mut *tx).await?;
943        tx.commit().await?;
944        Ok(true)
945    }
946
947    /// Separate post-commit cleanup. Durable authority/receipt never cascades away.
948    pub async fn cleanup_adopted_seed_capture(
949        &self,
950        source: &SeedSourceAuthority,
951        intent: &SeedPublicationIntent,
952    ) -> Result<()> {
953        let mut conn = self.acquire_writer().await?;
954        let mut tx = db::begin_immediate(&mut conn).await?;
955        source_matches(&mut tx, source).await?;
956        let ready: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_publication_intent WHERE intent = ? AND state = 'adopted' AND association = (SELECT value FROM meta WHERE key = 'e2ee_association') AND association_generation = CAST((SELECT value FROM meta WHERE key = 'sync_generation') AS INTEGER))").bind(&intent.bytes).fetch_one(&mut *tx).await?;
957        ensure!(ready, "error seed-adoption-cleanup-not-authorized");
958        sqlx::query("DELETE FROM local_shared_capture_journal WHERE candidate_id = ?")
959            .bind(&intent.data.candidate)
960            .execute(&mut *tx)
961            .await?;
962        tx.commit().await?;
963        Ok(())
964    }
965}
966
967#[cfg(test)]
968type IntentBarrier = (
969    String,
970    tokio::sync::oneshot::Sender<()>,
971    tokio::sync::oneshot::Receiver<()>,
972);
973#[cfg(test)]
974static INTENT_BARRIER: std::sync::Mutex<Option<IntentBarrier>> = std::sync::Mutex::new(None);
975#[cfg(test)]
976async fn wait_intent_boundary(candidate: &str) {
977    let barrier = {
978        let mut slot = INTENT_BARRIER.lock().unwrap();
979        if slot.as_ref().is_some_and(|(id, _, _)| id == candidate) {
980            slot.take()
981        } else {
982            None
983        }
984    };
985    if let Some((_, entered, resume)) = barrier {
986        let _ = entered.send(());
987        let _ = resume.await;
988    }
989}
990#[cfg(test)]
991mod tests;
aven_core/sync/shared_state/adoption.rs

aven_core/sync/shared_state/
adoption.rs
