aven_core/sync/
persistence.rs

1use std::collections::HashSet;
2use std::path::Path;
3use std::time::Instant;
4
5use anyhow::{Context, Result, bail};
6use sqlx::{QueryBuilder, Sqlite, SqliteConnection};
7
8use super::apply::apply_remote_change;
9use super::wire::{
10    AttachmentAddPayload, ChangeRow, ChangeWire, PushAck, SyncRequest, SyncResponse,
11};
12use crate::change_log::op_type;
13use crate::db::{Database, begin_immediate, get_meta, set_meta};
14
15#[derive(Debug, Clone, Default, PartialEq, Eq)]
16pub struct SyncPersistenceStatus {
17    pub pinned_server: Option<String>,
18    pub pending_changes: i64,
19    pub conflicts: i64,
20    pub sync_cursor: Option<String>,
21    pub local_sequence: Option<String>,
22    pub last_attempt: Option<String>,
23    pub last_success: Option<String>,
24    pub last_error: Option<String>,
25    pub last_pushed: Option<String>,
26    pub last_pulled: Option<String>,
27    pub last_cursor: Option<String>,
28}
29
30#[derive(Debug, Clone)]
31pub struct ClientSyncPage {
32    pub request: SyncRequest,
33    pub pending: usize,
34}
35
36#[derive(Debug)]
37pub struct ApplySyncPage {
38    pub request: SyncRequest,
39    pub response: SyncResponse,
40    pub attempted_at: String,
41    pub previous_pushed: i64,
42    pub previous_pulled: usize,
43}
44
45#[derive(Debug)]
46pub struct ServerSyncPage {
47    pub request: SyncRequest,
48}
49
50#[derive(Debug)]
51pub struct ServerSyncResult {
52    pub accepted_count: i64,
53    pub push_acks: Vec<PushAck>,
54    pub changes: Vec<ChangeWire>,
55    pub has_more: bool,
56    pub blob_prepare_ms: u128,
57    pub assign_ms: u128,
58    pub pull_query_ms: u128,
59}
60
61impl Database {
62    pub async fn sync_persistence_status(&self) -> Result<SyncPersistenceStatus> {
63        let mut conn = self.acquire_writer().await?;
64        let pending_changes =
65            sqlx::query_scalar("SELECT count(*) FROM changes WHERE server_seq IS NULL")
66                .fetch_one(&mut *conn)
67                .await?;
68        let conflicts = sqlx::query_scalar("SELECT count(*) FROM conflicts WHERE resolved = 0")
69            .fetch_one(&mut *conn)
70            .await?;
71        Ok(SyncPersistenceStatus {
72            pinned_server: get_meta(&mut conn, "sync_server_url").await?,
73            pending_changes,
74            conflicts,
75            sync_cursor: get_meta(&mut conn, "sync_cursor").await?,
76            local_sequence: get_meta(&mut conn, "local_seq").await?,
77            last_attempt: get_meta(&mut conn, "sync_last_attempt_at").await?,
78            last_success: get_meta(&mut conn, "sync_last_success_at").await?,
79            last_error: get_meta(&mut conn, "sync_last_error").await?,
80            last_pushed: get_meta(&mut conn, "sync_last_pushed").await?,
81            last_pulled: get_meta(&mut conn, "sync_last_pulled").await?,
82            last_cursor: get_meta(&mut conn, "sync_last_cursor").await?,
83        })
84    }
85
86    pub async fn begin_sync_attempt(&self, attempted_at: String) -> Result<()> {
87        let mut conn = self.acquire_writer().await?;
88        set_meta(&mut conn, "sync_last_attempt_at", &attempted_at).await
89    }
90
91    pub async fn record_sync_error(&self, error: String) -> Result<()> {
92        let mut conn = self.acquire_writer().await?;
93        set_meta(&mut conn, "sync_last_error", &error).await
94    }
95
96    pub(super) async fn pending_sync_change_count(&self) -> Result<i64> {
97        let mut conn = self.acquire_reader().await?;
98        Ok(
99            sqlx::query_scalar("SELECT COUNT(*) FROM changes WHERE server_seq IS NULL")
100                .fetch_one(&mut *conn)
101                .await?,
102        )
103    }
104
105    pub(super) async fn pending_blob_counts(
106        &self,
107        known_server_blobs: &HashSet<String>,
108    ) -> Result<crate::attachments::lifecycle::ByteCount> {
109        let mut conn = self.acquire_reader().await?;
110        let known = serde_json::to_string(known_server_blobs)?;
111        let (count, bytes): (i64, i64) = sqlx::query_as(
112            "WITH known(sha256) AS (
113               SELECT value FROM json_each(?)
114             ), pending AS (
115               SELECT json_extract(payload, '$.workspace_id') AS workspace_id,
116                      json_extract(payload, '$.sha256') AS sha256,
117                      MAX(CAST(json_extract(payload, '$.byte_size') AS INTEGER)) AS byte_size
118               FROM changes
119               WHERE server_seq IS NULL AND op_type = 'attachment_add'
120               GROUP BY workspace_id, sha256
121             )
122             SELECT COUNT(*), COALESCE(SUM(byte_size), 0)
123             FROM pending LEFT JOIN known USING (sha256)
124             WHERE known.sha256 IS NULL",
125        )
126        .bind(known)
127        .fetch_one(&mut *conn)
128        .await?;
129        Ok(crate::attachments::lifecycle::ByteCount {
130            count: u64::try_from(count)?,
131            bytes: u64::try_from(bytes)?,
132        })
133    }
134
135    pub async fn prepare_client_sync_page(
136        &self,
137        server: String,
138        push_limit: usize,
139        pull_limit: u32,
140    ) -> Result<ClientSyncPage> {
141        let mut conn = self.acquire_writer().await?;
142        validate_sync_server(&mut conn, &server).await?;
143        let client_id = get_meta(&mut conn, "client_id")
144            .await?
145            .context("missing client id")?;
146        let after = sync_cursor(&mut conn).await?;
147        let changes = load_unsynced_changes(&mut conn, push_limit).await?;
148        let pending = changes.len();
149        Ok(ClientSyncPage {
150            request: SyncRequest {
151                protocol_version: Some(super::wire::SYNC_PROTOCOL_VERSION),
152                client_id,
153                after,
154                pull_limit: Some(pull_limit),
155                changes,
156            },
157            pending,
158        })
159    }
160
161    pub async fn apply_client_sync_page(&self, page: ApplySyncPage) -> Result<usize> {
162        let envelope = super::wire::validate_sync_request_envelope(&page.request)?;
163        let request_change_ids = page
164            .request
165            .changes
166            .iter()
167            .map(|change| change.change_id.clone())
168            .collect::<Vec<_>>();
169        super::wire::validate_sync_response_for_request(
170            envelope.after,
171            envelope.pull_limit,
172            &request_change_ids,
173            &page.response,
174        )?;
175        let mut conn = self.acquire_writer().await?;
176        apply_sync_response(&mut conn, page).await
177    }
178
179    pub async fn persist_server_sync_page(&self, page: ServerSyncPage) -> Result<ServerSyncResult> {
180        self.persist_server_sync_page_inner(page, None).await
181    }
182
183    pub async fn persist_server_sync_page_with_blobs(
184        &self,
185        page: ServerSyncPage,
186        blob_dir: &Path,
187    ) -> Result<ServerSyncResult> {
188        self.persist_server_sync_page_inner(page, Some(blob_dir))
189            .await
190    }
191
192    async fn persist_server_sync_page_inner(
193        &self,
194        page: ServerSyncPage,
195        blob_dir: Option<&Path>,
196    ) -> Result<ServerSyncResult> {
197        let envelope = super::wire::validate_sync_request_envelope(&page.request)?;
198        for change in &page.request.changes {
199            super::wire::validate_pushed_change(change)?;
200        }
201        if blob_dir.is_none()
202            && page
203                .request
204                .changes
205                .iter()
206                .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
207        {
208            bail!("error attachment-blob-storage-required");
209        }
210        let blob_prepare_started = Instant::now();
211        if let Some(blob_dir) = blob_dir {
212            let mut conn = self.acquire_reader().await?;
213            prepare_server_blobs(&mut conn, blob_dir, &page.request.changes).await?;
214        }
215        let blob_prepare_ms = blob_prepare_started.elapsed().as_millis();
216        let mut conn = self.acquire_writer().await?;
217        let assign_started = Instant::now();
218        let (accepted_count, push_acks) =
219            assign_server_sequences(&mut conn, page.request.changes, blob_dir).await?;
220        let assign_ms = assign_started.elapsed().as_millis();
221        let pull_query_started = Instant::now();
222        let (changes, has_more) =
223            load_server_changes_after(&mut conn, envelope.after, envelope.pull_limit).await?;
224        let pull_query_ms = pull_query_started.elapsed().as_millis();
225        Ok(ServerSyncResult {
226            accepted_count,
227            push_acks,
228            changes,
229            has_more,
230            blob_prepare_ms,
231            assign_ms,
232            pull_query_ms,
233        })
234    }
235}
236
237async fn sync_cursor(conn: &mut SqliteConnection) -> Result<i64> {
238    Ok(get_meta(conn, "sync_cursor")
239        .await?
240        .unwrap_or_else(|| "0".to_string())
241        .parse::<i64>()?)
242}
243
244async fn validate_sync_server(conn: &mut SqliteConnection, server: &str) -> Result<()> {
245    let normalized = server.trim_end_matches('/');
246    if let Some(existing) = get_meta(conn, "sync_server_url").await? {
247        if existing != normalized {
248            bail!(
249                "error sync-server-changed existing={} requested={} hint=\"use a fresh database for a different sync server\"",
250                existing,
251                normalized
252            );
253        }
254    } else {
255        set_meta(conn, "sync_server_url", normalized).await?;
256    }
257    Ok(())
258}
259
260async fn load_unsynced_changes(
261    conn: &mut SqliteConnection,
262    limit: usize,
263) -> Result<Vec<ChangeWire>> {
264    let limit = limit as i64;
265    let rows = sqlx::query_as!(
266        ChangeRow,
267        r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
268         local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
269         entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
270         payload AS "payload!: String", base_version, created_at AS "created_at!: String",
271         server_seq
272         FROM changes WHERE server_seq IS NULL ORDER BY local_seq, created_at LIMIT ?"#,
273        limit,
274    )
275    .fetch_all(&mut *conn)
276    .await?;
277    Ok(rows.into_iter().map(ChangeRow::into_wire).collect())
278}
279
280async fn apply_sync_response(conn: &mut SqliteConnection, page: ApplySyncPage) -> Result<usize> {
281    let mut applied = 0;
282    let mut tx = begin_immediate(conn).await?;
283    let current_cursor = sync_cursor(&mut tx).await?;
284    if current_cursor != page.request.after {
285        bail!(
286            "error stale-sync-page expected_cursor={} request_cursor={}",
287            current_cursor,
288            page.request.after
289        );
290    }
291    update_change_server_seqs_if_missing(&mut tx, &page.response.push_acks).await?;
292    let existing_change_ids = load_existing_change_ids(&mut tx, &page.response.changes).await?;
293    let mut affected_series = HashSet::new();
294    for change in &page.response.changes {
295        if existing_change_ids.contains(change.change_id.as_str()) {
296            verify_existing_change(&mut tx, change).await?;
297            update_change_server_seq(&mut tx, &change.change_id, change.server_seq).await?;
298            continue;
299        }
300        let related_mutation = matches!(
301            change.op_type.as_str(),
302            op_type::RELATED_ADD | op_type::RELATED_REMOVE
303        );
304        if related_mutation {
305            insert_wire_change(&mut tx, change).await?;
306        }
307        apply_remote_change(&mut tx, change).await?;
308        if change.entity_type == "recurrence_series" {
309            let workspace_id = change
310                .payload
311                .get("workspace_id")
312                .and_then(serde_json::Value::as_str)
313                .context("recurrence change missing workspace_id")?;
314            affected_series.insert((workspace_id.to_string(), change.entity_id.clone()));
315        } else if let Some(series_id) = change
316            .payload
317            .get("series_id")
318            .and_then(serde_json::Value::as_str)
319        {
320            let workspace_id = change
321                .payload
322                .get("workspace_id")
323                .and_then(serde_json::Value::as_str)
324                .context("recurrence task change missing workspace_id")?;
325            affected_series.insert((workspace_id.to_string(), series_id.to_string()));
326        }
327        if !related_mutation {
328            insert_wire_change(&mut tx, change).await?;
329        }
330        applied += 1;
331    }
332    for (workspace_id, series_id) in affected_series {
333        let workspace_id: crate::ids::WorkspaceId = workspace_id.parse()?;
334        let series_id: crate::recurrence::RecurrenceSeriesId = series_id.parse()?;
335        let workspace = crate::workspaces::workspace_for_id(&mut tx, &workspace_id).await?;
336        let at =
337            chrono::DateTime::parse_from_rfc3339(&page.attempted_at)?.with_timezone(&chrono::Utc);
338        crate::operations::recurrence::reconcile_recurrence_series_in_transaction(
339            &mut tx, &workspace, &series_id, at,
340        )
341        .await?;
342    }
343    let pushed = page.previous_pushed + page.response.push_acks.len() as i64;
344    let pulled = page.previous_pulled + applied;
345    set_meta(&mut tx, "sync_cursor", &page.response.cursor.to_string()).await?;
346    set_meta(&mut tx, "sync_last_success_at", &page.attempted_at).await?;
347    set_meta(&mut tx, "sync_last_error", "").await?;
348    set_meta(&mut tx, "sync_last_pushed", &pushed.to_string()).await?;
349    set_meta(&mut tx, "sync_last_pulled", &pulled.to_string()).await?;
350    set_meta(
351        &mut tx,
352        "sync_last_cursor",
353        &page.response.cursor.to_string(),
354    )
355    .await?;
356    tx.commit().await?;
357    Ok(applied)
358}
359
360async fn load_existing_change_ids(
361    conn: &mut SqliteConnection,
362    changes: &[ChangeWire],
363) -> Result<HashSet<String>> {
364    if changes.is_empty() {
365        return Ok(HashSet::new());
366    }
367    let mut query_builder =
368        QueryBuilder::<Sqlite>::new("SELECT change_id FROM changes WHERE change_id IN (");
369    let mut separated = query_builder.separated(", ");
370    for change in changes {
371        separated.push_bind(&change.change_id);
372    }
373    separated.push_unseparated(")");
374    Ok(query_builder
375        .build_query_scalar::<String>()
376        .fetch_all(&mut *conn)
377        .await?
378        .into_iter()
379        .collect())
380}
381
382async fn load_assigned_change_ids(
383    conn: &mut SqliteConnection,
384    changes: &[ChangeWire],
385) -> Result<HashSet<String>> {
386    if changes.is_empty() {
387        return Ok(HashSet::new());
388    }
389    let mut query_builder = QueryBuilder::<Sqlite>::new(
390        "SELECT change_id FROM changes WHERE server_seq IS NOT NULL AND change_id IN (",
391    );
392    let mut separated = query_builder.separated(", ");
393    for change in changes {
394        separated.push_bind(&change.change_id);
395    }
396    separated.push_unseparated(")");
397    Ok(query_builder
398        .build_query_scalar::<String>()
399        .fetch_all(&mut *conn)
400        .await?
401        .into_iter()
402        .collect())
403}
404
405async fn verify_existing_change(conn: &mut SqliteConnection, incoming: &ChangeWire) -> Result<()> {
406    let row = sqlx::query(
407        "SELECT entity_type, entity_id, field, op_type, payload, base_version, created_at
408         FROM changes WHERE change_id = ?",
409    )
410    .bind(&incoming.change_id)
411    .fetch_one(&mut *conn)
412    .await?;
413    use sqlx::Row;
414    let stored_payload: String = row.try_get("payload")?;
415    let stored_payload: serde_json::Value = serde_json::from_str(&stored_payload)?;
416    let equal = row.try_get::<String, _>("entity_type")? == incoming.entity_type
417        && row.try_get::<String, _>("entity_id")? == incoming.entity_id
418        && row.try_get::<Option<String>, _>("field")? == incoming.field
419        && row.try_get::<String, _>("op_type")? == incoming.op_type
420        && stored_payload == incoming.payload
421        && row.try_get::<Option<String>, _>("base_version")? == incoming.base_version
422        && row.try_get::<String, _>("created_at")? == incoming.created_at;
423    if !equal {
424        bail!(
425            "error sync-change-id-payload-mismatch change_id={}",
426            incoming.change_id
427        );
428    }
429    Ok(())
430}
431
432async fn update_change_server_seq(
433    conn: &mut SqliteConnection,
434    change_id: &str,
435    server_seq: Option<i64>,
436) -> Result<()> {
437    if let Some(server_seq) = server_seq {
438        sqlx::query!(
439            "UPDATE changes SET server_seq = ? WHERE change_id = ? AND server_seq IS NULL",
440            server_seq,
441            change_id,
442        )
443        .execute(&mut *conn)
444        .await?;
445    }
446    Ok(())
447}
448
449async fn update_change_server_seqs_if_missing(
450    conn: &mut SqliteConnection,
451    push_acks: &[PushAck],
452) -> Result<()> {
453    if push_acks.is_empty() {
454        return Ok(());
455    }
456    let mut query_builder = QueryBuilder::<Sqlite>::new("WITH updates(change_id, server_seq) AS (");
457    query_builder.push_values(push_acks, |mut row, ack| {
458        row.push_bind(&ack.change_id).push_bind(ack.server_seq);
459    });
460    query_builder.push(
461        ") UPDATE changes
462         SET server_seq = (
463             SELECT updates.server_seq
464             FROM updates
465             WHERE updates.change_id = changes.change_id
466         )
467         WHERE server_seq IS NULL
468           AND change_id IN (SELECT change_id FROM updates)",
469    );
470    query_builder.build().execute(&mut *conn).await?;
471    Ok(())
472}
473
474async fn insert_wire_change(conn: &mut SqliteConnection, change: &ChangeWire) -> Result<()> {
475    let payload = change.payload.to_string();
476    sqlx::query!(
477        "INSERT OR IGNORE INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
478         op_type, payload, base_version, created_at, server_seq)
479         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
480        change.change_id,
481        change.client_id,
482        change.local_seq,
483        change.entity_type,
484        change.entity_id,
485        change.field,
486        change.op_type,
487        payload,
488        change.base_version,
489        change.created_at,
490        change.server_seq,
491    )
492    .execute(&mut *conn)
493    .await?;
494    Ok(())
495}
496
497async fn assign_server_sequences(
498    conn: &mut SqliteConnection,
499    changes: Vec<ChangeWire>,
500    blob_dir: Option<&Path>,
501) -> Result<(i64, Vec<PushAck>)> {
502    if changes.is_empty() {
503        return Ok((0, Vec::new()));
504    }
505    let mut tx = begin_immediate(conn).await?;
506    let assigned_change_ids = load_assigned_change_ids(&mut tx, &changes).await?;
507    let unassigned_changes = changes
508        .iter()
509        .filter(|change| !assigned_change_ids.contains(&change.change_id))
510        .cloned()
511        .collect::<Vec<_>>();
512    if unassigned_changes
513        .iter()
514        .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
515        && blob_dir.is_none()
516    {
517        bail!("error attachment-blob-storage-required");
518    }
519    if let Some(blob_dir) = blob_dir {
520        ensure_attachment_blobs_admitted(&mut tx, blob_dir, &unassigned_changes).await?;
521    }
522    let mut next_server_seq = next_available_server_seq(&mut tx).await?;
523    let mut accepted_count = 0_i64;
524    let mut push_acks = Vec::with_capacity(changes.len());
525    for change in changes {
526        let existing_server_seq = sqlx::query_scalar::<_, Option<i64>>(
527            "SELECT server_seq FROM changes WHERE change_id = ?",
528        )
529        .bind(&change.change_id)
530        .fetch_optional(&mut *tx)
531        .await?;
532        let server_seq = if let Some(existing_server_seq) = existing_server_seq {
533            verify_existing_change(&mut tx, &change).await?;
534            existing_server_seq.context("existing server change missing sequence")?
535        } else {
536            let server_seq = next_server_seq;
537            next_server_seq += 1;
538            let payload = change.payload.to_string();
539            sqlx::query(
540                "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
541                 op_type, payload, base_version, created_at, server_seq)
542                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
543            )
544            .bind(&change.change_id)
545            .bind(&change.client_id)
546            .bind(change.local_seq)
547            .bind(&change.entity_type)
548            .bind(&change.entity_id)
549            .bind(&change.field)
550            .bind(&change.op_type)
551            .bind(payload)
552            .bind(&change.base_version)
553            .bind(&change.created_at)
554            .bind(server_seq)
555            .execute(&mut *tx)
556            .await?;
557            apply_server_blob_reference(&mut tx, &change).await?;
558            accepted_count += 1;
559            server_seq
560        };
561        push_acks.push(PushAck {
562            change_id: change.change_id,
563            server_seq,
564        });
565    }
566    tx.commit().await?;
567    Ok((accepted_count, push_acks))
568}
569
570async fn next_available_server_seq(conn: &mut SqliteConnection) -> Result<i64> {
571    Ok(sqlx::query_scalar!(
572        r#"SELECT COALESCE(MAX(server_seq), 0) + 1 AS "seq!: i64" FROM changes"#
573    )
574    .fetch_one(&mut *conn)
575    .await?)
576}
577
578async fn apply_server_blob_reference(
579    conn: &mut SqliteConnection,
580    change: &ChangeWire,
581) -> Result<()> {
582    match change.op_type.as_str() {
583        op_type::ATTACHMENT_ADD => {
584            let payload = AttachmentAddPayload::from_change(change)?;
585            sqlx::query(
586                "INSERT INTO server_blob_references(
587                   workspace_id, attachment_id, task_id, sha256, byte_size, deleted
588                 ) VALUES (?, ?, ?, ?, ?, 0)
589                 ON CONFLICT(workspace_id, attachment_id) DO UPDATE SET
590                   task_id = excluded.task_id, sha256 = excluded.sha256,
591                   byte_size = excluded.byte_size, deleted = 0",
592            )
593            .bind(&payload.workspace_id)
594            .bind(&payload.attachment_id)
595            .bind(&change.entity_id)
596            .bind(&payload.sha256)
597            .bind(payload.byte_size)
598            .execute(&mut *conn)
599            .await?;
600            sqlx::query(
601                "DELETE FROM blob_upload_reservations WHERE workspace_id = ? AND sha256 = ?",
602            )
603            .bind(&payload.workspace_id)
604            .bind(&payload.sha256)
605            .execute(&mut *conn)
606            .await?;
607        }
608        op_type::ATTACHMENT_DELETE => {
609            sqlx::query(
610                "UPDATE server_blob_references SET deleted = 1
611                 WHERE workspace_id = ? AND attachment_id = ?",
612            )
613            .bind(
614                change.payload["workspace_id"]
615                    .as_str()
616                    .context("payload missing workspace_id")?,
617            )
618            .bind(
619                change.payload["attachment_id"]
620                    .as_str()
621                    .context("payload missing attachment_id")?,
622            )
623            .execute(&mut *conn)
624            .await?;
625        }
626        op_type::SET_FIELD if change.field.as_deref() == Some("deleted") => {
627            let deleted = change.payload["value"]
628                .as_str()
629                .is_some_and(|value| value == "1");
630            sqlx::query(
631                "INSERT INTO server_task_tombstones(workspace_id, task_id, deleted)
632                 VALUES (?, ?, ?)
633                 ON CONFLICT(workspace_id, task_id) DO UPDATE SET deleted = excluded.deleted",
634            )
635            .bind(
636                change.payload["workspace_id"]
637                    .as_str()
638                    .context("payload missing workspace_id")?,
639            )
640            .bind(&change.entity_id)
641            .bind(i64::from(deleted))
642            .execute(&mut *conn)
643            .await?;
644        }
645        _ => {}
646    }
647    Ok(())
648}
649
650async fn prepare_server_blobs(
651    conn: &mut SqliteConnection,
652    blob_dir: &Path,
653    changes: &[ChangeWire],
654) -> Result<()> {
655    let assigned_change_ids = load_assigned_change_ids(conn, changes).await?;
656    for change in changes {
657        if change.op_type != op_type::ATTACHMENT_ADD
658            || assigned_change_ids.contains(&change.change_id)
659        {
660            continue;
661        }
662        let contract = super::blob::attachment_blob_contract(change)?
663            .context("error attachment-blob-missing")?;
664        validate_server_blob_before_writer(conn, blob_dir, &contract).await?;
665    }
666    Ok(())
667}
668
669async fn validate_server_blob_before_writer(
670    conn: &mut SqliteConnection,
671    blob_dir: &Path,
672    contract: &super::wire::BlobUploadContract,
673) -> Result<()> {
674    let Some(row) = crate::attachments::storage::blob_inventory_row(conn, &contract.sha256).await?
675    else {
676        bail!("error attachment-blob-missing");
677    };
678    validate_server_blob_inventory(&row, contract)?;
679    let path = crate::attachments::object_path(blob_dir, &contract.sha256)?;
680    let bytes = match tokio::fs::read(path).await {
681        Ok(bytes) => bytes,
682        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
683            bail!("error attachment-blob-missing")
684        }
685        Err(error) => return Err(error.into()),
686    };
687    if i64::try_from(bytes.len()).ok() != Some(contract.byte_size)
688        || crate::attachments::storage::sha256_hex(&bytes) != contract.sha256
689    {
690        bail!("error attachment-blob-content-mismatch");
691    }
692    let validated =
693        crate::attachments::decode::validate_image(bytes, Some(contract.media_type.clone()))
694            .await?;
695    if (validated.facts.width, validated.facts.height) != (contract.width, contract.height) {
696        bail!("error blob-inventory-metadata-mismatch");
697    }
698    Ok(())
699}
700
701fn validate_server_blob_inventory(
702    row: &crate::types::BlobInventoryRow,
703    contract: &super::wire::BlobUploadContract,
704) -> Result<()> {
705    if !row.available {
706        bail!("error attachment-blob-missing");
707    }
708    if row.byte_size != contract.byte_size || row.media_type != contract.media_type {
709        bail!("error blob-inventory-metadata-mismatch");
710    }
711    Ok(())
712}
713
714async fn ensure_attachment_blobs_admitted(
715    conn: &mut SqliteConnection,
716    blob_dir: &Path,
717    changes: &[ChangeWire],
718) -> Result<()> {
719    for change in changes {
720        if change.op_type != op_type::ATTACHMENT_ADD {
721            continue;
722        }
723        let contract = super::blob::attachment_blob_contract(change)?
724            .context("error attachment-blob-missing")?;
725        let Some(row) =
726            crate::attachments::storage::blob_inventory_row(conn, &contract.sha256).await?
727        else {
728            bail!("error attachment-blob-missing");
729        };
730        validate_server_blob_inventory(&row, &contract)?;
731        if !crate::attachments::object_path(blob_dir, &contract.sha256)?.exists() {
732            bail!("error attachment-blob-missing");
733        }
734        let admitted: bool = sqlx::query_scalar(
735            "SELECT EXISTS(
736               SELECT 1 FROM server_blob_references sbr
737               LEFT JOIN server_task_tombstones st
738                 ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
739               WHERE sbr.workspace_id = ? AND sbr.sha256 = ? AND sbr.deleted = 0
740                 AND COALESCE(st.deleted, 0) = 0
741             ) OR EXISTS(
742               SELECT 1 FROM blob_upload_reservations
743               WHERE workspace_id = ? AND sha256 = ? AND byte_size = ? AND expires_at > ?
744             )",
745        )
746        .bind(&contract.workspace_id)
747        .bind(&contract.sha256)
748        .bind(&contract.workspace_id)
749        .bind(&contract.sha256)
750        .bind(contract.byte_size)
751        .bind(crate::ids::now())
752        .fetch_one(&mut *conn)
753        .await?;
754        if !admitted {
755            bail!("error attachment-blob-unreserved");
756        }
757    }
758    Ok(())
759}
760
761async fn load_server_changes_after(
762    conn: &mut SqliteConnection,
763    after: i64,
764    pull_limit: u32,
765) -> Result<(Vec<ChangeWire>, bool)> {
766    let fetch_limit = i64::from(pull_limit) + 1;
767    let rows = sqlx::query_as!(
768        ChangeRow,
769        r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
770         local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
771         entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
772         payload AS "payload!: String", base_version, created_at AS "created_at!: String",
773         server_seq
774         FROM changes WHERE server_seq > ? ORDER BY server_seq LIMIT ?"#,
775        after,
776        fetch_limit,
777    )
778    .fetch_all(&mut *conn)
779    .await?;
780    let has_more = rows.len() > pull_limit as usize;
781    let changes = rows
782        .into_iter()
783        .take(pull_limit as usize)
784        .map(ChangeRow::into_wire)
785        .collect();
786    Ok((changes, has_more))
787}
788
789#[cfg(test)]
790mod tests {
791    use std::io::Cursor;
792    use std::time::Duration;
793
794    use image::{DynamicImage, ImageFormat};
795    use serde_json::json;
796
797    use super::*;
798    use crate::attachments::storage::{object_path, sha256_hex, upsert_inventory_available};
799    use crate::sync::wire::{ChangeWire, SYNC_PROTOCOL_VERSION, SyncRequest};
800
801    #[tokio::test]
802    async fn related_comparison_observes_push_acknowledgement_first() {
803        let (_temp, mut conn) = crate::test_support::test_conn().await;
804        let workspace = crate::workspaces::Workspace::default();
805        let project = crate::projects::create_project(&mut conn, &workspace, "related-ack")
806            .await
807            .unwrap();
808        let task_id: crate::ids::TaskId = "AAAA000000000001".parse().unwrap();
809        let related_task_id: crate::ids::TaskId = "BBBB000000000002".parse().unwrap();
810        for id in [&task_id, &related_task_id] {
811            sqlx::query(
812                "INSERT INTO tasks(id, workspace_id, title, description, project_id, status, priority, created_at, updated_at)
813                 VALUES (?, ?, 'task', '', ?, 'todo', 'none', 't', 't')",
814            )
815            .bind(id)
816            .bind(&workspace.id)
817            .bind(&project.id)
818            .execute(&mut *conn)
819            .await
820            .unwrap();
821        }
822        let local = crate::operations::set_task_related_link_in_transaction(
823            &mut conn,
824            &workspace,
825            &task_id,
826            &related_task_id,
827            true,
828        )
829        .await
830        .unwrap();
831        let local_change_id = local.change_id.unwrap();
832        set_meta(&mut conn, "sync_cursor", "3").await.unwrap();
833
834        let remote = ChangeWire {
835            change_id: "CCCC000000000003".to_string(),
836            client_id: "remote".to_string(),
837            local_seq: 1,
838            entity_type: "task".to_string(),
839            entity_id: task_id.to_string(),
840            field: Some("related".to_string()),
841            op_type: op_type::RELATED_REMOVE.to_string(),
842            payload: json!({
843                "workspace_id": workspace.id,
844                "workspace_key": workspace.key,
845                "related_task_id": related_task_id,
846            }),
847            base_version: None,
848            created_at: "2026-08-22T00:00:00Z".to_string(),
849            server_seq: Some(4),
850        };
851        let page = ApplySyncPage {
852            request: SyncRequest {
853                protocol_version: Some(SYNC_PROTOCOL_VERSION),
854                client_id: "local".to_string(),
855                after: 3,
856                pull_limit: Some(100),
857                changes: Vec::new(),
858            },
859            response: SyncResponse {
860                protocol_version: SYNC_PROTOCOL_VERSION,
861                cursor: 4,
862                has_more: false,
863                push_acks: vec![PushAck {
864                    change_id: local_change_id.clone(),
865                    server_seq: 5,
866                }],
867                changes: vec![remote],
868            },
869            attempted_at: "2026-08-22T00:00:01Z".to_string(),
870            previous_pushed: 0,
871            previous_pulled: 0,
872        };
873
874        apply_sync_response(&mut conn, page).await.unwrap();
875
876        let state: (i64, String) = sqlx::query_as(
877            "SELECT linked, last_change_id FROM task_related_links
878             WHERE workspace_id = ? AND task_a_id = ? AND task_b_id = ?",
879        )
880        .bind(&workspace.id)
881        .bind(&task_id)
882        .bind(&related_task_id)
883        .fetch_one(&mut *conn)
884        .await
885        .unwrap();
886        assert_eq!(state, (1, local_change_id.clone()));
887        let acknowledged: Option<i64> =
888            sqlx::query_scalar("SELECT server_seq FROM changes WHERE change_id = ?")
889                .bind(local_change_id)
890                .fetch_one(&mut *conn)
891                .await
892                .unwrap();
893        assert_eq!(acknowledged, Some(5));
894    }
895
896    async fn corrupt_attachment_page() -> (
897        tempfile::TempDir,
898        Database,
899        std::path::PathBuf,
900        ServerSyncPage,
901    ) {
902        let temp = tempfile::tempdir().unwrap();
903        let database = Database::open(&temp.path().join("server.sqlite"))
904            .await
905            .unwrap();
906        let blob_dir = temp.path().join("blobs");
907        let expected = b"expected";
908        let sha256 = sha256_hex(expected);
909        {
910            let mut conn = database.acquire_writer().await.unwrap();
911            upsert_inventory_available(&mut conn, &sha256, expected.len() as i64, "image/png")
912                .await
913                .unwrap();
914        }
915        let path = object_path(&blob_dir, &sha256).unwrap();
916        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
917        std::fs::write(path, b"corrupt!").unwrap();
918        let change = ChangeWire {
919            change_id: "0123456789ABCDEF".to_string(),
920            client_id: "client-a".to_string(),
921            local_seq: 1,
922            entity_type: "task".to_string(),
923            entity_id: "0123456789ABCDE0".to_string(),
924            field: Some("attachments".to_string()),
925            op_type: op_type::ATTACHMENT_ADD.to_string(),
926            payload: json!({
927                "workspace_id": "0000000000000000",
928                "workspace_key": "default",
929                "attachment_id": "7KQ9A1X4MV2P8D6R",
930                "sha256": sha256,
931                "byte_size": expected.len(),
932                "media_type": "image/png",
933                "filename": "photo.png",
934                "alt_text": "photo",
935                "width": 1,
936                "height": 1,
937                "created_at": "2026-01-01T00:00:00Z"
938            }),
939            base_version: None,
940            created_at: "2026-01-01T00:00:00Z".to_string(),
941            server_seq: None,
942        };
943        let page = ServerSyncPage {
944            request: SyncRequest {
945                protocol_version: Some(SYNC_PROTOCOL_VERSION),
946                client_id: "client-a".to_string(),
947                after: 0,
948                pull_limit: Some(100),
949                changes: vec![change],
950            },
951        };
952        (temp, database, blob_dir, page)
953    }
954
955    fn page_for_contract(contract: &super::super::wire::BlobUploadContract) -> ServerSyncPage {
956        let change = ChangeWire {
957            change_id: "0123456789ABCDEF".to_string(),
958            client_id: "client-a".to_string(),
959            local_seq: 1,
960            entity_type: "task".to_string(),
961            entity_id: "0123456789ABCDE0".to_string(),
962            field: Some("attachments".to_string()),
963            op_type: op_type::ATTACHMENT_ADD.to_string(),
964            payload: json!({
965                "workspace_id": contract.workspace_id,
966                "workspace_key": "default",
967                "attachment_id": "7KQ9A1X4MV2P8D6R",
968                "sha256": contract.sha256,
969                "byte_size": contract.byte_size,
970                "media_type": contract.media_type,
971                "filename": "photo.png",
972                "alt_text": "photo",
973                "width": contract.width,
974                "height": contract.height,
975                "created_at": "2026-01-01T00:00:00Z"
976            }),
977            base_version: None,
978            created_at: "2026-01-01T00:00:00Z".to_string(),
979            server_seq: None,
980        };
981        ServerSyncPage {
982            request: SyncRequest {
983                protocol_version: Some(SYNC_PROTOCOL_VERSION),
984                client_id: "client-a".to_string(),
985                after: 0,
986                pull_limit: Some(100),
987                changes: vec![change],
988            },
989        }
990    }
991
992    #[tokio::test]
993    async fn corrupt_server_blob_is_rejected_before_writer_acquisition() {
994        let (_temp, database, blob_dir, page) = corrupt_attachment_page().await;
995        let writer = database.acquire_writer().await.unwrap();
996        let task = tokio::spawn({
997            let database = database.clone();
998            async move {
999                database
1000                    .persist_server_sync_page_with_blobs(page, &blob_dir)
1001                    .await
1002            }
1003        });
1004        tokio::time::sleep(Duration::from_millis(50)).await;
1005        assert!(
1006            task.is_finished(),
1007            "corrupt content validation should not wait for the writer gate"
1008        );
1009        drop(writer);
1010        let error = task.await.unwrap().unwrap_err();
1011        assert!(
1012            error
1013                .to_string()
1014                .contains("attachment-blob-content-mismatch")
1015        );
1016    }
1017
1018    #[tokio::test]
1019    async fn reservation_is_rechecked_after_blob_preparation() {
1020        let temp = tempfile::tempdir().unwrap();
1021        let database = Database::open(&temp.path().join("server.sqlite"))
1022            .await
1023            .unwrap();
1024        let blob_dir = temp.path().join("blobs");
1025        let mut encoded = Cursor::new(Vec::new());
1026        DynamicImage::new_rgba8(1, 1)
1027            .write_to(&mut encoded, ImageFormat::Png)
1028            .unwrap();
1029        let bytes = encoded.into_inner();
1030        let contract = super::super::wire::BlobUploadContract {
1031            workspace_id: "0000000000000000".to_string(),
1032            sha256: sha256_hex(&bytes),
1033            byte_size: bytes.len() as i64,
1034            media_type: "image/png".to_string(),
1035            width: 1,
1036            height: 1,
1037        };
1038        database
1039            .store_server_blob(
1040                &blob_dir,
1041                crate::attachments::lifecycle::LifecyclePolicy::default(),
1042                &contract,
1043                bytes,
1044            )
1045            .await
1046            .unwrap();
1047        let page = page_for_contract(&contract);
1048        {
1049            let mut reader = database.acquire_reader().await.unwrap();
1050            prepare_server_blobs(&mut reader, &blob_dir, &page.request.changes)
1051                .await
1052                .unwrap();
1053        }
1054        {
1055            let mut writer = database.acquire_writer().await.unwrap();
1056            sqlx::query("DELETE FROM blob_upload_reservations")
1057                .execute(&mut *writer)
1058                .await
1059                .unwrap();
1060        }
1061        let mut writer = database.acquire_writer().await.unwrap();
1062        let error = assign_server_sequences(&mut writer, page.request.changes, Some(&blob_dir))
1063            .await
1064            .unwrap_err();
1065        assert!(error.to_string().contains("attachment-blob-unreserved"));
1066        let accepted: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM changes")
1067            .fetch_one(&mut *writer)
1068            .await
1069            .unwrap();
1070        assert_eq!(accepted, 0);
1071    }
1072}
aven_core/sync/persistence.rs

aven_core/sync/
persistence.rs
