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