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, MAX_PUSH_BATCH, MAX_SYNC_REQUEST_BYTES, PushAck,
11 SyncRequest, SyncResponse,
12};
13use crate::change_log::op_type;
14use crate::db::{Database, begin_immediate, get_meta, set_meta};
15
16#[derive(Debug, Clone, Default, PartialEq, Eq)]
17pub struct SyncPersistenceStatus {
18 pub pinned_server: Option<String>,
19 pub pending_changes: i64,
20 pub pending_attachment_uploads: i64,
21 pub pending_attachment_upload_bytes: i64,
22 pub conflicts: i64,
23 pub sync_cursor: Option<String>,
24 pub local_sequence: Option<String>,
25 pub last_attempt: Option<String>,
26 pub last_success: Option<String>,
27 pub last_error: Option<String>,
28 pub last_pushed: Option<String>,
29 pub last_pulled: Option<String>,
30 pub last_cursor: Option<String>,
31}
32
33#[derive(Debug, Clone)]
34pub struct ClientSyncPage {
35 pub request: SyncRequest,
36 pub pending: usize,
37}
38
39#[derive(Debug)]
40pub struct ApplySyncPage {
41 pub request: SyncRequest,
42 pub response: SyncResponse,
43 pub attempted_at: String,
44 pub previous_pushed: i64,
45 pub previous_pulled: usize,
46}
47
48#[derive(Debug)]
49pub struct ServerSyncPage {
50 pub request: SyncRequest,
51}
52
53#[derive(Debug)]
54pub struct ServerSyncResult {
55 pub accepted_count: i64,
56 pub push_acks: Vec<PushAck>,
57 pub changes: Vec<ChangeWire>,
58 pub has_more: bool,
59 pub blob_prepare_ms: u128,
60 pub assign_ms: u128,
61 pub pull_query_ms: u128,
62}
63
64impl Database {
65 pub async fn sync_persistence_status(&self) -> Result<SyncPersistenceStatus> {
66 let mut conn = self.acquire_reader().await?;
67 let pending_changes =
68 sqlx::query_scalar("SELECT count(*) FROM changes WHERE server_seq IS NULL")
69 .fetch_one(&mut *conn)
70 .await?;
71 let (pending_attachment_uploads, pending_attachment_upload_bytes): (i64, i64) =
72 sqlx::query_as(
73 "SELECT COUNT(*), COALESCE(SUM(byte_size), 0)
74 FROM (
75 SELECT json_extract(payload, '$.workspace_id') AS workspace_id,
76 json_extract(payload, '$.sha256') AS sha256,
77 MAX(CAST(json_extract(payload, '$.byte_size') AS INTEGER)) AS byte_size
78 FROM changes
79 WHERE server_seq IS NULL AND op_type = 'attachment_add'
80 GROUP BY workspace_id, sha256
81 )",
82 )
83 .fetch_one(&mut *conn)
84 .await?;
85 let conflicts = sqlx::query_scalar("SELECT count(*) FROM conflicts WHERE resolved = 0")
86 .fetch_one(&mut *conn)
87 .await?;
88 Ok(SyncPersistenceStatus {
89 pinned_server: get_meta(&mut conn, "sync_server_url").await?,
90 pending_changes,
91 pending_attachment_uploads,
92 pending_attachment_upload_bytes,
93 conflicts,
94 sync_cursor: get_meta(&mut conn, "sync_cursor").await?,
95 local_sequence: get_meta(&mut conn, "local_seq").await?,
96 last_attempt: get_meta(&mut conn, "sync_last_attempt_at").await?,
97 last_success: get_meta(&mut conn, "sync_last_success_at").await?,
98 last_error: get_meta(&mut conn, "sync_last_error").await?,
99 last_pushed: get_meta(&mut conn, "sync_last_pushed").await?,
100 last_pulled: get_meta(&mut conn, "sync_last_pulled").await?,
101 last_cursor: get_meta(&mut conn, "sync_last_cursor").await?,
102 })
103 }
104
105 pub async fn begin_sync_attempt(&self, attempted_at: String) -> Result<()> {
106 let mut conn = self.acquire_writer().await?;
107 set_meta(&mut conn, "sync_last_attempt_at", &attempted_at).await
108 }
109
110 pub async fn record_sync_error(&self, error: String) -> Result<()> {
111 let mut conn = self.acquire_writer().await?;
112 set_meta(&mut conn, "sync_last_error", &error).await
113 }
114
115 pub(super) async fn pending_sync_changes_exist(&self) -> Result<bool> {
116 let mut conn = self.acquire_reader().await?;
117 Ok(
118 sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM changes WHERE server_seq IS NULL)")
119 .fetch_one(&mut *conn)
120 .await?,
121 )
122 }
123
124 pub(super) async fn pending_blob_counts(
125 &self,
126 known_server_blobs: &HashSet<String>,
127 ) -> Result<crate::attachments::lifecycle::ByteCount> {
128 let mut conn = self.acquire_reader().await?;
129 let known = serde_json::to_string(known_server_blobs)?;
130 let (count, bytes): (i64, i64) = sqlx::query_as(
131 "WITH known(sha256) AS (
132 SELECT value FROM json_each(?)
133 ), pending AS (
134 SELECT json_extract(payload, '$.workspace_id') AS workspace_id,
135 json_extract(payload, '$.sha256') AS sha256,
136 MAX(CAST(json_extract(payload, '$.byte_size') AS INTEGER)) AS byte_size
137 FROM changes
138 WHERE server_seq IS NULL AND op_type = 'attachment_add'
139 GROUP BY workspace_id, sha256
140 )
141 SELECT COUNT(*), COALESCE(SUM(byte_size), 0)
142 FROM pending LEFT JOIN known USING (sha256)
143 WHERE known.sha256 IS NULL",
144 )
145 .bind(known)
146 .fetch_one(&mut *conn)
147 .await?;
148 Ok(crate::attachments::lifecycle::ByteCount {
149 count: u64::try_from(count)?,
150 bytes: u64::try_from(bytes)?,
151 })
152 }
153
154 pub async fn prepare_client_sync_page(
155 &self,
156 server: String,
157 push_limit: usize,
158 pull_limit: u32,
159 ) -> Result<ClientSyncPage> {
160 let mut conn = self.acquire_writer().await?;
161 validate_sync_server(&mut conn, &server).await?;
162 let client_id = get_meta(&mut conn, "client_id")
163 .await?
164 .context("missing client id")?;
165 let after = sync_cursor(&mut conn).await?;
166 let changes = load_unsynced_changes(&mut conn, push_limit.min(MAX_PUSH_BATCH)).await?;
167 let request = bound_push_request(
168 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 MAX_SYNC_REQUEST_BYTES,
176 )?;
177 Ok(ClientSyncPage {
178 pending: request.changes.len(),
179 request,
180 })
181 }
182
183 pub async fn apply_client_sync_page(&self, page: ApplySyncPage) -> Result<usize> {
184 let envelope = super::wire::validate_sync_request_envelope(&page.request)?;
185 let request_change_ids = page
186 .request
187 .changes
188 .iter()
189 .map(|change| change.change_id.clone())
190 .collect::<Vec<_>>();
191 super::wire::validate_sync_response_for_request(
192 envelope.after,
193 envelope.pull_limit,
194 &request_change_ids,
195 &page.response,
196 )?;
197 let mut conn = self.acquire_writer().await?;
198 apply_sync_response(&mut conn, page).await
199 }
200
201 pub async fn persist_server_sync_page(&self, page: ServerSyncPage) -> Result<ServerSyncResult> {
202 self.persist_server_sync_page_inner(page, None).await
203 }
204
205 pub async fn persist_server_sync_page_with_blobs(
206 &self,
207 page: ServerSyncPage,
208 blob_dir: &Path,
209 ) -> Result<ServerSyncResult> {
210 self.persist_server_sync_page_inner(page, Some(blob_dir))
211 .await
212 }
213
214 async fn persist_server_sync_page_inner(
215 &self,
216 page: ServerSyncPage,
217 blob_dir: Option<&Path>,
218 ) -> Result<ServerSyncResult> {
219 let envelope = super::wire::validate_sync_request_envelope(&page.request)?;
220 for change in &page.request.changes {
221 super::wire::validate_pushed_change(change)?;
222 }
223 if blob_dir.is_none()
224 && page
225 .request
226 .changes
227 .iter()
228 .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
229 {
230 bail!("error attachment-blob-storage-required");
231 }
232 let blob_prepare_started = Instant::now();
233 if let Some(blob_dir) = blob_dir {
234 let mut conn = self.acquire_reader().await?;
235 prepare_server_blobs(&mut conn, blob_dir, &page.request.changes).await?;
236 }
237 let blob_prepare_ms = blob_prepare_started.elapsed().as_millis();
238 let mut conn = self.acquire_writer().await?;
239 let assign_started = Instant::now();
240 let (accepted_count, push_acks) =
241 assign_server_sequences(&mut conn, page.request.changes, blob_dir).await?;
242 let assign_ms = assign_started.elapsed().as_millis();
243 let pull_query_started = Instant::now();
244 let (changes, has_more) =
245 load_server_changes_after(&mut conn, envelope.after, envelope.pull_limit).await?;
246 let pull_query_ms = pull_query_started.elapsed().as_millis();
247 Ok(ServerSyncResult {
248 accepted_count,
249 push_acks,
250 changes,
251 has_more,
252 blob_prepare_ms,
253 assign_ms,
254 pull_query_ms,
255 })
256 }
257}
258
259async fn sync_cursor(conn: &mut SqliteConnection) -> Result<i64> {
260 Ok(get_meta(conn, "sync_cursor")
261 .await?
262 .unwrap_or_else(|| "0".to_string())
263 .parse::<i64>()?)
264}
265
266async fn validate_sync_server(conn: &mut SqliteConnection, server: &str) -> Result<()> {
267 let normalized = server.trim_end_matches('/');
268 if let Some(existing) = get_meta(conn, "sync_server_url").await? {
269 if existing != normalized {
270 bail!(
271 "error sync-server-changed existing={} requested={} hint=\"use a fresh database for a different sync server\"",
272 existing,
273 normalized
274 );
275 }
276 } else {
277 set_meta(conn, "sync_server_url", normalized).await?;
278 }
279 Ok(())
280}
281
282fn bound_push_request(mut request: SyncRequest, byte_limit: usize) -> Result<SyncRequest> {
283 let changes = std::mem::take(&mut request.changes);
284 let mut bytes = serde_json::to_vec(&request)?.len();
286 if bytes > byte_limit {
287 bail!("error sync-request-envelope-too-large limit={byte_limit}");
288 }
289 for change in changes.into_iter().take(MAX_PUSH_BATCH) {
290 let change_bytes = serde_json::to_vec(&change)?.len();
291 let separator = usize::from(!request.changes.is_empty());
292 if change_bytes + separator > byte_limit - bytes {
293 if request.changes.is_empty() {
294 bail!(
295 "error sync-change-exceeds-request-budget local_seq={} limit={byte_limit} hint=repair-pending-change",
296 change.local_seq
297 );
298 }
299 break;
300 }
301 bytes += change_bytes + separator;
302 request.changes.push(change);
303 }
304 Ok(request)
305}
306
307async fn load_unsynced_changes(
308 conn: &mut SqliteConnection,
309 limit: usize,
310) -> Result<Vec<ChangeWire>> {
311 let limit = limit as i64;
312 let rows = sqlx::query_as!(
313 ChangeRow,
314 r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
315 local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
316 entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
317 payload AS "payload!: String", base_version, created_at AS "created_at!: String",
318 server_seq
319 FROM changes WHERE server_seq IS NULL ORDER BY local_seq, created_at LIMIT ?"#,
320 limit,
321 )
322 .fetch_all(&mut *conn)
323 .await?;
324 Ok(rows.into_iter().map(ChangeRow::into_wire).collect())
325}
326
327async fn apply_sync_response(conn: &mut SqliteConnection, page: ApplySyncPage) -> Result<usize> {
328 let mut applied = 0;
329 let mut tx = begin_immediate(conn).await?;
330 let current_cursor = sync_cursor(&mut tx).await?;
331 if current_cursor != page.request.after {
332 bail!(
333 "error stale-sync-page expected_cursor={} request_cursor={}",
334 current_cursor,
335 page.request.after
336 );
337 }
338 update_change_server_seqs_if_missing(&mut tx, &page.response.push_acks).await?;
339 reconcile_acknowledged_epic_memberships(&mut tx, &page.response.push_acks).await?;
340 let existing_change_ids = load_existing_change_ids(&mut tx, &page.response.changes).await?;
341 let mut affected_series = HashSet::new();
342 let mut affected_attachment_hashes = HashSet::new();
343 for change in &page.response.changes {
344 if existing_change_ids.contains(change.change_id.as_str()) {
345 verify_existing_change(&mut tx, change).await?;
346 update_change_server_seq(&mut tx, &change.change_id, change.server_seq).await?;
347 reconcile_epic_change(&mut tx, change).await?;
348 continue;
349 }
350 collect_attachment_liveness_hashes(&mut tx, change, &mut affected_attachment_hashes)
351 .await?;
352 if is_epic_change(change) {
353 let workspace_id = epic_change_workspace(change)?;
354 crate::epic_membership::capture_snapshot_baseline(
355 &mut tx,
356 workspace_id,
357 &change.entity_id,
358 )
359 .await?;
360 }
361 let related_mutation = matches!(
362 change.op_type.as_str(),
363 op_type::RELATED_ADD | op_type::RELATED_REMOVE
364 );
365 if related_mutation {
366 insert_wire_change(&mut tx, change).await?;
367 }
368 apply_remote_change(&mut tx, change).await?;
369 if change.entity_type == "recurrence_series" {
370 let workspace_id = change
371 .payload
372 .get("workspace_id")
373 .and_then(serde_json::Value::as_str)
374 .context("recurrence change missing workspace_id")?;
375 affected_series.insert((workspace_id.to_string(), change.entity_id.clone()));
376 } else if let Some(series_id) = change
377 .payload
378 .get("series_id")
379 .and_then(serde_json::Value::as_str)
380 {
381 let workspace_id = change
382 .payload
383 .get("workspace_id")
384 .and_then(serde_json::Value::as_str)
385 .context("recurrence task change missing workspace_id")?;
386 affected_series.insert((workspace_id.to_string(), series_id.to_string()));
387 }
388 if !related_mutation {
389 insert_wire_change(&mut tx, change).await?;
390 }
391 reconcile_epic_change(&mut tx, change).await?;
392 applied += 1;
393 }
394 for (workspace_id, series_id) in affected_series {
395 let workspace_id: crate::ids::WorkspaceId = workspace_id.parse()?;
396 let series_id: crate::recurrence::RecurrenceSeriesId = series_id.parse()?;
397 let workspace = crate::workspaces::workspace_for_id(&mut tx, &workspace_id).await?;
398 let at =
399 chrono::DateTime::parse_from_rfc3339(&page.attempted_at)?.with_timezone(&chrono::Utc);
400 crate::operations::recurrence::reconcile_recurrence_series_in_transaction(
401 &mut tx, &workspace, &series_id, at,
402 )
403 .await?;
404 }
405 let affected_attachment_hashes = affected_attachment_hashes.into_iter().collect::<Vec<_>>();
406 crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
407 &mut tx,
408 &affected_attachment_hashes,
409 &crate::attachments::lifecycle::SystemClock,
410 )
411 .await?;
412 let pushed = page.previous_pushed + page.response.push_acks.len() as i64;
413 let pulled = page.previous_pulled + applied;
414 set_meta(&mut tx, "sync_cursor", &page.response.cursor.to_string()).await?;
415 set_meta(&mut tx, "sync_last_success_at", &page.attempted_at).await?;
416 set_meta(&mut tx, "sync_last_error", "").await?;
417 set_meta(&mut tx, "sync_last_pushed", &pushed.to_string()).await?;
418 set_meta(&mut tx, "sync_last_pulled", &pulled.to_string()).await?;
419 set_meta(
420 &mut tx,
421 "sync_last_cursor",
422 &page.response.cursor.to_string(),
423 )
424 .await?;
425 tx.commit().await?;
426 Ok(applied)
427}
428
429fn is_epic_change(change: &ChangeWire) -> bool {
430 matches!(
431 change.op_type.as_str(),
432 op_type::EPIC_LINK_ADD | op_type::EPIC_LINK_REMOVE
433 )
434}
435
436fn epic_change_workspace(change: &ChangeWire) -> Result<&str> {
437 change.payload["workspace_id"]
438 .as_str()
439 .context("epic change missing workspace_id")
440}
441
442async fn reconcile_epic_change(conn: &mut SqliteConnection, change: &ChangeWire) -> Result<()> {
443 if is_epic_change(change) {
444 crate::epic_membership::reconcile_child(
445 conn,
446 epic_change_workspace(change)?,
447 &change.entity_id,
448 )
449 .await?;
450 }
451 Ok(())
452}
453
454async fn reconcile_acknowledged_epic_memberships(
455 conn: &mut SqliteConnection,
456 acknowledgements: &[PushAck],
457) -> Result<()> {
458 if acknowledgements.is_empty() {
459 return Ok(());
460 }
461 let mut query = QueryBuilder::<Sqlite>::new(
462 "SELECT DISTINCT json_extract(payload, '$.workspace_id'), entity_id FROM changes
463 WHERE op_type IN ('epic_link_add', 'epic_link_remove') AND change_id IN (",
464 );
465 let mut ids = query.separated(", ");
466 for acknowledgement in acknowledgements {
467 ids.push_bind(&acknowledgement.change_id);
468 }
469 ids.push_unseparated(")");
470 let children = query
471 .build_query_as::<(String, String)>()
472 .fetch_all(&mut *conn)
473 .await?;
474 for (workspace_id, child_id) in children {
475 crate::epic_membership::reconcile_child(conn, &workspace_id, &child_id).await?;
476 }
477 Ok(())
478}
479
480async fn load_existing_change_ids(
481 conn: &mut SqliteConnection,
482 changes: &[ChangeWire],
483) -> Result<HashSet<String>> {
484 if changes.is_empty() {
485 return Ok(HashSet::new());
486 }
487 let mut query_builder =
488 QueryBuilder::<Sqlite>::new("SELECT change_id FROM changes WHERE change_id IN (");
489 let mut separated = query_builder.separated(", ");
490 for change in changes {
491 separated.push_bind(&change.change_id);
492 }
493 separated.push_unseparated(")");
494 Ok(query_builder
495 .build_query_scalar::<String>()
496 .fetch_all(&mut *conn)
497 .await?
498 .into_iter()
499 .collect())
500}
501
502async fn load_assigned_change_ids(
503 conn: &mut SqliteConnection,
504 changes: &[ChangeWire],
505) -> Result<HashSet<String>> {
506 if changes.is_empty() {
507 return Ok(HashSet::new());
508 }
509 let mut query_builder = QueryBuilder::<Sqlite>::new(
510 "SELECT change_id FROM changes WHERE server_seq IS NOT NULL AND change_id IN (",
511 );
512 let mut separated = query_builder.separated(", ");
513 for change in changes {
514 separated.push_bind(&change.change_id);
515 }
516 separated.push_unseparated(")");
517 Ok(query_builder
518 .build_query_scalar::<String>()
519 .fetch_all(&mut *conn)
520 .await?
521 .into_iter()
522 .collect())
523}
524
525async fn verify_existing_change(conn: &mut SqliteConnection, incoming: &ChangeWire) -> Result<()> {
526 let row = sqlx::query(
527 "SELECT entity_type, entity_id, field, op_type, payload, base_version, created_at
528 FROM changes WHERE change_id = ?",
529 )
530 .bind(&incoming.change_id)
531 .fetch_one(&mut *conn)
532 .await?;
533 use sqlx::Row;
534 let stored_payload: String = row.try_get("payload")?;
535 let stored_payload: serde_json::Value = serde_json::from_str(&stored_payload)?;
536 let equal = row.try_get::<String, _>("entity_type")? == incoming.entity_type
537 && row.try_get::<String, _>("entity_id")? == incoming.entity_id
538 && row.try_get::<Option<String>, _>("field")? == incoming.field
539 && row.try_get::<String, _>("op_type")? == incoming.op_type
540 && stored_payload == incoming.payload
541 && row.try_get::<Option<String>, _>("base_version")? == incoming.base_version
542 && row.try_get::<String, _>("created_at")? == incoming.created_at;
543 if !equal {
544 bail!(
545 "error sync-change-id-payload-mismatch change_id={}",
546 incoming.change_id
547 );
548 }
549 Ok(())
550}
551
552async fn update_change_server_seq(
553 conn: &mut SqliteConnection,
554 change_id: &str,
555 server_seq: Option<i64>,
556) -> Result<()> {
557 if let Some(server_seq) = server_seq {
558 sqlx::query!(
559 "UPDATE changes SET server_seq = ? WHERE change_id = ? AND server_seq IS NULL",
560 server_seq,
561 change_id,
562 )
563 .execute(&mut *conn)
564 .await?;
565 }
566 Ok(())
567}
568
569async fn update_change_server_seqs_if_missing(
570 conn: &mut SqliteConnection,
571 push_acks: &[PushAck],
572) -> Result<()> {
573 if push_acks.is_empty() {
574 return Ok(());
575 }
576 let mut query_builder = QueryBuilder::<Sqlite>::new("WITH updates(change_id, server_seq) AS (");
577 query_builder.push_values(push_acks, |mut row, ack| {
578 row.push_bind(&ack.change_id).push_bind(ack.server_seq);
579 });
580 query_builder.push(
581 ") UPDATE changes
582 SET server_seq = (
583 SELECT updates.server_seq
584 FROM updates
585 WHERE updates.change_id = changes.change_id
586 )
587 WHERE server_seq IS NULL
588 AND change_id IN (SELECT change_id FROM updates)",
589 );
590 query_builder.build().execute(&mut *conn).await?;
591 Ok(())
592}
593
594async fn insert_wire_change(conn: &mut SqliteConnection, change: &ChangeWire) -> Result<()> {
595 let payload = change.payload.to_string();
596 sqlx::query!(
597 "INSERT OR IGNORE INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
598 op_type, payload, base_version, created_at, server_seq)
599 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
600 change.change_id,
601 change.client_id,
602 change.local_seq,
603 change.entity_type,
604 change.entity_id,
605 change.field,
606 change.op_type,
607 payload,
608 change.base_version,
609 change.created_at,
610 change.server_seq,
611 )
612 .execute(&mut *conn)
613 .await?;
614 Ok(())
615}
616
617async fn assign_server_sequences(
618 conn: &mut SqliteConnection,
619 changes: Vec<ChangeWire>,
620 blob_dir: Option<&Path>,
621) -> Result<(i64, Vec<PushAck>)> {
622 if changes.is_empty() {
623 return Ok((0, Vec::new()));
624 }
625 let mut tx = begin_immediate(conn).await?;
626 let assigned_change_ids = load_assigned_change_ids(&mut tx, &changes).await?;
627 let unassigned_changes = changes
628 .iter()
629 .filter(|change| !assigned_change_ids.contains(&change.change_id))
630 .cloned()
631 .collect::<Vec<_>>();
632 if unassigned_changes
633 .iter()
634 .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
635 && blob_dir.is_none()
636 {
637 bail!("error attachment-blob-storage-required");
638 }
639 if let Some(blob_dir) = blob_dir {
640 ensure_attachment_blobs_admitted(&mut tx, blob_dir, &unassigned_changes).await?;
641 }
642 let mut next_server_seq = next_available_server_seq(&mut tx).await?;
643 let mut accepted_count = 0_i64;
644 let mut push_acks = Vec::with_capacity(changes.len());
645 let mut affected_attachment_hashes = HashSet::new();
646 for change in changes {
647 let existing_server_seq = sqlx::query_scalar::<_, Option<i64>>(
648 "SELECT server_seq FROM changes WHERE change_id = ?",
649 )
650 .bind(&change.change_id)
651 .fetch_optional(&mut *tx)
652 .await?;
653 let server_seq = if let Some(existing_server_seq) = existing_server_seq {
654 verify_existing_change(&mut tx, &change).await?;
655 existing_server_seq.context("existing server change missing sequence")?
656 } else {
657 let server_seq = next_server_seq;
658 next_server_seq += 1;
659 let payload = change.payload.to_string();
660 sqlx::query(
661 "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
662 op_type, payload, base_version, created_at, server_seq)
663 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
664 )
665 .bind(&change.change_id)
666 .bind(&change.client_id)
667 .bind(change.local_seq)
668 .bind(&change.entity_type)
669 .bind(&change.entity_id)
670 .bind(&change.field)
671 .bind(&change.op_type)
672 .bind(payload)
673 .bind(&change.base_version)
674 .bind(&change.created_at)
675 .bind(server_seq)
676 .execute(&mut *tx)
677 .await?;
678 apply_server_blob_reference(&mut tx, &change, &mut affected_attachment_hashes).await?;
679 accepted_count += 1;
680 server_seq
681 };
682 push_acks.push(PushAck {
683 change_id: change.change_id,
684 server_seq,
685 });
686 }
687 let affected_attachment_hashes = affected_attachment_hashes.into_iter().collect::<Vec<_>>();
688 crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
689 &mut tx,
690 &affected_attachment_hashes,
691 &crate::attachments::lifecycle::SystemClock,
692 )
693 .await?;
694 tx.commit().await?;
695 Ok((accepted_count, push_acks))
696}
697
698async fn next_available_server_seq(conn: &mut SqliteConnection) -> Result<i64> {
699 Ok(sqlx::query_scalar!(
700 r#"SELECT COALESCE(MAX(server_seq), 0) + 1 AS "seq!: i64" FROM changes"#
701 )
702 .fetch_one(&mut *conn)
703 .await?)
704}
705
706async fn apply_server_blob_reference(
707 conn: &mut SqliteConnection,
708 change: &ChangeWire,
709 affected_attachment_hashes: &mut HashSet<String>,
710) -> Result<()> {
711 match change.op_type.as_str() {
712 op_type::ATTACHMENT_ADD => {
713 let payload = AttachmentAddPayload::from_change(change)?;
714 let previous_sha256: Option<String> = sqlx::query_scalar(
715 "SELECT sha256 FROM server_blob_references
716 WHERE workspace_id = ? AND attachment_id = ?",
717 )
718 .bind(&payload.workspace_id)
719 .bind(&payload.attachment_id)
720 .fetch_optional(&mut *conn)
721 .await?;
722 if let Some(previous_sha256) = previous_sha256 {
723 affected_attachment_hashes.insert(previous_sha256);
724 }
725 affected_attachment_hashes.insert(payload.sha256.clone());
726 sqlx::query(
727 "INSERT INTO server_blob_references(
728 workspace_id, attachment_id, task_id, sha256, byte_size, deleted
729 ) VALUES (?, ?, ?, ?, ?, 0)
730 ON CONFLICT(workspace_id, attachment_id) DO UPDATE SET
731 task_id = excluded.task_id, sha256 = excluded.sha256,
732 byte_size = excluded.byte_size, deleted = 0",
733 )
734 .bind(&payload.workspace_id)
735 .bind(&payload.attachment_id)
736 .bind(&change.entity_id)
737 .bind(&payload.sha256)
738 .bind(payload.byte_size)
739 .execute(&mut *conn)
740 .await?;
741 sqlx::query(
742 "DELETE FROM blob_upload_reservations WHERE workspace_id = ? AND sha256 = ?",
743 )
744 .bind(&payload.workspace_id)
745 .bind(&payload.sha256)
746 .execute(&mut *conn)
747 .await?;
748 }
749 op_type::ATTACHMENT_DELETE => {
750 let workspace_id = change.payload["workspace_id"]
751 .as_str()
752 .context("payload missing workspace_id")?;
753 let attachment_id = change.payload["attachment_id"]
754 .as_str()
755 .context("payload missing attachment_id")?;
756 let sha256: Option<String> = sqlx::query_scalar(
757 "SELECT sha256 FROM server_blob_references
758 WHERE workspace_id = ? AND attachment_id = ?",
759 )
760 .bind(workspace_id)
761 .bind(attachment_id)
762 .fetch_optional(&mut *conn)
763 .await?;
764 if let Some(sha256) = sha256 {
765 affected_attachment_hashes.insert(sha256);
766 }
767 sqlx::query(
768 "UPDATE server_blob_references SET deleted = 1
769 WHERE workspace_id = ? AND attachment_id = ?",
770 )
771 .bind(workspace_id)
772 .bind(attachment_id)
773 .execute(&mut *conn)
774 .await?;
775 }
776 op_type::SET_FIELD | op_type::RESOLVE_FIELD
777 if change.field.as_deref() == Some("deleted") =>
778 {
779 let workspace_id = change.payload["workspace_id"]
780 .as_str()
781 .context("payload missing workspace_id")?;
782 let hashes: Vec<String> = sqlx::query_scalar(
783 "SELECT DISTINCT sha256 FROM server_blob_references
784 WHERE workspace_id = ? AND task_id = ?",
785 )
786 .bind(workspace_id)
787 .bind(&change.entity_id)
788 .fetch_all(&mut *conn)
789 .await?;
790 affected_attachment_hashes.extend(hashes);
791 let deleted = change.payload["value"]
792 .as_str()
793 .is_some_and(|value| value == "1");
794 sqlx::query(
795 "INSERT INTO server_task_tombstones(workspace_id, task_id, deleted)
796 VALUES (?, ?, ?)
797 ON CONFLICT(workspace_id, task_id) DO UPDATE SET deleted = excluded.deleted",
798 )
799 .bind(workspace_id)
800 .bind(&change.entity_id)
801 .bind(i64::from(deleted))
802 .execute(&mut *conn)
803 .await?;
804 }
805 _ => {}
806 }
807 Ok(())
808}
809
810async fn collect_attachment_liveness_hashes(
811 conn: &mut SqliteConnection,
812 change: &ChangeWire,
813 affected_attachment_hashes: &mut HashSet<String>,
814) -> Result<()> {
815 match change.op_type.as_str() {
816 op_type::ATTACHMENT_ADD => {
817 let payload = AttachmentAddPayload::from_change(change)?;
818 affected_attachment_hashes.insert(payload.sha256);
819 }
820 op_type::ATTACHMENT_DELETE => {
821 let workspace_id = change.payload["workspace_id"]
822 .as_str()
823 .context("payload missing workspace_id")?;
824 let attachment_id = change.payload["attachment_id"]
825 .as_str()
826 .context("payload missing attachment_id")?;
827 let sha256: Option<String> = sqlx::query_scalar(
828 "SELECT sha256 FROM task_attachments
829 WHERE workspace_id = ? AND attachment_id = ?",
830 )
831 .bind(workspace_id)
832 .bind(attachment_id)
833 .fetch_optional(&mut *conn)
834 .await?;
835 if let Some(sha256) = sha256 {
836 affected_attachment_hashes.insert(sha256);
837 }
838 }
839 op_type::SET_FIELD | op_type::RESOLVE_FIELD
840 if change.field.as_deref() == Some("deleted") =>
841 {
842 let workspace_id = change.payload["workspace_id"]
843 .as_str()
844 .context("payload missing workspace_id")?;
845 let hashes: Vec<String> = sqlx::query_scalar(
846 "SELECT DISTINCT sha256 FROM task_attachments
847 WHERE workspace_id = ? AND task_id = ?",
848 )
849 .bind(workspace_id)
850 .bind(&change.entity_id)
851 .fetch_all(&mut *conn)
852 .await?;
853 affected_attachment_hashes.extend(hashes);
854 }
855 _ => {}
856 }
857 Ok(())
858}
859
860async fn prepare_server_blobs(
861 conn: &mut SqliteConnection,
862 blob_dir: &Path,
863 changes: &[ChangeWire],
864) -> Result<()> {
865 let assigned_change_ids = load_assigned_change_ids(conn, changes).await?;
866 let contracts = changes
867 .iter()
868 .filter(|change| {
869 change.op_type == op_type::ATTACHMENT_ADD
870 && !assigned_change_ids.contains(&change.change_id)
871 })
872 .map(|change| {
873 super::blob::attachment_blob_contract(change)?.context("error attachment-blob-missing")
874 })
875 .collect::<Result<Vec<_>>>()?;
876 for contract in super::blob::unique_blob_content_contracts(&contracts)? {
877 validate_server_blob_before_writer(conn, blob_dir, &contract).await?;
878 }
879 Ok(())
880}
881
882async fn validate_server_blob_before_writer(
883 conn: &mut SqliteConnection,
884 blob_dir: &Path,
885 contract: &super::wire::BlobUploadContract,
886) -> Result<()> {
887 let Some(row) = crate::attachments::storage::blob_inventory_row(conn, &contract.sha256).await?
888 else {
889 bail!("error attachment-blob-missing");
890 };
891 validate_server_blob_inventory(&row, contract)?;
892 let path = crate::attachments::object_path(blob_dir, &contract.sha256)?;
893 let bytes = match tokio::fs::read(path).await {
894 Ok(bytes) => bytes,
895 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
896 bail!("error attachment-blob-missing")
897 }
898 Err(error) => return Err(error.into()),
899 };
900 if i64::try_from(bytes.len()).ok() != Some(contract.byte_size)
901 || crate::attachments::storage::sha256_hex(&bytes) != contract.sha256
902 {
903 bail!("error attachment-blob-content-mismatch");
904 }
905 let validated =
906 crate::attachments::decode::validate_image(bytes, Some(contract.media_type.clone()))
907 .await?;
908 if (validated.facts.width, validated.facts.height) != (contract.width, contract.height) {
909 bail!("error blob-inventory-metadata-mismatch");
910 }
911 Ok(())
912}
913
914fn validate_server_blob_inventory(
915 row: &crate::types::BlobInventoryRow,
916 contract: &super::wire::BlobUploadContract,
917) -> Result<()> {
918 if !row.available {
919 bail!("error attachment-blob-missing");
920 }
921 if row.byte_size != contract.byte_size || row.media_type != contract.media_type {
922 bail!("error blob-inventory-metadata-mismatch");
923 }
924 Ok(())
925}
926
927async fn ensure_attachment_blobs_admitted(
928 conn: &mut SqliteConnection,
929 blob_dir: &Path,
930 changes: &[ChangeWire],
931) -> Result<()> {
932 let contracts = super::blob::attachment_blob_contracts(changes)?;
933 for contract in super::blob::unique_blob_content_contracts(&contracts)? {
934 let Some(row) =
935 crate::attachments::storage::blob_inventory_row(conn, &contract.sha256).await?
936 else {
937 bail!("error attachment-blob-missing");
938 };
939 validate_server_blob_inventory(&row, &contract)?;
940 if !crate::attachments::object_path(blob_dir, &contract.sha256)?.exists() {
941 bail!("error attachment-blob-missing");
942 }
943 }
944 for contract in super::blob::unique_blob_admission_contracts(&contracts) {
945 let admitted: bool = sqlx::query_scalar(
946 "SELECT EXISTS(
947 SELECT 1 FROM server_blob_references sbr
948 LEFT JOIN server_task_tombstones st
949 ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
950 WHERE sbr.workspace_id = ? AND sbr.sha256 = ? AND sbr.deleted = 0
951 AND COALESCE(st.deleted, 0) = 0
952 ) OR EXISTS(
953 SELECT 1 FROM blob_upload_reservations
954 WHERE workspace_id = ? AND sha256 = ? AND byte_size = ? AND expires_at > ?
955 )",
956 )
957 .bind(&contract.workspace_id)
958 .bind(&contract.sha256)
959 .bind(&contract.workspace_id)
960 .bind(&contract.sha256)
961 .bind(contract.byte_size)
962 .bind(crate::ids::now())
963 .fetch_one(&mut *conn)
964 .await?;
965 if !admitted {
966 bail!("error attachment-blob-unreserved");
967 }
968 }
969 Ok(())
970}
971
972async fn load_server_changes_after(
973 conn: &mut SqliteConnection,
974 after: i64,
975 pull_limit: u32,
976) -> Result<(Vec<ChangeWire>, bool)> {
977 let fetch_limit = i64::from(pull_limit) + 1;
978 let rows = sqlx::query_as!(
979 ChangeRow,
980 r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
981 local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
982 entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
983 payload AS "payload!: String", base_version, created_at AS "created_at!: String",
984 server_seq
985 FROM changes WHERE server_seq > ? ORDER BY server_seq LIMIT ?"#,
986 after,
987 fetch_limit,
988 )
989 .fetch_all(&mut *conn)
990 .await?;
991 let has_more = rows.len() > pull_limit as usize;
992 let changes = rows
993 .into_iter()
994 .take(pull_limit as usize)
995 .map(ChangeRow::into_wire)
996 .collect();
997 Ok((changes, has_more))
998}
999
1000#[cfg(test)]
1001mod tests {
1002 use std::io::Cursor;
1003 use std::time::Duration;
1004
1005 use image::{DynamicImage, ImageFormat};
1006 use serde_json::json;
1007
1008 use super::*;
1009 use crate::attachments::storage::{object_path, sha256_hex, upsert_inventory_available};
1010 use crate::sync::wire::{ChangeWire, SYNC_PROTOCOL_VERSION, SyncRequest};
1011
1012 fn budget_request() -> SyncRequest {
1013 SyncRequest {
1014 protocol_version: Some(SYNC_PROTOCOL_VERSION),
1015 client_id: "budget-client\"\n".to_string(),
1016 after: i64::MAX,
1017 pull_limit: Some(super::super::wire::MAX_PULL_BATCH),
1018 changes: (0..3)
1019 .map(|index| ChangeWire {
1020 change_id: format!("AAAAAAAAAAAAAAA{index}"),
1021 client_id: "budget-client".to_string(),
1022 local_seq: index + 1,
1023 entity_type: "task".to_string(),
1024 entity_id: "BBBBBBBBBBBBBBB0".to_string(),
1025 field: Some("description".to_string()),
1026 op_type: op_type::SET_FIELD.to_string(),
1027 payload: json!({
1028 "workspace_id": "0000000000000000",
1029 "workspace_key": "default",
1030 "value": "quoted \"text\"\n雪".repeat(8),
1031 }),
1032 base_version: None,
1033 created_at: "2026-01-01T00:00:00Z".to_string(),
1034 server_seq: None,
1035 })
1036 .collect(),
1037 }
1038 }
1039
1040 #[test]
1041 fn push_byte_budget_selects_exact_ordered_prefix() {
1042 let request = budget_request();
1043 for change in &request.changes {
1044 super::super::wire::validate_pushed_change(change).unwrap();
1045 }
1046 let mut prefix = request.clone();
1047 prefix.changes.truncate(2);
1048 let limit = serde_json::to_vec(&prefix).unwrap().len();
1049 let selected = bound_push_request(request.clone(), limit).unwrap();
1050 assert_eq!(
1051 serde_json::to_vec(&selected).unwrap(),
1052 serde_json::to_vec(&prefix).unwrap()
1053 );
1054 let selected = bound_push_request(request, limit - 1).unwrap();
1055 assert_eq!(selected.changes.len(), 1);
1056 assert_eq!(selected.changes[0].local_seq, 1);
1057 assert!(serde_json::to_vec(&selected).unwrap().len() < limit);
1058 }
1059
1060 #[test]
1061 fn push_byte_budget_rejects_unfit_first_change_and_envelope() {
1062 let mut request = budget_request();
1063 request.changes.truncate(1);
1064 let limit = serde_json::to_vec(&request).unwrap().len();
1065 assert_eq!(
1066 bound_push_request(request.clone(), limit)
1067 .unwrap()
1068 .changes
1069 .len(),
1070 1
1071 );
1072 assert!(
1073 bound_push_request(request.clone(), limit - 1)
1074 .unwrap_err()
1075 .to_string()
1076 .contains("sync-change-exceeds-request-budget")
1077 );
1078 request.changes.clear();
1079 let limit = serde_json::to_vec(&request).unwrap().len();
1080 assert!(
1081 bound_push_request(request.clone(), limit)
1082 .unwrap()
1083 .changes
1084 .is_empty()
1085 );
1086 assert!(
1087 bound_push_request(request, limit - 1)
1088 .unwrap_err()
1089 .to_string()
1090 .contains("sync-request-envelope-too-large")
1091 );
1092 }
1093
1094 #[test]
1095 fn push_byte_budget_does_not_skip_a_change_that_does_not_fit() {
1096 let mut request = budget_request();
1097 let mut prefix = request.clone();
1098 prefix.changes.truncate(2);
1099 let limit = serde_json::to_vec(&prefix).unwrap().len();
1100 request.changes[1].payload["value"] = json!("middle".repeat(limit));
1101 let selected = bound_push_request(request, limit).unwrap();
1102 assert_eq!(selected.changes.len(), 1);
1103 assert_eq!(selected.changes[0].local_seq, 1);
1104 }
1105
1106 #[test]
1107 fn push_byte_budget_preserves_count_bound() {
1108 let mut request = budget_request();
1109 request.changes = vec![request.changes[0].clone(); MAX_PUSH_BATCH + 1];
1110 let selected = bound_push_request(request, MAX_SYNC_REQUEST_BYTES).unwrap();
1111 assert_eq!(selected.changes.len(), MAX_PUSH_BATCH);
1112 }
1113
1114 #[tokio::test]
1115 async fn server_task_deletion_operations_reconcile_attachment_liveness() {
1116 let (_temp, mut conn) = crate::test_support::test_conn().await;
1117 for (index, operation) in [op_type::SET_FIELD, op_type::RESOLVE_FIELD]
1118 .into_iter()
1119 .enumerate()
1120 {
1121 let task_id = format!("BBBBBBBBBBBBBBB{index}");
1122 let attachment_id = format!("CCCCCCCCCCCCCCC{index}");
1123 let sha256 = format!("{index:064x}");
1124 upsert_inventory_available(&mut conn, &sha256, 1, "image/png")
1125 .await
1126 .unwrap();
1127 sqlx::query("INSERT INTO blob_lifecycle(sha256, unreferenced_at) VALUES (?, NULL)")
1128 .bind(&sha256)
1129 .execute(&mut *conn)
1130 .await
1131 .unwrap();
1132 sqlx::query(
1133 "INSERT INTO server_blob_references(
1134 workspace_id, attachment_id, task_id, sha256, byte_size, deleted
1135 ) VALUES ('0000000000000000', ?, ?, ?, 1, 0)",
1136 )
1137 .bind(attachment_id)
1138 .bind(&task_id)
1139 .bind(&sha256)
1140 .execute(&mut *conn)
1141 .await
1142 .unwrap();
1143
1144 let deletion_change = |value: &str| ChangeWire {
1145 change_id: format!("AAAAAAAAAAAAAA{index}{value}"),
1146 client_id: "client".to_string(),
1147 local_seq: 1,
1148 entity_type: "task".to_string(),
1149 entity_id: task_id.clone(),
1150 field: Some("deleted".to_string()),
1151 op_type: operation.to_string(),
1152 payload: json!({
1153 "workspace_id": "0000000000000000",
1154 "workspace_key": "default",
1155 "value": value,
1156 }),
1157 base_version: None,
1158 created_at: "2026-01-01T00:00:00Z".to_string(),
1159 server_seq: None,
1160 };
1161
1162 let mut affected_hashes = HashSet::new();
1163 apply_server_blob_reference(&mut conn, &deletion_change("1"), &mut affected_hashes)
1164 .await
1165 .unwrap();
1166 let affected_hashes = affected_hashes.into_iter().collect::<Vec<_>>();
1167 crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
1168 &mut conn,
1169 &affected_hashes,
1170 &crate::attachments::lifecycle::SystemClock,
1171 )
1172 .await
1173 .unwrap();
1174 let deleted: bool = sqlx::query_scalar(
1175 "SELECT deleted FROM server_task_tombstones
1176 WHERE workspace_id = '0000000000000000' AND task_id = ?",
1177 )
1178 .bind(&task_id)
1179 .fetch_one(&mut *conn)
1180 .await
1181 .unwrap();
1182 let unreferenced_at: Option<String> =
1183 sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
1184 .bind(&sha256)
1185 .fetch_one(&mut *conn)
1186 .await
1187 .unwrap();
1188 assert!(deleted, "{operation} must apply live-to-deleted state");
1189 assert!(unreferenced_at.is_some());
1190
1191 let mut affected_hashes = HashSet::new();
1192 apply_server_blob_reference(&mut conn, &deletion_change("0"), &mut affected_hashes)
1193 .await
1194 .unwrap();
1195 let affected_hashes = affected_hashes.into_iter().collect::<Vec<_>>();
1196 crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
1197 &mut conn,
1198 &affected_hashes,
1199 &crate::attachments::lifecycle::SystemClock,
1200 )
1201 .await
1202 .unwrap();
1203 let deleted: bool = sqlx::query_scalar(
1204 "SELECT deleted FROM server_task_tombstones
1205 WHERE workspace_id = '0000000000000000' AND task_id = ?",
1206 )
1207 .bind(&task_id)
1208 .fetch_one(&mut *conn)
1209 .await
1210 .unwrap();
1211 let unreferenced_at: Option<String> =
1212 sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
1213 .bind(&sha256)
1214 .fetch_one(&mut *conn)
1215 .await
1216 .unwrap();
1217 assert!(!deleted, "{operation} must apply deleted-to-live state");
1218 assert_eq!(unreferenced_at, None);
1219 }
1220 }
1221
1222 #[tokio::test]
1223 async fn related_comparison_observes_push_acknowledgement_first() {
1224 let (_temp, mut conn) = crate::test_support::test_conn().await;
1225 let workspace = crate::workspaces::Workspace::default();
1226 let project = crate::projects::create_project(&mut conn, &workspace, "related-ack")
1227 .await
1228 .unwrap();
1229 let task_id: crate::ids::TaskId = "AAAA000000000001".parse().unwrap();
1230 let related_task_id: crate::ids::TaskId = "BBBB000000000002".parse().unwrap();
1231 for id in [&task_id, &related_task_id] {
1232 sqlx::query(
1233 "INSERT INTO tasks(id, workspace_id, title, description, project_id, status, priority, created_at, updated_at)
1234 VALUES (?, ?, 'task', '', ?, 'todo', 'none', 't', 't')",
1235 )
1236 .bind(id)
1237 .bind(&workspace.id)
1238 .bind(&project.id)
1239 .execute(&mut *conn)
1240 .await
1241 .unwrap();
1242 }
1243 let local = crate::operations::set_task_related_link_in_transaction(
1244 &mut conn,
1245 &workspace,
1246 &task_id,
1247 &related_task_id,
1248 true,
1249 )
1250 .await
1251 .unwrap();
1252 let local_change_id = local.change_id.unwrap();
1253 set_meta(&mut conn, "sync_cursor", "3").await.unwrap();
1254
1255 let remote = ChangeWire {
1256 change_id: "CCCC000000000003".to_string(),
1257 client_id: "remote".to_string(),
1258 local_seq: 1,
1259 entity_type: "task".to_string(),
1260 entity_id: task_id.to_string(),
1261 field: Some("related".to_string()),
1262 op_type: op_type::RELATED_REMOVE.to_string(),
1263 payload: json!({
1264 "workspace_id": workspace.id,
1265 "workspace_key": workspace.key,
1266 "related_task_id": related_task_id,
1267 }),
1268 base_version: None,
1269 created_at: "2026-08-22T00:00:00Z".to_string(),
1270 server_seq: Some(4),
1271 };
1272 let page = ApplySyncPage {
1273 request: SyncRequest {
1274 protocol_version: Some(SYNC_PROTOCOL_VERSION),
1275 client_id: "local".to_string(),
1276 after: 3,
1277 pull_limit: Some(100),
1278 changes: Vec::new(),
1279 },
1280 response: SyncResponse {
1281 protocol_version: SYNC_PROTOCOL_VERSION,
1282 cursor: 4,
1283 has_more: false,
1284 push_acks: vec![PushAck {
1285 change_id: local_change_id.clone(),
1286 server_seq: 5,
1287 }],
1288 changes: vec![remote],
1289 },
1290 attempted_at: "2026-08-22T00:00:01Z".to_string(),
1291 previous_pushed: 0,
1292 previous_pulled: 0,
1293 };
1294
1295 apply_sync_response(&mut conn, page).await.unwrap();
1296
1297 let state: (i64, String) = sqlx::query_as(
1298 "SELECT linked, last_change_id FROM task_related_links
1299 WHERE workspace_id = ? AND task_a_id = ? AND task_b_id = ?",
1300 )
1301 .bind(&workspace.id)
1302 .bind(&task_id)
1303 .bind(&related_task_id)
1304 .fetch_one(&mut *conn)
1305 .await
1306 .unwrap();
1307 assert_eq!(state, (1, local_change_id.clone()));
1308 let acknowledged: Option<i64> =
1309 sqlx::query_scalar("SELECT server_seq FROM changes WHERE change_id = ?")
1310 .bind(local_change_id)
1311 .fetch_one(&mut *conn)
1312 .await
1313 .unwrap();
1314 assert_eq!(acknowledged, Some(5));
1315 }
1316
1317 async fn corrupt_attachment_page() -> (
1318 tempfile::TempDir,
1319 Database,
1320 std::path::PathBuf,
1321 ServerSyncPage,
1322 ) {
1323 let temp = tempfile::tempdir().unwrap();
1324 let database = Database::open(&temp.path().join("server.sqlite"))
1325 .await
1326 .unwrap();
1327 let blob_dir = temp.path().join("blobs");
1328 let expected = b"expected";
1329 let sha256 = sha256_hex(expected);
1330 {
1331 let mut conn = database.acquire_writer().await.unwrap();
1332 upsert_inventory_available(&mut conn, &sha256, expected.len() as i64, "image/png")
1333 .await
1334 .unwrap();
1335 }
1336 let path = object_path(&blob_dir, &sha256).unwrap();
1337 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
1338 std::fs::write(path, b"corrupt!").unwrap();
1339 let change = ChangeWire {
1340 change_id: "0123456789ABCDEF".to_string(),
1341 client_id: "client-a".to_string(),
1342 local_seq: 1,
1343 entity_type: "task".to_string(),
1344 entity_id: "0123456789ABCDE0".to_string(),
1345 field: Some("attachments".to_string()),
1346 op_type: op_type::ATTACHMENT_ADD.to_string(),
1347 payload: json!({
1348 "workspace_id": "0000000000000000",
1349 "workspace_key": "default",
1350 "attachment_id": "7KQ9A1X4MV2P8D6R",
1351 "sha256": sha256,
1352 "byte_size": expected.len(),
1353 "media_type": "image/png",
1354 "filename": "photo.png",
1355 "alt_text": "photo",
1356 "width": 1,
1357 "height": 1,
1358 "created_at": "2026-01-01T00:00:00Z"
1359 }),
1360 base_version: None,
1361 created_at: "2026-01-01T00:00:00Z".to_string(),
1362 server_seq: None,
1363 };
1364 let page = ServerSyncPage {
1365 request: SyncRequest {
1366 protocol_version: Some(SYNC_PROTOCOL_VERSION),
1367 client_id: "client-a".to_string(),
1368 after: 0,
1369 pull_limit: Some(100),
1370 changes: vec![change],
1371 },
1372 };
1373 (temp, database, blob_dir, page)
1374 }
1375
1376 fn page_for_contract(contract: &super::super::wire::BlobUploadContract) -> ServerSyncPage {
1377 let change = ChangeWire {
1378 change_id: "0123456789ABCDEF".to_string(),
1379 client_id: "client-a".to_string(),
1380 local_seq: 1,
1381 entity_type: "task".to_string(),
1382 entity_id: "0123456789ABCDE0".to_string(),
1383 field: Some("attachments".to_string()),
1384 op_type: op_type::ATTACHMENT_ADD.to_string(),
1385 payload: json!({
1386 "workspace_id": contract.workspace_id,
1387 "workspace_key": "default",
1388 "attachment_id": "7KQ9A1X4MV2P8D6R",
1389 "sha256": contract.sha256,
1390 "byte_size": contract.byte_size,
1391 "media_type": contract.media_type,
1392 "filename": "photo.png",
1393 "alt_text": "photo",
1394 "width": contract.width,
1395 "height": contract.height,
1396 "created_at": "2026-01-01T00:00:00Z"
1397 }),
1398 base_version: None,
1399 created_at: "2026-01-01T00:00:00Z".to_string(),
1400 server_seq: None,
1401 };
1402 ServerSyncPage {
1403 request: SyncRequest {
1404 protocol_version: Some(SYNC_PROTOCOL_VERSION),
1405 client_id: "client-a".to_string(),
1406 after: 0,
1407 pull_limit: Some(100),
1408 changes: vec![change],
1409 },
1410 }
1411 }
1412
1413 #[tokio::test]
1414 async fn corrupt_server_blob_is_rejected_before_writer_acquisition() {
1415 let (_temp, database, blob_dir, page) = corrupt_attachment_page().await;
1416 let writer = database.acquire_writer().await.unwrap();
1417 let task = tokio::spawn({
1418 let database = database.clone();
1419 async move {
1420 database
1421 .persist_server_sync_page_with_blobs(page, &blob_dir)
1422 .await
1423 }
1424 });
1425 tokio::time::sleep(Duration::from_millis(50)).await;
1426 assert!(
1427 task.is_finished(),
1428 "corrupt content validation should not wait for the writer gate"
1429 );
1430 drop(writer);
1431 let error = task.await.unwrap().unwrap_err();
1432 assert!(
1433 error
1434 .to_string()
1435 .contains("attachment-blob-content-mismatch")
1436 );
1437 }
1438
1439 #[tokio::test]
1440 async fn reservation_is_rechecked_after_blob_preparation() {
1441 let temp = tempfile::tempdir().unwrap();
1442 let database = Database::open(&temp.path().join("server.sqlite"))
1443 .await
1444 .unwrap();
1445 let blob_dir = temp.path().join("blobs");
1446 let mut encoded = Cursor::new(Vec::new());
1447 DynamicImage::new_rgba8(1, 1)
1448 .write_to(&mut encoded, ImageFormat::Png)
1449 .unwrap();
1450 let bytes = encoded.into_inner();
1451 let contract = super::super::wire::BlobUploadContract {
1452 workspace_id: "0000000000000000".to_string(),
1453 sha256: sha256_hex(&bytes),
1454 byte_size: bytes.len() as i64,
1455 media_type: "image/png".to_string(),
1456 width: 1,
1457 height: 1,
1458 };
1459 database
1460 .store_server_blob(
1461 &blob_dir,
1462 crate::attachments::lifecycle::LifecyclePolicy::default(),
1463 &contract,
1464 bytes,
1465 )
1466 .await
1467 .unwrap();
1468 let page = page_for_contract(&contract);
1469 {
1470 let mut reader = database.acquire_reader().await.unwrap();
1471 prepare_server_blobs(&mut reader, &blob_dir, &page.request.changes)
1472 .await
1473 .unwrap();
1474 }
1475 {
1476 let mut writer = database.acquire_writer().await.unwrap();
1477 sqlx::query("DELETE FROM blob_upload_reservations")
1478 .execute(&mut *writer)
1479 .await
1480 .unwrap();
1481 }
1482 let mut writer = database.acquire_writer().await.unwrap();
1483 let error = assign_server_sequences(&mut writer, page.request.changes, Some(&blob_dir))
1484 .await
1485 .unwrap_err();
1486 assert!(error.to_string().contains("attachment-blob-unreserved"));
1487 let accepted: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM changes")
1488 .fetch_one(&mut *writer)
1489 .await
1490 .unwrap();
1491 assert_eq!(accepted, 0);
1492 }
1493}