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