aven_core/sync/encrypted_tail/
client.rs

1use super::*;
2use crate::db::{self, Database, begin_immediate};
3use crate::sync::{persistence, wire::ChangeWire};
4use anyhow::Context as _;
5use sqlx::{Row, SqliteConnection};
6use std::collections::HashSet;
7
8pub(super) const INITIAL_IMAGE_WATERMARK: &str = "e2ee_initial_image_watermark";
9
10// Missing catch-up state never proves that initial image demand is safe.
11pub(super) async fn initial_image_watermark(conn: &mut SqliteConnection) -> Result<String> {
12    let state = db::get_meta(conn, INITIAL_IMAGE_WATERMARK)
13        .await?
14        .context("error encrypted-image-initial-catch-up")?;
15    valid(state == "pending" || state == "ready" || state.parse::<i64>().is_ok_and(|n| n >= 0))?;
16    Ok(state)
17}
18
19pub(super) async fn validate_binding_and_cursor(
20    conn: &mut SqliteConnection,
21    authority: &Authority,
22) -> Result<i64> {
23    authority.validate()?;
24    valid(crate::sync::protocol::replica_protocol(conn).await? <= 18)?;
25    valid(
26        db::get_meta(conn, "e2ee_association").await?.as_deref()
27            == Some(authority.association.as_str()),
28    )?;
29    valid(
30        db::get_meta(conn, "sync_generation").await?.as_deref()
31            == Some(authority.sync_generation.to_string().as_str()),
32    )?;
33    let cursor = db::get_meta(conn, "sync_cursor")
34        .await?
35        .context("error encrypted-tail-cursor")?
36        .parse::<i64>()?;
37    valid(cursor >= authority.prefix)?;
38    super::dependencies::validate(conn, &authority.association, authority.prefix).await?;
39    super::attachments::client::validate(
40        conn,
41        &authority.association,
42        authority.prefix,
43        &authority.context.descriptor,
44    )
45    .await?;
46    Ok(cursor)
47}
48
49pub(super) async fn load_change(
50    conn: &mut SqliteConnection,
51    id: &str,
52) -> Result<Option<ChangeWire>> {
53    let row = sqlx::query("SELECT * FROM changes WHERE change_id = ?")
54        .bind(id)
55        .fetch_optional(conn)
56        .await?;
57    row.map(|row| {
58        Ok(ChangeWire {
59            change_id: row.try_get("change_id")?,
60            client_id: row.try_get("client_id")?,
61            local_seq: row.try_get("local_seq")?,
62            entity_type: row.try_get("entity_type")?,
63            entity_id: row.try_get("entity_id")?,
64            field: row.try_get("field")?,
65            op_type: row.try_get("op_type")?,
66            payload: domain::strict_value(row.try_get::<String, _>("payload")?.as_bytes())?,
67            base_version: row.try_get("base_version")?,
68            created_at: row.try_get("created_at")?,
69            server_seq: row.try_get("server_seq")?,
70        })
71    })
72    .transpose()
73}
74
75/// Ephemeral validation coverage for a serialized push run.
76/// The mutation sequence invalidates it after local edits; the ordering endpoint
77/// prevents reuse beyond the checked prefix, including after pending-row deletion.
78#[derive(Clone)]
79pub struct Preflight {
80    local_seq: i64,
81    through: (i64, String, String),
82}
83
84impl Preflight {
85    fn covers(&self, local_seq: i64, change: &ChangeWire) -> bool {
86        self.local_seq == local_seq
87            && (
88                change.local_seq,
89                change.created_at.as_str(),
90                change.change_id.as_str(),
91            ) <= (
92                self.through.0,
93                self.through.1.as_str(),
94                self.through.2.as_str(),
95            )
96    }
97}
98
99/// Validates one bounded ordered pending window before any of it is frozen.
100async fn preflight(
101    conn: &mut SqliteConnection,
102    local_seq: i64,
103) -> Result<Option<(ChangeWire, Preflight)>> {
104    let pending_ids: Vec<String> = sqlx::query_scalar(
105        "SELECT change_id FROM changes WHERE server_seq IS NULL
106         ORDER BY local_seq, created_at, change_id LIMIT 4096",
107    )
108    .fetch_all(&mut *conn)
109    .await?;
110    let mut preflight_bytes = 0;
111    let mut head = None;
112    let mut through = None;
113    for id in pending_ids {
114        let change = load_change(&mut *conn, &id)
115            .await?
116            .context("error encrypted-tail-history-lost")?;
117        let bytes = serde_json::to_vec(&change)?.len();
118        if head.is_some() && preflight_bytes + bytes > 16 * 1048576 {
119            break;
120        }
121        domain::validate(&change)?;
122        preflight_bytes += bytes;
123        through = Some((
124            change.local_seq,
125            change.created_at.clone(),
126            change.change_id.clone(),
127        ));
128        if head.is_none() {
129            head = Some(change);
130        }
131    }
132    Ok(head
133        .zip(through)
134        .map(|(head, through)| (head, Preflight { local_seq, through })))
135}
136
137fn without_server_sequence(change: &ChangeWire) -> ChangeWire {
138    let mut change = change.clone();
139    change.server_seq = None;
140    change
141}
142
143fn require_canonical_equality(local: &ChangeWire, incoming: &ChangeWire) -> Result<()> {
144    domain::validate(&without_server_sequence(local))?;
145    domain::validate(&without_server_sequence(incoming))?;
146    ensure!(
147        persistence::changes::canonical_equal(local, incoming),
148        "error encrypted-tail-same-id-divergence"
149    );
150    Ok(())
151}
152
153async fn validate_mapping(
154    conn: &mut SqliteConnection,
155    mapping: &Mapping,
156    cursor: i64,
157) -> Result<()> {
158    let rows: Vec<(String, i64, Vec<u8>)> = sqlx::query_as(
159        "SELECT operation_id, sequence, commitment FROM local_e2ee_accepted
160         WHERE operation_id = ? OR sequence = ?",
161    )
162    .bind(&mapping.operation_id)
163    .bind(mapping.sequence)
164    .fetch_all(&mut *conn)
165    .await?;
166    for (operation_id, sequence, commitment) in &rows {
167        valid(
168            operation_id == &mapping.operation_id
169                && *sequence == mapping.sequence
170                && commitment.as_slice() == mapping.commitment,
171        )?;
172    }
173    valid(mapping.sequence > cursor || !rows.is_empty())?;
174    if let Some(change) = load_change(conn, &mapping.operation_id).await? {
175        valid(change.server_seq.is_none_or(|s| s == mapping.sequence))?;
176    }
177    let collision: bool = sqlx::query_scalar(
178        "SELECT EXISTS(SELECT 1 FROM changes WHERE server_seq = ? AND change_id != ?)",
179    )
180    .bind(mapping.sequence)
181    .bind(&mapping.operation_id)
182    .fetch_one(conn)
183    .await?;
184    valid(!collision)
185}
186
187async fn record_acceptance_and_clear_outbox(
188    conn: &mut SqliteConnection,
189    record: &Accepted,
190) -> Result<()> {
191    sqlx::query(
192        "INSERT INTO local_e2ee_accepted(operation_id, sequence, commitment, record)
193         VALUES (?, ?, ?, ?) ON CONFLICT(operation_id) DO NOTHING",
194    )
195    .bind(&record.mapping.operation_id)
196    .bind(record.mapping.sequence)
197    .bind(record.mapping.commitment.as_slice())
198    .bind(&record.record)
199    .execute(&mut *conn)
200    .await?;
201    sqlx::query("DELETE FROM local_e2ee_outbox WHERE operation_id = ?")
202        .bind(&record.mapping.operation_id)
203        .execute(conn)
204        .await?;
205    Ok(())
206}
207
208fn open_accepted_change(authority: &Authority, accepted: &Accepted) -> Result<ChangeWire> {
209    valid(
210        accepted.mapping.sequence > authority.prefix
211            && hash(&accepted.record) == accepted.mapping.commitment,
212    )?;
213    let e = codec::parse(&accepted.record)?;
214    valid(
215        authority
216            .membership
217            .generation_allows(e.generation, u64::try_from(accepted.mapping.sequence)?),
218    )?;
219    let change = codec::open(authority, &accepted.record)?;
220    valid(change.change_id == accepted.mapping.operation_id)?;
221    Ok(change)
222}
223
224async fn load_observed_outcome(
225    conn: &mut SqliteConnection,
226    operation_id: &str,
227) -> Result<Option<(Option<i64>, Option<Vec<u8>>)>> {
228    Ok(sqlx::query_as(
229        "SELECT observed_sequence, observed_commitment FROM local_e2ee_outbox
230         WHERE operation_id = ?",
231    )
232    .bind(operation_id)
233    .fetch_optional(conn)
234    .await?)
235}
236
237async fn task_exists(conn: &mut SqliteConnection, task_id: &str) -> Result<bool> {
238    Ok(
239        sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM tasks WHERE id = ?)")
240            .bind(task_id)
241            .fetch_one(conn)
242            .await?,
243    )
244}
245
246pub(super) async fn apply_new_remote_change(
247    conn: &mut SqliteConnection,
248    change: &ChangeWire,
249    attachment_hashes: &mut HashSet<String>,
250) -> Result<()> {
251    persistence::collect_attachment_liveness_hashes(conn, change, attachment_hashes).await?;
252    if persistence::is_epic_change(change) && super::graphs::owns_graphs(conn).await? {
253        persistence::insert_wire_change(conn, change).await?;
254        let prefix: i64 = sqlx::query_scalar(
255            "SELECT prefix_count FROM local_e2ee_dependency_baseline WHERE singleton=1",
256        )
257        .fetch_one(&mut *conn)
258        .await?;
259        super::graphs::reconcile(conn, prefix).await?;
260        return Ok(());
261    }
262    if super::moves::is_metadata(change) {
263        let prefix: i64 = sqlx::query_scalar(
264            "SELECT prefix_count FROM local_e2ee_dependency_baseline WHERE singleton=1",
265        )
266        .fetch_one(&mut *conn)
267        .await?;
268        if super::moves::is_moved(conn, prefix, &change.entity_id).await? {
269            persistence::insert_wire_change(conn, change).await?;
270            return Ok(());
271        }
272    }
273    if persistence::is_epic_change(change) {
274        crate::epic_membership::capture_snapshot_baseline(
275            conn,
276            persistence::epic_change_workspace(change)?,
277            &change.entity_id,
278        )
279        .await?;
280    }
281    let related = matches!(change.op_type.as_str(), "related_add" | "related_remove");
282    // Related links reference their establishing change through a foreign key,
283    // so their history row must exist before applying the presence register.
284    if related {
285        persistence::insert_wire_change(conn, change).await?;
286    }
287    crate::sync::apply::apply_remote_change(conn, change)
288        .await
289        .map_err(|error| apply_failure(error, change))?;
290    if !related {
291        persistence::insert_wire_change(conn, change).await?;
292    }
293    persistence::reconcile_epic_change(conn, change).await?;
294    Ok(())
295}
296
297/// Transient SQLite failures retry; anything else is this record being
298/// rejected. Only the inner stable code is kept, never decrypted content.
299pub(super) fn apply_failure(error: anyhow::Error, change: &ChangeWire) -> anyhow::Error {
300    let sqlx_error = error
301        .chain()
302        .find_map(|cause| cause.downcast_ref::<sqlx::Error>());
303    // sqlx reports the extended result code; the low byte is the primary.
304    let sqlite_code = match sqlx_error {
305        Some(sqlx::Error::Database(db)) => db.code().and_then(|code| code.parse::<i32>().ok()),
306        _ => None,
307    };
308    let transient = matches!(
309        sqlx_error,
310        Some(sqlx::Error::Io(_) | sqlx::Error::PoolTimedOut)
311    ) || matches!(
312        sqlite_code.map(|code| code & 0xff),
313        // SQLITE_BUSY, SQLITE_LOCKED, SQLITE_IOERR, SQLITE_FULL
314        Some(5 | 6 | 10 | 13)
315    );
316    if transient {
317        return anyhow::anyhow!("error encrypted-tail-storage");
318    }
319    // A database error displays as "error returned from database", so sqlx
320    // causes never supply the stable code.
321    let stable = error
322        .chain()
323        .filter(|cause| cause.downcast_ref::<sqlx::Error>().is_none())
324        .find_map(|cause| {
325            cause
326                .to_string()
327                .strip_prefix("error ")
328                .and_then(|rest| rest.split_whitespace().next())
329                .map(str::to_owned)
330        });
331    let cause = stable
332        .or_else(|| sqlite_code.map(|code| format!("sqlite-{code}")))
333        .unwrap_or_else(|| "unknown".into());
334    anyhow::anyhow!(
335        "error {cause} sequence={} op_type={}",
336        change.server_seq.unwrap_or_default(),
337        change.op_type
338    )
339    .context("error encrypted-tail-apply")
340}
341
342async fn accept_local_change(
343    conn: &mut SqliteConnection,
344    authority: &Authority,
345    accepted: &Accepted,
346    change: &ChangeWire,
347) -> Result<()> {
348    persistence::update_change_server_seq(
349        &mut *conn,
350        &change.change_id,
351        Some(accepted.mapping.sequence),
352    )
353    .await?;
354    persistence::reconcile_epic_change(&mut *conn, change).await?;
355    super::notes::reconcile(&mut *conn, authority.prefix, change).await?;
356    super::labels::reconcile(&mut *conn, authority.prefix, change).await?;
357    if let Some(workspace) = super::dependencies::affected_workspace(change)? {
358        super::dependencies::reconcile(&mut *conn, authority.prefix, workspace).await?;
359    }
360    super::attachments::client::accept(&mut *conn, accepted, change).await?;
361    record_acceptance_and_clear_outbox(&mut *conn, accepted).await?;
362    Ok(())
363}
364
365impl Database {
366    /// Validates attachment coverage before a drain can push or apply records.
367    /// Transaction-local checks continue to validate the active association.
368    pub(crate) async fn validate_encrypted_attachment_integrity(
369        &self,
370        authority: &Authority,
371    ) -> Result<()> {
372        let mut conn = self.acquire_reader().await?;
373        let mut tx = sqlx::Connection::begin(&mut *conn).await?;
374        validate_binding_and_cursor(&mut tx, authority).await?;
375        super::attachments::client::validate_integrity(&mut tx).await?;
376        tx.commit().await?;
377        Ok(())
378    }
379
380    /// One validated read of local round state. Observation never freezes work
381    /// or advances the download selector.
382    pub async fn encrypted_round_state(&self, authority: &Authority) -> Result<RoundState> {
383        let mut conn = self.acquire_reader().await?;
384        let mut tx = sqlx::Connection::begin(&mut *conn).await?;
385        let cursor = validate_binding_and_cursor(&mut tx, authority).await?;
386        let initial = initial_image_watermark(&mut tx).await?;
387        let (idle, upload_pending): (bool, bool) = sqlx::query_as(
388            "SELECT NOT EXISTS(SELECT 1 FROM local_e2ee_outbox)
389                    AND NOT EXISTS(SELECT 1 FROM changes WHERE server_seq IS NULL),
390                    EXISTS(SELECT 1 FROM changes
391                           WHERE server_seq IS NULL AND op_type = 'attachment_add')",
392        )
393        .fetch_one(&mut *tx)
394        .await?;
395        let downloads = if initial == "ready" {
396            Some(super::attachments::client::downloads(&mut tx).await?)
397        } else {
398            None
399        };
400        tx.commit().await?;
401        Ok(RoundState {
402            cursor,
403            initial_watermark: initial.parse().ok(),
404            idle,
405            upload_pending,
406            downloads,
407        })
408    }
409    /// Highest tail rank this replica applied or accepted. An honest allocator
410    /// mark is never below a rank it assigned.
411    pub async fn encrypted_tail_observed_rank(&self) -> Result<u64> {
412        let mut conn = self.acquire_reader().await?;
413        let rank: Option<i64> = sqlx::query_scalar(
414            "SELECT max(rank) FROM (
415                 SELECT CAST(value AS INTEGER) AS rank FROM meta WHERE key = 'sync_cursor'
416                 UNION ALL SELECT sequence FROM local_e2ee_accepted
417             )",
418        )
419        .fetch_one(&mut *conn)
420        .await?;
421        Ok(u64::try_from(rank.unwrap_or(0))?)
422    }
423    /// Returns the frozen head, or freezes the next ordered pending change.
424    /// Image heads also return exact staged ciphertext for upload.
425    pub async fn prepare_encrypted_push(
426        &self,
427        authority: &Authority,
428        blob_dir: &std::path::Path,
429    ) -> Result<Option<Push>> {
430        Ok(self
431            .prepare_encrypted_push_inner(authority, blob_dir, None, 1)
432            .await?
433            .0)
434    }
435    /// Freezes the next ordered pending group in one serialized push run.
436    /// Reuses only the checked window while the local mutation sequence is unchanged.
437    pub async fn prepare_encrypted_batch_in_run(
438        &self,
439        authority: &Authority,
440        blob_dir: &std::path::Path,
441        preflight: Option<Preflight>,
442        count: usize,
443    ) -> Result<(Option<Push>, Option<Preflight>)> {
444        self.prepare_encrypted_push_inner(authority, blob_dir, preflight, count)
445            .await
446    }
447    async fn prepare_encrypted_push_inner(
448        &self,
449        authority: &Authority,
450        blob_dir: &std::path::Path,
451        preflight: Option<Preflight>,
452        count: usize,
453    ) -> Result<(Option<Push>, Option<Preflight>)> {
454        valid((1..=BATCH_COUNT).contains(&count))?;
455        let mut conn = self.acquire_writer().await?;
456        let mut tx = begin_immediate(&mut conn).await?;
457        validate_binding_and_cursor(&mut tx, authority).await?;
458        if authority.rotation_pending() {
459            tx.commit().await?;
460            return Ok((None, preflight));
461        }
462        let frozen: Option<(String, i64, Vec<u8>, bool)> = sqlx::query_as(
463            "SELECT association, sync_generation, record, blocked
464             FROM local_e2ee_outbox ORDER BY position LIMIT 1",
465        )
466        .fetch_optional(&mut *tx)
467        .await?;
468        if let Some((association, generation, record, blocked)) = frozen {
469            valid(association == authority.association && generation == authority.sync_generation)?;
470            ensure!(!blocked, "error encrypted-tail-integrity-blocked");
471            ensure!(
472                !authority.record_is_closed(&record)?,
473                "error encrypted-tail-outcome-required"
474            );
475            let change = codec::open(authority, &record)?;
476            require_canonical_equality(
477                &change,
478                &load_change(&mut tx, &change.change_id)
479                    .await?
480                    .context("error encrypted-tail-history-lost")?,
481            )?;
482            let upload = if change.op_type == "attachment_add" {
483                Some(
484                    super::attachments::client::frozen_upload(
485                        &mut tx, authority, &change, &record, blob_dir,
486                    )
487                    .await?,
488                )
489            } else {
490                None
491            };
492            tx.commit().await?;
493            return Ok((Some(Push { record, upload }), preflight));
494        }
495        let local_seq = db::get_meta(&mut tx, "local_seq")
496            .await?
497            .context("error encrypted-tail-local-sequence")?
498            .parse::<i64>()?;
499        let id: Option<String> = sqlx::query_scalar(
500            "SELECT change_id FROM changes WHERE server_seq IS NULL
501             ORDER BY local_seq, created_at, change_id LIMIT 1",
502        )
503        .fetch_optional(&mut *tx)
504        .await?;
505        let Some(id) = id else {
506            tx.commit().await?;
507            return Ok((None, preflight));
508        };
509        let mut change = load_change(&mut tx, &id)
510            .await?
511            .context("error encrypted-tail-history-lost")?;
512        let preflight = match preflight {
513            Some(coverage) if coverage.covers(local_seq, &change) => coverage,
514            _ => {
515                let (head, coverage) = self::preflight(&mut tx, local_seq)
516                    .await?
517                    .context("error encrypted-tail-history-lost")?;
518                change = head;
519                coverage
520            }
521        };
522        let (projection, upload) = if change.op_type == "attachment_add" {
523            let (projection, upload) =
524                super::attachments::client::stage(&mut tx, authority, &change, blob_dir).await?;
525            (projection, Some(upload))
526        } else {
527            (domain::validate(&change)?, None)
528        };
529        let record = codec::seal_projection(authority, &change, &projection)?;
530        // Check the full decoder contract before committing dispatch authority.
531        require_canonical_equality(&change, &codec::open(authority, &record)?)?;
532        sqlx::query(
533            "INSERT INTO local_e2ee_outbox(
534                 position, operation_id, association, sync_generation, record
535             ) VALUES (1, ?, ?, ?, ?)",
536        )
537        .bind(&change.change_id)
538        .bind(&authority.association)
539        .bind(authority.sync_generation)
540        .bind(&record)
541        .execute(&mut *tx)
542        .await?;
543        let mut frozen_count = 1;
544        if upload.is_none() && count > 1 {
545            let ids: Vec<String> = sqlx::query_scalar(
546                "SELECT change_id FROM changes WHERE server_seq IS NULL AND change_id != ?
547                 ORDER BY local_seq, created_at, change_id LIMIT ?",
548            )
549            .bind(&change.change_id)
550            .bind((count - 1) as i64)
551            .fetch_all(&mut *tx)
552            .await?;
553            let mut bytes = record.len();
554            for (index, id) in ids.iter().enumerate() {
555                let next = load_change(&mut tx, id)
556                    .await?
557                    .context("error encrypted-tail-history-lost")?;
558                if !preflight.covers(local_seq, &next) || next.op_type == "attachment_add" {
559                    break;
560                }
561                let projection = domain::validate(&next)?;
562                let next_record = codec::seal_projection(authority, &next, &projection)?;
563                if bytes + next_record.len() > BATCH_BYTES {
564                    break;
565                }
566                require_canonical_equality(&next, &codec::open(authority, &next_record)?)?;
567                bytes += next_record.len();
568                sqlx::query("INSERT INTO local_e2ee_outbox(position, operation_id, association, sync_generation, record) VALUES (?, ?, ?, ?, ?)")
569                    .bind((index + 2) as i64).bind(id).bind(&authority.association)
570                    .bind(authority.sync_generation).bind(next_record).execute(&mut *tx).await?;
571                frozen_count += 1;
572            }
573        }
574        tx.commit().await?;
575        if frozen_count > 1 {
576            crate::sync::crash::Crash::Tail.at("batch-frozen");
577        }
578        crate::sync::crash::Crash::Tail.at(if upload.is_some() {
579            "image-frozen"
580        } else {
581            "frozen"
582        });
583        Ok((Some(Push { record, upload }), Some(preflight)))
584    }
585    /// Closed-generation absence and replacement commit under one outbox/history owner.
586    /// A failed transaction retains old bytes and requires a new lookup on retry.
587    pub async fn reconcile_encrypted_tail_absence(
588        &self,
589        authority: &Authority,
590        absence: &AbsentOperation,
591        blob_dir: &std::path::Path,
592    ) -> Result<bool> {
593        self.reconcile_encrypted_tail_absences(authority, std::slice::from_ref(absence), blob_dir)
594            .await
595    }
596    pub async fn reconcile_encrypted_tail_absences(
597        &self,
598        authority: &Authority,
599        absences: &[AbsentOperation],
600        blob_dir: &std::path::Path,
601    ) -> Result<bool> {
602        valid(absences.len() <= BATCH_COUNT)?;
603        let mut conn = self.acquire_writer().await?;
604        let mut tx = begin_immediate(&mut conn).await?;
605        validate_binding_and_cursor(&mut tx, authority).await?;
606        let mut changed = false;
607        for absence in absences {
608            valid(absence.context == authority.context)?;
609            let (record, association, generation, observed, blocked): (Vec<u8>, String, i64, bool, bool) = sqlx::query_as("SELECT record,association,sync_generation,observed_sequence IS NOT NULL OR observed_commitment IS NOT NULL,blocked FROM local_e2ee_outbox WHERE record=?").bind(&absence.record).fetch_one(&mut *tx).await?;
610            valid(
611                record == absence.record
612                    && association == authority.association
613                    && generation == authority.sync_generation
614                    && !observed
615                    && !blocked,
616            )?;
617            let change = codec::open(authority, &record)?;
618            let local = load_change(&mut tx, &change.change_id)
619                .await?
620                .context("error encrypted-tail-history-lost")?;
621            require_canonical_equality(&local, &change)?;
622            let accepted: bool = sqlx::query_scalar(
623                "SELECT EXISTS(SELECT 1 FROM local_e2ee_accepted WHERE operation_id=?)",
624            )
625            .bind(&change.change_id)
626            .fetch_one(&mut *tx)
627            .await?;
628            valid(local.server_seq.is_none() && !accepted)?;
629            if authority.rotation_pending() {
630                tx.commit().await?;
631                return Ok(false);
632            }
633            if !authority.record_is_closed(&record)? {
634                continue;
635            }
636            let mut projection = codec::parse(&record)?.projection;
637            if change.op_type == "attachment_add" {
638                projection = super::attachments::client::supersede(
639                    &mut tx, authority, &change, projection, blob_dir,
640                )
641                .await?;
642            }
643            let replacement = codec::seal_projection(authority, &change, &projection)?;
644            require_canonical_equality(&change, &codec::open(authority, &replacement)?)?;
645            let n = sqlx::query("UPDATE local_e2ee_outbox SET record=? WHERE record=? AND observed_sequence IS NULL AND observed_commitment IS NULL AND blocked=0")
646            .bind(&replacement).bind(&record).execute(&mut *tx).await?.rows_affected();
647            valid(n == 1)?;
648            changed = true;
649        }
650        if changed {
651            crate::sync::crash::Crash::Tail.at("before-supersede-commit");
652        }
653        tx.commit().await?;
654        if changed {
655            crate::sync::crash::Crash::Tail.at("after-supersede-commit");
656        }
657        Ok(true)
658    }
659    /// Retain an observed immutable outcome before fetching a different representation.
660    pub async fn observe_encrypted_tail(
661        &self,
662        authority: &Authority,
663        mapping: &Mapping,
664    ) -> Result<Vec<u8>> {
665        let mut conn = self.acquire_writer().await?;
666        let mut tx = begin_immediate(&mut conn).await?;
667        let cursor = validate_binding_and_cursor(&mut tx, authority).await?;
668        validate_mapping(&mut tx, mapping, cursor).await?;
669        valid(mapping.sequence > authority.prefix)?;
670        let (operation_id, record, observed_sequence, observed_commitment): (
671            String,
672            Vec<u8>,
673            Option<i64>,
674            Option<Vec<u8>>,
675        ) = sqlx::query_as(
676            "SELECT operation_id, record, observed_sequence, observed_commitment
677             FROM local_e2ee_outbox WHERE operation_id = ?",
678        )
679        .bind(&mapping.operation_id)
680        .fetch_one(&mut *tx)
681        .await?;
682        valid(
683            operation_id == mapping.operation_id
684                && observed_sequence.is_none_or(|sequence| sequence == mapping.sequence)
685                && observed_commitment
686                    .as_ref()
687                    .is_none_or(|h| h.as_slice() == mapping.commitment),
688        )?;
689        if hash(&record) == mapping.commitment {
690            let envelope = codec::parse(&record)?;
691            valid(
692                authority
693                    .membership
694                    .generation_allows(envelope.generation, u64::try_from(mapping.sequence)?),
695            )?;
696        }
697        sqlx::query(
698            "UPDATE local_e2ee_outbox SET observed_sequence = ?, observed_commitment = ?
699             WHERE operation_id = ?",
700        )
701        .bind(mapping.sequence)
702        .bind(mapping.commitment.as_slice())
703        .bind(&mapping.operation_id)
704        .execute(&mut *tx)
705        .await?;
706        tx.commit().await?;
707        Ok(record)
708    }
709    pub async fn verify_encrypted_tail_outcome(
710        &self,
711        authority: &Authority,
712        accepted: &Accepted,
713    ) -> Result<()> {
714        let result = self
715            .commit_encrypted_tail_outcome(authority, accepted)
716            .await;
717        if result.as_ref().err().is_some_and(|error| {
718            crate::sync::client::errors::has_code(error, "encrypted-tail-same-id-divergence")
719        }) {
720            let mut conn = self.acquire_writer().await?;
721            sqlx::query("UPDATE local_e2ee_outbox SET blocked = 1 WHERE operation_id = ?")
722                .bind(&accepted.mapping.operation_id)
723                .execute(&mut *conn)
724                .await?;
725        }
726        result
727    }
728    async fn commit_encrypted_tail_outcome(
729        &self,
730        authority: &Authority,
731        accepted: &Accepted,
732    ) -> Result<()> {
733        let change = open_accepted_change(authority, accepted)?;
734        let mut conn = self.acquire_writer().await?;
735        let mut tx = begin_immediate(&mut conn).await?;
736        let cursor = validate_binding_and_cursor(&mut tx, authority).await?;
737        validate_mapping(&mut tx, &accepted.mapping, cursor).await?;
738        let observed = load_observed_outcome(&mut tx, &change.change_id).await?;
739        if let Some((sequence, commitment)) = observed {
740            valid(
741                sequence == Some(accepted.mapping.sequence)
742                    && commitment.as_deref() == Some(accepted.mapping.commitment.as_slice()),
743            )?;
744        }
745        let local = load_change(&mut tx, &change.change_id)
746            .await?
747            .context("error encrypted-tail-local-missing")?;
748        require_canonical_equality(&local, &change)?;
749        accept_local_change(&mut tx, authority, accepted, &change).await?;
750        if super::moves::needs_reconcile(&change) {
751            super::moves::reconcile(&mut tx, authority.prefix).await?;
752        }
753        if super::graphs::needs_reconcile(&change) {
754            super::graphs::reconcile(&mut tx, authority.prefix).await?;
755        }
756        tx.commit().await?;
757        Ok(())
758    }
759    pub async fn encrypted_tail_has_batch_work(&self, authority: &Authority) -> Result<bool> {
760        let mut conn = self.acquire_reader().await?;
761        validate_binding_and_cursor(&mut conn, authority).await?;
762        let count: i64 = sqlx::query_scalar("SELECT count(*) FROM local_e2ee_outbox")
763            .fetch_one(&mut *conn)
764            .await?;
765        if count > 1 {
766            return Ok(true);
767        }
768        if count == 1 {
769            return Ok(false);
770        }
771        let ops: Vec<String> = sqlx::query_scalar("SELECT op_type FROM changes WHERE server_seq IS NULL ORDER BY local_seq,created_at,change_id LIMIT 2").fetch_all(&mut *conn).await?;
772        Ok(ops.len() == 2 && ops.iter().all(|op| op != "attachment_add"))
773    }
774    /// Returns the immutable unresolved inventory in dispatch order.
775    pub async fn encrypted_tail_frozen_records(
776        &self,
777        authority: &Authority,
778    ) -> Result<Vec<(String, Vec<u8>)>> {
779        let mut conn = self.acquire_reader().await?;
780        let mut tx = sqlx::Connection::begin(&mut *conn).await?;
781        validate_binding_and_cursor(&mut tx, authority).await?;
782        let rows: Vec<(String, Vec<u8>, String, i64, bool)> = sqlx::query_as(
783            "SELECT operation_id,record,association,sync_generation,blocked FROM local_e2ee_outbox ORDER BY position LIMIT 129"
784        ).fetch_all(&mut *tx).await?;
785        valid(rows.len() <= BATCH_COUNT)?;
786        let mut result = Vec::with_capacity(rows.len());
787        for (id, record, association, generation, blocked) in rows {
788            valid(association == authority.association && generation == authority.sync_generation)?;
789            ensure!(!blocked, "error encrypted-tail-integrity-blocked");
790            let change = codec::open(authority, &record)?;
791            valid(change.change_id == id)?;
792            result.push((id, record));
793        }
794        tx.commit().await?;
795        Ok(result)
796    }
797
798    /// Accepts exact frozen ciphertext together with every local domain effect.
799    pub async fn accept_encrypted_tail_batch(
800        &self,
801        authority: &Authority,
802        mappings: &[Mapping],
803    ) -> Result<()> {
804        valid(!mappings.is_empty() && mappings.len() <= BATCH_COUNT)?;
805        let mut conn = self.acquire_writer().await?;
806        let mut tx = begin_immediate(&mut conn).await?;
807        let cursor = validate_binding_and_cursor(&mut tx, authority).await?;
808        let mut accepted_records = Vec::new();
809        let mut ids = HashSet::new();
810        let mut sequences = HashSet::new();
811        for mapping in mappings {
812            valid(ids.insert(&mapping.operation_id) && sequences.insert(mapping.sequence))?;
813            validate_mapping(&mut tx, mapping, cursor).await?;
814            let (record, sequence, commitment): (Vec<u8>, Option<i64>, Option<Vec<u8>>) = sqlx::query_as(
815                "SELECT record,observed_sequence,observed_commitment FROM local_e2ee_outbox WHERE operation_id=?"
816            ).bind(&mapping.operation_id).fetch_one(&mut *tx).await?;
817            valid(
818                sequence.is_none_or(|n| n == mapping.sequence)
819                    && commitment
820                        .as_deref()
821                        .is_none_or(|h| h == mapping.commitment),
822            )?;
823            let accepted = Accepted {
824                mapping: mapping.clone(),
825                record,
826            };
827            let change = open_accepted_change(authority, &accepted)?;
828            let local = load_change(&mut tx, &change.change_id)
829                .await?
830                .context("error encrypted-tail-local-missing")?;
831            require_canonical_equality(&local, &change)?;
832            accepted_records.push((accepted, change));
833        }
834        for (accepted, change) in &accepted_records {
835            accept_local_change(&mut tx, authority, accepted, change).await?;
836        }
837        if accepted_records
838            .iter()
839            .any(|(_, change)| super::moves::needs_reconcile(change))
840        {
841            super::moves::reconcile(&mut tx, authority.prefix).await?;
842        }
843        if accepted_records
844            .iter()
845            .any(|(_, change)| super::graphs::needs_reconcile(change))
846        {
847            super::graphs::reconcile(&mut tx, authority.prefix).await?;
848        }
849        crate::sync::crash::Crash::Tail.at("before-batch-commit");
850        tx.commit().await?;
851        crate::sync::crash::Crash::Tail.at("after-batch-commit");
852        Ok(())
853    }
854
855    pub async fn apply_encrypted_tail_page(
856        &self,
857        authority: &Authority,
858        page: &Page,
859    ) -> Result<()> {
860        valid(
861            page.records.len() <= PAGE_COUNT
862                && page.after >= authority.prefix
863                && page.watermark >= page.after
864                && page
865                    .records
866                    .iter()
867                    .map(|accepted| accepted.record.len())
868                    .sum::<usize>()
869                    <= PAGE_BYTES,
870        )?;
871        let mut operation_ids = HashSet::new();
872        let mut previous_sequence = page.after;
873        let mut changes = Vec::new();
874        for accepted in &page.records {
875            valid(
876                accepted.mapping.sequence > previous_sequence
877                    && accepted.mapping.sequence <= page.watermark
878                    && operation_ids.insert(&accepted.mapping.operation_id),
879            )?;
880            changes.push(open_accepted_change(authority, accepted)?);
881            previous_sequence = accepted.mapping.sequence;
882        }
883        valid(
884            page.cursor == previous_sequence && (!page.has_more || previous_sequence > page.after),
885        )?;
886        let mut conn = self.acquire_writer().await?;
887        let mut tx = begin_immediate(&mut conn).await?;
888        valid(validate_binding_and_cursor(&mut tx, authority).await? == page.after)?;
889        let initial = initial_image_watermark(&mut tx).await?;
890        let target = if initial == "pending" {
891            Some(page.watermark)
892        } else {
893            initial.parse::<i64>().ok()
894        };
895        if let Some(target) = target {
896            valid(page.watermark == target && page.after <= target)?;
897        }
898        let mut attachment_hashes = HashSet::new();
899        let mut dependency_workspaces = HashSet::new();
900        // Validate every mapping and local comparison before any domain effects.
901        let placement_changed = changes.iter().any(super::moves::needs_reconcile);
902        let graphs_changed = changes.iter().any(super::graphs::needs_reconcile);
903        let mut local_presence = Vec::with_capacity(changes.len());
904        for (accepted, change) in page.records.iter().zip(&changes) {
905            validate_mapping(&mut tx, &accepted.mapping, page.after).await?;
906            let local = load_change(&mut tx, &change.change_id).await?;
907            if let Some(local) = &local {
908                require_canonical_equality(local, change)?;
909            }
910            let observed = load_observed_outcome(&mut tx, &change.change_id).await?;
911            if let Some((Some(sequence), Some(commitment))) = observed {
912                valid(
913                    sequence == accepted.mapping.sequence
914                        && commitment.as_slice() == accepted.mapping.commitment,
915                )?;
916            }
917            local_presence.push(local.is_some());
918        }
919        // All local ranks precede incoming relation application. Page IDs are unique,
920        // and ranking does not insert history, so validated presence remains stable.
921        for ((accepted, change), is_local) in page.records.iter().zip(&changes).zip(&local_presence)
922        {
923            if *is_local {
924                persistence::update_change_server_seq(
925                    &mut tx,
926                    &change.change_id,
927                    Some(accepted.mapping.sequence),
928                )
929                .await?;
930                persistence::reconcile_epic_change(&mut tx, change).await?;
931            }
932        }
933        let mut remote_seq = 0;
934        for ((accepted, mut change), is_local) in
935            page.records.iter().zip(changes).zip(local_presence)
936        {
937            if !is_local {
938                remote_seq = remote_seq.max(change.local_seq);
939                // Applying lifecycle resolution can materialize deterministic history
940                // needed by a later record in this page. It is not an identity-only echo.
941                if let Some(generated) = load_change(&mut tx, &change.change_id).await? {
942                    valid(super::recurrence::is_deterministic(&change))?;
943                    require_canonical_equality(&generated, &change)?;
944                    valid(generated.server_seq.is_none())?;
945                    persistence::update_change_server_seq(
946                        &mut tx,
947                        &change.change_id,
948                        Some(accepted.mapping.sequence),
949                    )
950                    .await?;
951                } else {
952                    change.server_seq = Some(accepted.mapping.sequence);
953                    let generated = change.op_type == "create_task"
954                        && super::recurrence::is_deterministic(&change)
955                        && task_exists(&mut tx, &change.entity_id).await?;
956                    apply_new_remote_change(&mut tx, &change, &mut attachment_hashes).await?;
957                    if generated {
958                        super::recurrence::adopt_earlier_generation(
959                            &mut tx,
960                            authority.prefix,
961                            &change,
962                        )
963                        .await?;
964                    }
965                }
966            }
967            if let Some(workspace) = super::dependencies::affected_workspace(&change)? {
968                dependency_workspaces.insert(workspace.to_owned());
969            }
970            super::notes::reconcile(&mut tx, authority.prefix, &change).await?;
971            super::labels::reconcile(&mut tx, authority.prefix, &change).await?;
972            super::attachments::client::accept(&mut tx, accepted, &change).await?;
973            record_acceptance_and_clear_outbox(&mut tx, accepted).await?;
974        }
975        for workspace in dependency_workspaces {
976            super::dependencies::reconcile(&mut tx, authority.prefix, &workspace).await?;
977        }
978        if placement_changed {
979            super::moves::reconcile(&mut tx, authority.prefix).await?;
980        }
981        if graphs_changed {
982            super::graphs::reconcile(&mut tx, authority.prefix).await?;
983        }
984        crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
985            &mut tx,
986            &attachment_hashes.into_iter().collect::<Vec<_>>(),
987            &crate::attachments::lifecycle::SystemClock,
988        )
989        .await?;
990        // Like installation, keep the local sequence at or above every stored
991        // change, so later local writes sort after applied remote history.
992        let local_seq = db::get_meta(&mut tx, "local_seq")
993            .await?
994            .unwrap_or_else(|| "0".to_string())
995            .parse::<i64>()?;
996        if remote_seq > local_seq {
997            db::set_meta(&mut tx, "local_seq", &remote_seq.to_string()).await?;
998        }
999        crate::sync::protocol::establish_protocol(&mut tx, 18).await?;
1000        db::set_meta(&mut tx, "sync_cursor", &page.cursor.to_string()).await?;
1001        if let Some(target) = target {
1002            let state = if page.cursor >= target {
1003                "ready".into()
1004            } else {
1005                target.to_string()
1006            };
1007            db::set_meta(&mut tx, INITIAL_IMAGE_WATERMARK, &state).await?;
1008        }
1009        crate::sync::crash::Crash::Tail.at("before-page-commit");
1010        tx.commit().await?;
1011        crate::sync::crash::Crash::Tail.at("after-page-commit");
1012        Ok(())
1013    }
1014}
aven_core/sync/encrypted_tail/client.rs

aven_core/sync/encrypted_tail/
client.rs
