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 apply_remote_change(&mut tx, change).await?;
301 if change.entity_type == "recurrence_series" {
302 let workspace_id = change
303 .payload
304 .get("workspace_id")
305 .and_then(serde_json::Value::as_str)
306 .context("recurrence change missing workspace_id")?;
307 affected_series.insert((workspace_id.to_string(), change.entity_id.clone()));
308 } else if let Some(series_id) = change
309 .payload
310 .get("series_id")
311 .and_then(serde_json::Value::as_str)
312 {
313 let workspace_id = change
314 .payload
315 .get("workspace_id")
316 .and_then(serde_json::Value::as_str)
317 .context("recurrence task change missing workspace_id")?;
318 affected_series.insert((workspace_id.to_string(), series_id.to_string()));
319 }
320 insert_wire_change(&mut tx, change).await?;
321 applied += 1;
322 }
323 for (workspace_id, series_id) in affected_series {
324 let workspace_id: crate::ids::WorkspaceId = workspace_id.parse()?;
325 let series_id: crate::recurrence::RecurrenceSeriesId = series_id.parse()?;
326 let workspace = crate::workspaces::workspace_for_id(&mut tx, &workspace_id).await?;
327 let at =
328 chrono::DateTime::parse_from_rfc3339(&page.attempted_at)?.with_timezone(&chrono::Utc);
329 crate::operations::recurrence::reconcile_recurrence_series_in_transaction(
330 &mut tx, &workspace, &series_id, at,
331 )
332 .await?;
333 }
334 let pushed = page.previous_pushed + page.response.push_acks.len() as i64;
335 let pulled = page.previous_pulled + applied;
336 set_meta(&mut tx, "sync_cursor", &page.response.cursor.to_string()).await?;
337 set_meta(&mut tx, "sync_last_success_at", &page.attempted_at).await?;
338 set_meta(&mut tx, "sync_last_error", "").await?;
339 set_meta(&mut tx, "sync_last_pushed", &pushed.to_string()).await?;
340 set_meta(&mut tx, "sync_last_pulled", &pulled.to_string()).await?;
341 set_meta(
342 &mut tx,
343 "sync_last_cursor",
344 &page.response.cursor.to_string(),
345 )
346 .await?;
347 tx.commit().await?;
348 Ok(applied)
349}
350
351async fn load_existing_change_ids(
352 conn: &mut SqliteConnection,
353 changes: &[ChangeWire],
354) -> Result<HashSet<String>> {
355 if changes.is_empty() {
356 return Ok(HashSet::new());
357 }
358 let mut query_builder =
359 QueryBuilder::<Sqlite>::new("SELECT change_id FROM changes WHERE change_id IN (");
360 let mut separated = query_builder.separated(", ");
361 for change in changes {
362 separated.push_bind(&change.change_id);
363 }
364 separated.push_unseparated(")");
365 Ok(query_builder
366 .build_query_scalar::<String>()
367 .fetch_all(&mut *conn)
368 .await?
369 .into_iter()
370 .collect())
371}
372
373async fn load_assigned_change_ids(
374 conn: &mut SqliteConnection,
375 changes: &[ChangeWire],
376) -> Result<HashSet<String>> {
377 if changes.is_empty() {
378 return Ok(HashSet::new());
379 }
380 let mut query_builder = QueryBuilder::<Sqlite>::new(
381 "SELECT change_id FROM changes WHERE server_seq IS NOT NULL AND change_id IN (",
382 );
383 let mut separated = query_builder.separated(", ");
384 for change in changes {
385 separated.push_bind(&change.change_id);
386 }
387 separated.push_unseparated(")");
388 Ok(query_builder
389 .build_query_scalar::<String>()
390 .fetch_all(&mut *conn)
391 .await?
392 .into_iter()
393 .collect())
394}
395
396async fn verify_existing_change(conn: &mut SqliteConnection, incoming: &ChangeWire) -> Result<()> {
397 let row = sqlx::query(
398 "SELECT entity_type, entity_id, field, op_type, payload, base_version, created_at
399 FROM changes WHERE change_id = ?",
400 )
401 .bind(&incoming.change_id)
402 .fetch_one(&mut *conn)
403 .await?;
404 use sqlx::Row;
405 let stored_payload: String = row.try_get("payload")?;
406 let stored_payload: serde_json::Value = serde_json::from_str(&stored_payload)?;
407 let equal = row.try_get::<String, _>("entity_type")? == incoming.entity_type
408 && row.try_get::<String, _>("entity_id")? == incoming.entity_id
409 && row.try_get::<Option<String>, _>("field")? == incoming.field
410 && row.try_get::<String, _>("op_type")? == incoming.op_type
411 && stored_payload == incoming.payload
412 && row.try_get::<Option<String>, _>("base_version")? == incoming.base_version
413 && row.try_get::<String, _>("created_at")? == incoming.created_at;
414 if !equal {
415 bail!(
416 "error sync-change-id-payload-mismatch change_id={}",
417 incoming.change_id
418 );
419 }
420 Ok(())
421}
422
423async fn update_change_server_seq(
424 conn: &mut SqliteConnection,
425 change_id: &str,
426 server_seq: Option<i64>,
427) -> Result<()> {
428 if let Some(server_seq) = server_seq {
429 sqlx::query!(
430 "UPDATE changes SET server_seq = ? WHERE change_id = ? AND server_seq IS NULL",
431 server_seq,
432 change_id,
433 )
434 .execute(&mut *conn)
435 .await?;
436 }
437 Ok(())
438}
439
440async fn update_change_server_seqs_if_missing(
441 conn: &mut SqliteConnection,
442 push_acks: &[PushAck],
443) -> Result<()> {
444 if push_acks.is_empty() {
445 return Ok(());
446 }
447 let mut query_builder = QueryBuilder::<Sqlite>::new("WITH updates(change_id, server_seq) AS (");
448 query_builder.push_values(push_acks, |mut row, ack| {
449 row.push_bind(&ack.change_id).push_bind(ack.server_seq);
450 });
451 query_builder.push(
452 ") UPDATE changes
453 SET server_seq = (
454 SELECT updates.server_seq
455 FROM updates
456 WHERE updates.change_id = changes.change_id
457 )
458 WHERE server_seq IS NULL
459 AND change_id IN (SELECT change_id FROM updates)",
460 );
461 query_builder.build().execute(&mut *conn).await?;
462 Ok(())
463}
464
465async fn insert_wire_change(conn: &mut SqliteConnection, change: &ChangeWire) -> Result<()> {
466 let payload = change.payload.to_string();
467 sqlx::query!(
468 "INSERT OR IGNORE INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
469 op_type, payload, base_version, created_at, server_seq)
470 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
471 change.change_id,
472 change.client_id,
473 change.local_seq,
474 change.entity_type,
475 change.entity_id,
476 change.field,
477 change.op_type,
478 payload,
479 change.base_version,
480 change.created_at,
481 change.server_seq,
482 )
483 .execute(&mut *conn)
484 .await?;
485 Ok(())
486}
487
488async fn assign_server_sequences(
489 conn: &mut SqliteConnection,
490 changes: Vec<ChangeWire>,
491 blob_dir: Option<&Path>,
492) -> Result<(i64, Vec<PushAck>)> {
493 if changes.is_empty() {
494 return Ok((0, Vec::new()));
495 }
496 let mut tx = begin_immediate(conn).await?;
497 let assigned_change_ids = load_assigned_change_ids(&mut tx, &changes).await?;
498 let unassigned_changes = changes
499 .iter()
500 .filter(|change| !assigned_change_ids.contains(&change.change_id))
501 .cloned()
502 .collect::<Vec<_>>();
503 if unassigned_changes
504 .iter()
505 .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
506 && blob_dir.is_none()
507 {
508 bail!("error attachment-blob-storage-required");
509 }
510 if let Some(blob_dir) = blob_dir {
511 ensure_attachment_blobs_admitted(&mut tx, blob_dir, &unassigned_changes).await?;
512 }
513 let mut next_server_seq = next_available_server_seq(&mut tx).await?;
514 let mut accepted_count = 0_i64;
515 let mut push_acks = Vec::with_capacity(changes.len());
516 for change in changes {
517 let existing_server_seq = sqlx::query_scalar::<_, Option<i64>>(
518 "SELECT server_seq FROM changes WHERE change_id = ?",
519 )
520 .bind(&change.change_id)
521 .fetch_optional(&mut *tx)
522 .await?;
523 let server_seq = if let Some(existing_server_seq) = existing_server_seq {
524 verify_existing_change(&mut tx, &change).await?;
525 existing_server_seq.context("existing server change missing sequence")?
526 } else {
527 let server_seq = next_server_seq;
528 next_server_seq += 1;
529 let payload = change.payload.to_string();
530 sqlx::query(
531 "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
532 op_type, payload, base_version, created_at, server_seq)
533 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
534 )
535 .bind(&change.change_id)
536 .bind(&change.client_id)
537 .bind(change.local_seq)
538 .bind(&change.entity_type)
539 .bind(&change.entity_id)
540 .bind(&change.field)
541 .bind(&change.op_type)
542 .bind(payload)
543 .bind(&change.base_version)
544 .bind(&change.created_at)
545 .bind(server_seq)
546 .execute(&mut *tx)
547 .await?;
548 apply_server_blob_reference(&mut tx, &change).await?;
549 accepted_count += 1;
550 server_seq
551 };
552 push_acks.push(PushAck {
553 change_id: change.change_id,
554 server_seq,
555 });
556 }
557 tx.commit().await?;
558 Ok((accepted_count, push_acks))
559}
560
561async fn next_available_server_seq(conn: &mut SqliteConnection) -> Result<i64> {
562 Ok(sqlx::query_scalar!(
563 r#"SELECT COALESCE(MAX(server_seq), 0) + 1 AS "seq!: i64" FROM changes"#
564 )
565 .fetch_one(&mut *conn)
566 .await?)
567}
568
569async fn apply_server_blob_reference(
570 conn: &mut SqliteConnection,
571 change: &ChangeWire,
572) -> Result<()> {
573 match change.op_type.as_str() {
574 op_type::ATTACHMENT_ADD => {
575 let payload = AttachmentAddPayload::from_change(change)?;
576 sqlx::query(
577 "INSERT INTO server_blob_references(
578 workspace_id, attachment_id, task_id, sha256, byte_size, deleted
579 ) VALUES (?, ?, ?, ?, ?, 0)
580 ON CONFLICT(workspace_id, attachment_id) DO UPDATE SET
581 task_id = excluded.task_id, sha256 = excluded.sha256,
582 byte_size = excluded.byte_size, deleted = 0",
583 )
584 .bind(&payload.workspace_id)
585 .bind(&payload.attachment_id)
586 .bind(&change.entity_id)
587 .bind(&payload.sha256)
588 .bind(payload.byte_size)
589 .execute(&mut *conn)
590 .await?;
591 sqlx::query(
592 "DELETE FROM blob_upload_reservations WHERE workspace_id = ? AND sha256 = ?",
593 )
594 .bind(&payload.workspace_id)
595 .bind(&payload.sha256)
596 .execute(&mut *conn)
597 .await?;
598 }
599 op_type::ATTACHMENT_DELETE => {
600 sqlx::query(
601 "UPDATE server_blob_references SET deleted = 1
602 WHERE workspace_id = ? AND attachment_id = ?",
603 )
604 .bind(
605 change.payload["workspace_id"]
606 .as_str()
607 .context("payload missing workspace_id")?,
608 )
609 .bind(
610 change.payload["attachment_id"]
611 .as_str()
612 .context("payload missing attachment_id")?,
613 )
614 .execute(&mut *conn)
615 .await?;
616 }
617 op_type::SET_FIELD if change.field.as_deref() == Some("deleted") => {
618 let deleted = change.payload["value"]
619 .as_str()
620 .is_some_and(|value| value == "1");
621 sqlx::query(
622 "INSERT INTO server_task_tombstones(workspace_id, task_id, deleted)
623 VALUES (?, ?, ?)
624 ON CONFLICT(workspace_id, task_id) DO UPDATE SET deleted = excluded.deleted",
625 )
626 .bind(
627 change.payload["workspace_id"]
628 .as_str()
629 .context("payload missing workspace_id")?,
630 )
631 .bind(&change.entity_id)
632 .bind(i64::from(deleted))
633 .execute(&mut *conn)
634 .await?;
635 }
636 _ => {}
637 }
638 Ok(())
639}
640
641async fn prepare_server_blobs(
642 conn: &mut SqliteConnection,
643 blob_dir: &Path,
644 changes: &[ChangeWire],
645) -> Result<()> {
646 let assigned_change_ids = load_assigned_change_ids(conn, changes).await?;
647 for change in changes {
648 if change.op_type != op_type::ATTACHMENT_ADD
649 || assigned_change_ids.contains(&change.change_id)
650 {
651 continue;
652 }
653 let contract = super::blob::attachment_blob_contract(change)?
654 .context("error attachment-blob-missing")?;
655 validate_server_blob_before_writer(conn, blob_dir, &contract).await?;
656 }
657 Ok(())
658}
659
660async fn validate_server_blob_before_writer(
661 conn: &mut SqliteConnection,
662 blob_dir: &Path,
663 contract: &super::wire::BlobUploadContract,
664) -> Result<()> {
665 let Some(row) = crate::attachments::storage::blob_inventory_row(conn, &contract.sha256).await?
666 else {
667 bail!("error attachment-blob-missing");
668 };
669 validate_server_blob_inventory(&row, contract)?;
670 let path = crate::attachments::object_path(blob_dir, &contract.sha256)?;
671 let bytes = match tokio::fs::read(path).await {
672 Ok(bytes) => bytes,
673 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
674 bail!("error attachment-blob-missing")
675 }
676 Err(error) => return Err(error.into()),
677 };
678 if i64::try_from(bytes.len()).ok() != Some(contract.byte_size)
679 || crate::attachments::storage::sha256_hex(&bytes) != contract.sha256
680 {
681 bail!("error attachment-blob-content-mismatch");
682 }
683 let validated =
684 crate::attachments::decode::validate_image(bytes, Some(contract.media_type.clone()))
685 .await?;
686 if (validated.facts.width, validated.facts.height) != (contract.width, contract.height) {
687 bail!("error blob-inventory-metadata-mismatch");
688 }
689 Ok(())
690}
691
692fn validate_server_blob_inventory(
693 row: &crate::types::BlobInventoryRow,
694 contract: &super::wire::BlobUploadContract,
695) -> Result<()> {
696 if !row.available {
697 bail!("error attachment-blob-missing");
698 }
699 if row.byte_size != contract.byte_size || row.media_type != contract.media_type {
700 bail!("error blob-inventory-metadata-mismatch");
701 }
702 Ok(())
703}
704
705async fn ensure_attachment_blobs_admitted(
706 conn: &mut SqliteConnection,
707 blob_dir: &Path,
708 changes: &[ChangeWire],
709) -> Result<()> {
710 for change in changes {
711 if change.op_type != op_type::ATTACHMENT_ADD {
712 continue;
713 }
714 let contract = super::blob::attachment_blob_contract(change)?
715 .context("error attachment-blob-missing")?;
716 let Some(row) =
717 crate::attachments::storage::blob_inventory_row(conn, &contract.sha256).await?
718 else {
719 bail!("error attachment-blob-missing");
720 };
721 validate_server_blob_inventory(&row, &contract)?;
722 if !crate::attachments::object_path(blob_dir, &contract.sha256)?.exists() {
723 bail!("error attachment-blob-missing");
724 }
725 let admitted: bool = sqlx::query_scalar(
726 "SELECT EXISTS(
727 SELECT 1 FROM server_blob_references sbr
728 LEFT JOIN server_task_tombstones st
729 ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
730 WHERE sbr.workspace_id = ? AND sbr.sha256 = ? AND sbr.deleted = 0
731 AND COALESCE(st.deleted, 0) = 0
732 ) OR EXISTS(
733 SELECT 1 FROM blob_upload_reservations
734 WHERE workspace_id = ? AND sha256 = ? AND byte_size = ? AND expires_at > ?
735 )",
736 )
737 .bind(&contract.workspace_id)
738 .bind(&contract.sha256)
739 .bind(&contract.workspace_id)
740 .bind(&contract.sha256)
741 .bind(contract.byte_size)
742 .bind(crate::ids::now())
743 .fetch_one(&mut *conn)
744 .await?;
745 if !admitted {
746 bail!("error attachment-blob-unreserved");
747 }
748 }
749 Ok(())
750}
751
752async fn load_server_changes_after(
753 conn: &mut SqliteConnection,
754 after: i64,
755 pull_limit: u32,
756) -> Result<(Vec<ChangeWire>, bool)> {
757 let fetch_limit = i64::from(pull_limit) + 1;
758 let rows = sqlx::query_as!(
759 ChangeRow,
760 r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
761 local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
762 entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
763 payload AS "payload!: String", base_version, created_at AS "created_at!: String",
764 server_seq
765 FROM changes WHERE server_seq > ? ORDER BY server_seq LIMIT ?"#,
766 after,
767 fetch_limit,
768 )
769 .fetch_all(&mut *conn)
770 .await?;
771 let has_more = rows.len() > pull_limit as usize;
772 let changes = rows
773 .into_iter()
774 .take(pull_limit as usize)
775 .map(ChangeRow::into_wire)
776 .collect();
777 Ok((changes, has_more))
778}
779
780#[cfg(test)]
781mod tests {
782 use std::io::Cursor;
783 use std::time::Duration;
784
785 use image::{DynamicImage, ImageFormat};
786 use serde_json::json;
787
788 use super::*;
789 use crate::attachments::storage::{object_path, sha256_hex, upsert_inventory_available};
790 use crate::sync::wire::{ChangeWire, SYNC_PROTOCOL_VERSION, SyncRequest};
791
792 async fn corrupt_attachment_page() -> (
793 tempfile::TempDir,
794 Database,
795 std::path::PathBuf,
796 ServerSyncPage,
797 ) {
798 let temp = tempfile::tempdir().unwrap();
799 let database = Database::open(&temp.path().join("server.sqlite"))
800 .await
801 .unwrap();
802 let blob_dir = temp.path().join("blobs");
803 let expected = b"expected";
804 let sha256 = sha256_hex(expected);
805 {
806 let mut conn = database.acquire_writer().await.unwrap();
807 upsert_inventory_available(&mut conn, &sha256, expected.len() as i64, "image/png")
808 .await
809 .unwrap();
810 }
811 let path = object_path(&blob_dir, &sha256).unwrap();
812 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
813 std::fs::write(path, b"corrupt!").unwrap();
814 let change = ChangeWire {
815 change_id: "0123456789ABCDEF".to_string(),
816 client_id: "client-a".to_string(),
817 local_seq: 1,
818 entity_type: "task".to_string(),
819 entity_id: "0123456789ABCDE0".to_string(),
820 field: Some("attachments".to_string()),
821 op_type: op_type::ATTACHMENT_ADD.to_string(),
822 payload: json!({
823 "workspace_id": "0000000000000000",
824 "workspace_key": "default",
825 "attachment_id": "7KQ9A1X4MV2P8D6R",
826 "sha256": sha256,
827 "byte_size": expected.len(),
828 "media_type": "image/png",
829 "filename": "photo.png",
830 "alt_text": "photo",
831 "width": 1,
832 "height": 1,
833 "created_at": "2026-01-01T00:00:00Z"
834 }),
835 base_version: None,
836 created_at: "2026-01-01T00:00:00Z".to_string(),
837 server_seq: None,
838 };
839 let page = ServerSyncPage {
840 request: SyncRequest {
841 protocol_version: Some(SYNC_PROTOCOL_VERSION),
842 client_id: "client-a".to_string(),
843 after: 0,
844 pull_limit: Some(100),
845 changes: vec![change],
846 },
847 };
848 (temp, database, blob_dir, page)
849 }
850
851 fn page_for_contract(contract: &super::super::wire::BlobUploadContract) -> ServerSyncPage {
852 let change = ChangeWire {
853 change_id: "0123456789ABCDEF".to_string(),
854 client_id: "client-a".to_string(),
855 local_seq: 1,
856 entity_type: "task".to_string(),
857 entity_id: "0123456789ABCDE0".to_string(),
858 field: Some("attachments".to_string()),
859 op_type: op_type::ATTACHMENT_ADD.to_string(),
860 payload: json!({
861 "workspace_id": contract.workspace_id,
862 "workspace_key": "default",
863 "attachment_id": "7KQ9A1X4MV2P8D6R",
864 "sha256": contract.sha256,
865 "byte_size": contract.byte_size,
866 "media_type": contract.media_type,
867 "filename": "photo.png",
868 "alt_text": "photo",
869 "width": contract.width,
870 "height": contract.height,
871 "created_at": "2026-01-01T00:00:00Z"
872 }),
873 base_version: None,
874 created_at: "2026-01-01T00:00:00Z".to_string(),
875 server_seq: None,
876 };
877 ServerSyncPage {
878 request: SyncRequest {
879 protocol_version: Some(SYNC_PROTOCOL_VERSION),
880 client_id: "client-a".to_string(),
881 after: 0,
882 pull_limit: Some(100),
883 changes: vec![change],
884 },
885 }
886 }
887
888 #[tokio::test]
889 async fn corrupt_server_blob_is_rejected_before_writer_acquisition() {
890 let (_temp, database, blob_dir, page) = corrupt_attachment_page().await;
891 let writer = database.acquire_writer().await.unwrap();
892 let task = tokio::spawn({
893 let database = database.clone();
894 async move {
895 database
896 .persist_server_sync_page_with_blobs(page, &blob_dir)
897 .await
898 }
899 });
900 tokio::time::sleep(Duration::from_millis(50)).await;
901 assert!(
902 task.is_finished(),
903 "corrupt content validation should not wait for the writer gate"
904 );
905 drop(writer);
906 let error = task.await.unwrap().unwrap_err();
907 assert!(
908 error
909 .to_string()
910 .contains("attachment-blob-content-mismatch")
911 );
912 }
913
914 #[tokio::test]
915 async fn reservation_is_rechecked_after_blob_preparation() {
916 let temp = tempfile::tempdir().unwrap();
917 let database = Database::open(&temp.path().join("server.sqlite"))
918 .await
919 .unwrap();
920 let blob_dir = temp.path().join("blobs");
921 let mut encoded = Cursor::new(Vec::new());
922 DynamicImage::new_rgba8(1, 1)
923 .write_to(&mut encoded, ImageFormat::Png)
924 .unwrap();
925 let bytes = encoded.into_inner();
926 let contract = super::super::wire::BlobUploadContract {
927 workspace_id: "0000000000000000".to_string(),
928 sha256: sha256_hex(&bytes),
929 byte_size: bytes.len() as i64,
930 media_type: "image/png".to_string(),
931 width: 1,
932 height: 1,
933 };
934 database
935 .store_server_blob(
936 &blob_dir,
937 crate::attachments::lifecycle::LifecyclePolicy::default(),
938 &contract,
939 bytes,
940 )
941 .await
942 .unwrap();
943 let page = page_for_contract(&contract);
944 {
945 let mut reader = database.acquire_reader().await.unwrap();
946 prepare_server_blobs(&mut reader, &blob_dir, &page.request.changes)
947 .await
948 .unwrap();
949 }
950 {
951 let mut writer = database.acquire_writer().await.unwrap();
952 sqlx::query("DELETE FROM blob_upload_reservations")
953 .execute(&mut *writer)
954 .await
955 .unwrap();
956 }
957 let mut writer = database.acquire_writer().await.unwrap();
958 let error = assign_server_sequences(&mut writer, page.request.changes, Some(&blob_dir))
959 .await
960 .unwrap_err();
961 assert!(error.to_string().contains("attachment-blob-unreserved"));
962 let accepted: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM changes")
963 .fetch_one(&mut *writer)
964 .await
965 .unwrap();
966 assert_eq!(accepted, 0);
967 }
968}