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

aven_core/sync/
persistence.rs
