Skip to main content

max / goingson

Write backups one collection at a time collect_full_export pulled all fifteen collections into a FullExport before the write began, so peak memory was the sum of every table with email bodies the largest term. Serialization already streamed into the gzip encoder; holding the source data was the remaining cost. write_backup_streaming opens the top-level map on a Serializer over the same encoder, writes the header fields, and hands an ExportSink to a callback. stream_one fetches one collection, serializes it, and drops the Vec before returning. The drop lives in the helper rather than the caller so a binding cannot keep the previous collection alive across the next query. Peak resident data is now the largest single collection. The on-disk bytes are unchanged: serialize_struct delegates to serialize_map with the same formatter, and serde_json ignores the declared map length. FullExport itself is untouched, since read_backup and the JSON export still want it. Nothing in the type system ties the two producers together, which is the risk the collect_full_export doc comment exists to name: the last time they drifted, auto-backup silently omitted five tables. stream_matches_full_export closes it by comparing key sets neither side restates. The streamed set is read back out of the gzip file the code actually produced; the reference set is serde_json::to_value over collect_full_export, so camelCase renaming and the header fields are serde's answer rather than a literal. The auto and manual backup paths stream. export_json does not: it writes pretty-printed uncompressed JSON through write_json, a different format with no streaming equivalent, so it keeps building a FullExport and is pinned by the same parity assertion. write_backup is now test-only and documented as the reference form. Both tests take the multi-thread flavor because the per-collection fetches bridge back onto the runtime with Handle::block_on from inside spawn_blocking, which is legal from a blocking-pool thread but needs a runtime with workers to drive.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-29 02:02 UTC
Signed with PGP, not checked
Commit: a1f4bedc7171593aafd8700bb0d284598fc7f33e
Parent: 8b166cb
2 files changed, +529 insertions, -45 deletions
M src-tauri/src/backup_scheduler.rs +408 -44
@@ -3,11 +3,19 @@
3 3 //! Runs in the background and creates compressed backups based on user settings.
4 4 //! Handles backup retention by pruning old backups when max count is exceeded.
5 5
6 - use crate::export::backup::{FullExport, write_backup};
6 + use crate::export::backup::{
7 + BackupError, ExportSink, FullExport, StreamedBackup, write_backup_streaming,
8 + };
7 9 use crate::state::{AppState, DESKTOP_USER_ID};
8 - use chrono::Utc;
10 + use chrono::{DateTime, Utc};
11 + use goingson_core::UserId;
12 + use serde::Serialize;
13 + use std::future::Future;
14 + use std::io::Write;
15 + use std::path::PathBuf;
9 16 use std::sync::Arc;
10 17 use tauri::Manager;
18 + use tokio::runtime::Handle;
11 19 use tokio_util::sync::CancellationToken;
12 20 use tracing::{debug, error, info, warn};
13 21
@@ -33,10 +41,15 @@
33 41
34 42 /// Collect every syncable collection into a `FullExport`.
35 43 ///
36 - /// Single source of truth for what a backup contains, so the auto-backup, manual
37 - /// backup, and JSON-export paths cannot drift in which tables they capture (the
38 - /// drift behind the ultra-fuzz finding that auto-backup silently omitted
39 - /// time_sessions, milestones, daily_notes, attachments, and sync_accounts).
44 + /// The whole-struct path, used by the JSON export, which pretty-prints a single
45 + /// document and has no streaming equivalent. The gzip backup paths use
46 + /// [`stream_full_export`] instead so they never hold every table at once.
47 + ///
48 + /// `FullExport::new` takes all fifteen collections positionally, so this
49 + /// function cannot silently omit a table; `stream_full_export` can, which is
50 + /// what `stream_matches_full_export` in the tests below guards. That drift is
51 + /// the ultra-fuzz finding where auto-backup omitted time_sessions, milestones,
52 + /// daily_notes, attachments, and sync_accounts.
40 53 pub(crate) async fn collect_full_export(
41 54 state: &AppState,
42 55 user_id: goingson_core::UserId,
@@ -80,6 +93,142 @@
80 93 ))
81 94 }
82 95
96 + /// Fetch one collection, serialize it into `sink`, and drop it before returning.
97 + ///
98 + /// The drop is the point. Binding each collection inside this call instead of in
99 + /// the caller's body means the previous one is already freed before the next
100 + /// query runs, so peak memory is the largest single collection.
101 + fn stream_one<W, T, Fut>(
102 + handle: &Handle,
103 + sink: &mut ExportSink<'_, W>,
104 + key: &'static str,
105 + fetch: Fut,
106 + ) -> Result<(), BackupError>
107 + where
108 + W: Write,
109 + T: Serialize,
110 + Fut: Future<Output = Result<Vec<T>, goingson_core::CoreError>>,
111 + {
112 + let items = handle
113 + .block_on(fetch)
114 + .map_err(|e| BackupError::Collect(Box::new(e)))?;
115 + sink.collection(key, &items)
116 + }
117 +
118 + /// Append every syncable collection to `sink`, in `FullExport` field order.
119 + ///
120 + /// The streaming counterpart of [`collect_full_export`]: same collections, same
121 + /// on-disk key names, fetched and serialized one at a time. Nothing in the type
122 + /// system ties the two together, so `stream_matches_full_export` in the tests
123 + /// below asserts that the key set this emits equals `FullExport`'s serde key
124 + /// set. Adding a field to `FullExport` without adding the matching line here
125 + /// (or the reverse) fails that test.
126 + fn stream_full_export<W: Write>(
127 + handle: &Handle,
128 + state: &AppState,
129 + user_id: UserId,
130 + sink: &mut ExportSink<'_, W>,
131 + ) -> Result<(), BackupError> {
132 + stream_one(handle, sink, "projects", state.projects.list_all(user_id))?;
133 + stream_one(handle, sink, "tasks", state.tasks.list_all(user_id))?;
134 + stream_one(handle, sink, "events", state.events.list_all(user_id))?;
135 + stream_one(handle, sink, "emails", state.emails.list_all(user_id, true))?;
136 + stream_one(handle, sink, "contacts", state.contacts.list_all(user_id))?;
137 + stream_one(
138 + handle,
139 + sink,
140 + "timeSessions",
141 + state.tasks.list_all_time_sessions(user_id),
142 + )?;
143 + stream_one(
144 + handle,
145 + sink,
146 + "milestones",
147 + state.milestones.list_all(user_id),
148 + )?;
149 + stream_one(
150 + handle,
151 + sink,
152 + "dailyNotes",
153 + state.daily_notes.list_all(user_id),
154 + )?;
155 + stream_one(
156 + handle,
157 + sink,
158 + "attachments",
159 + state.attachments.list_all(user_id),
160 + )?;
161 + stream_one(
162 + handle,
163 + sink,
164 + "syncAccounts",
165 + state.sync_accounts.list_all(user_id),
166 + )?;
167 + stream_one(
168 + handle,
169 + sink,
170 + "savedViews",
171 + state.saved_views.list_all(user_id),
172 + )?;
173 + stream_one(
174 + handle,
175 + sink,
176 + "weeklyReviews",
177 + state.weekly_reviews.list_all(user_id),
178 + )?;
179 + stream_one(
180 + handle,
181 + sink,
182 + "monthlyGoals",
183 + state.monthly_reviews.list_all_goals(user_id),
184 + )?;
185 + stream_one(
186 + handle,
187 + sink,
188 + "monthlyReflections",
189 + state.monthly_reviews.list_all_reflections(user_id),
190 + )?;
191 + // Unfiltered: a backup captures settled problems too, since dismissals and
192 + // promotions are the part that cannot be re-pulled.
193 + stream_one(
194 + handle,
195 + sink,
196 + "problems",
197 + state
198 + .problems
199 + .list(user_id, &goingson_core::ProblemFilter::default()),
200 + )?;
201 + Ok(())
202 + }
203 +
204 + /// Write a full gzip backup to `file_path`, one collection at a time.
205 + ///
206 + /// Runs on the blocking pool because JSON serialization, gzip compression, and
207 + /// directory creation all block for seconds on a large DB and would otherwise
208 + /// stall the reactor. The per-collection fetches are bridged back onto the
209 + /// runtime with `Handle::block_on`, which is legal here: a blocking-pool thread
210 + /// is not a runtime worker.
211 + pub(crate) async fn write_streaming_backup(
212 + state: &Arc<AppState>,
213 + user_id: UserId,
214 + backup_dir: PathBuf,
215 + file_path: PathBuf,
216 + exported_at: DateTime<Utc>,
217 + ) -> Result<StreamedBackup, String> {
218 + let state = Arc::clone(state);
219 + let handle = Handle::current();
220 + tokio::task::spawn_blocking(move || -> Result<StreamedBackup, String> {
221 + std::fs::create_dir_all(&backup_dir)
222 + .map_err(|e| format!("Failed to create backup directory: {e}"))?;
223 + write_backup_streaming(&file_path, exported_at, |sink| {
224 + stream_full_export(&handle, &state, user_id, sink)
225 + })
226 + .map_err(|e| format!("Failed to write backup: {e}"))
227 + })
228 + .await
229 + .map_err(|e| format!("Backup task panicked: {e}"))?
230 + }
231 +
83 232 /// Starts the background backup scheduler that creates automatic backups
84 233 /// based on user settings and prunes old backups.
85 234 pub async fn start_backup_scheduler(app: tauri::AppHandle, cancel: CancellationToken) {
@@ -171,31 +320,24 @@
171 320 let filename = backup_filename(now);
172 321 let file_path = backup_dir.join(&filename);
173 322
174 - let export = collect_full_export(state, DESKTOP_USER_ID)
175 - .await
176 - .map_err(|e| e.to_string())?;
177 - let item_count = export.total_count();
178 323 let max_to_keep = settings.max_backups_to_keep as usize;
179 324
180 - // Directory creation, gzip serialization, and pruning are all blocking and
181 - // can take seconds on a large DB, run them on the blocking pool so the
182 - // async reactor isn't stalled (email_sync uses the same pattern).
325 + // Streamed: each table is fetched, serialized into the compressor, and
326 + // dropped before the next one is read, so a big mailbox costs one collection
327 + // of resident memory rather than the whole database at once.
183 328 let log_path = file_path.clone();
184 - let size = tokio::task::spawn_blocking(move || -> Result<u64, String> {
185 - std::fs::create_dir_all(&backup_dir)
186 - .map_err(|e| format!("Failed to create backup directory: {e}"))?;
187 - let size = write_backup(&export, &file_path)
188 - .map_err(|e| format!("Failed to write backup: {e}"))?;
189 - prune_old_backups(&backup_dir, max_to_keep)?;
190 - Ok(size)
191 - })
192 - .await
193 - .map_err(|e| format!("Backup task panicked: {e}"))??;
329 + let written =
330 + write_streaming_backup(state, DESKTOP_USER_ID, backup_dir.clone(), file_path, now).await?;
331 +
332 + // Pruning is blocking directory IO, keep it off the reactor as before.
333 + tokio::task::spawn_blocking(move || prune_old_backups(&backup_dir, max_to_keep))
334 + .await
335 + .map_err(|e| format!("Backup prune task panicked: {e}"))??;
194 336
195 337 info!(
196 338 path = %log_path.display(),
197 - size_bytes = size,
198 - items = item_count,
339 + size_bytes = written.size_bytes,
340 + items = written.item_count,
199 341 "Automated backup completed"
200 342 );
201 343
@@ -262,23 +404,17 @@
262 404 let filename = backup_filename(now);
263 405 let file_path = backup_dir.join(&filename);
264 406
265 - let export = collect_full_export(state, DESKTOP_USER_ID)
266 - .await
267 - .map_err(|e| e.to_string())?;
268 - let item_count = export.total_count();
269 -
270 - // Directory creation + gzip serialization are blocking and take seconds on a
271 - // large DB; run them on the blocking pool so the manual "Backup now" action
272 - // doesn't freeze the UI (the scheduled path already does this, Perf S6).
273 - let backup_dir_task = backup_dir.clone();
274 - let file_path_task = file_path.clone();
275 - let size_bytes = tokio::task::spawn_blocking(move || -> Result<u64, String> {
276 - std::fs::create_dir_all(&backup_dir_task)
277 - .map_err(|e| format!("Failed to create backup directory: {e}"))?;
278 - write_backup(&export, &file_path_task).map_err(|e| format!("Failed to write backup: {e}"))
279 - })
280 - .await
281 - .map_err(|e| format!("Backup task panicked: {e}"))??;
407 + // Streamed and run on the blocking pool: gzip serialization takes seconds on
408 + // a large DB and would otherwise freeze the UI on "Backup now" (Perf S6),
409 + // and streaming keeps peak memory at the largest single collection.
410 + let written = write_streaming_backup(
411 + state,
412 + DESKTOP_USER_ID,
413 + backup_dir.clone(),
414 + file_path.clone(),
415 + now,
416 + )
417 + .await?;
282 418
283 419 // Update last backup timestamp
284 420 state
@@ -294,8 +430,8 @@
294 430
295 431 Ok(crate::commands::ExportResponse {
296 432 file_path: file_path.to_string_lossy().into_owned(),
297 - item_count,
298 - size_bytes,
433 + item_count: written.item_count,
434 + size_bytes: written.size_bytes,
299 435 })
300 436 }
301 437
@@ -358,6 +494,278 @@
358 494 assert_eq!(gz_count(dir.path()), 5, "0 means keep everything");
359 495 }
360 496
497 + /// Top-level object keys of a gzip backup, read back off the bytes on disk.
498 + fn keys_on_disk(path: &std::path::Path) -> std::collections::BTreeSet<String> {
499 + let file = std::fs::File::open(path).unwrap();
500 + let value: serde_json::Value =
501 + serde_json::from_reader(flate2::read::GzDecoder::new(file)).unwrap();
502 + value
503 + .as_object()
504 + .expect("backup root is a JSON object")
505 + .keys()
506 + .cloned()
507 + .collect()
508 + }
509 +
510 + /// The parity guard for the two backup shapes.
511 + ///
512 + /// `stream_full_export` writes its key names by hand, so nothing but this
513 + /// test stops it from dropping a table the way auto-backup once dropped
514 + /// time_sessions, milestones, daily_notes, attachments, and sync_accounts.
515 + /// Both key sets are derived, not restated: the streamed set comes from
516 + /// parsing the file the writer actually produced, and the reference set from
517 + /// serializing what `collect_full_export` returns, which puts serde's
518 + /// `rename_all = "camelCase"` and the `version` / `exportedAt` header on the
519 + /// reference side rather than in an assertion literal.
520 + ///
521 + /// `collect_full_export` is also the JSON-export command's producer, so
522 + /// pinning the stream to it covers that consumer too.
523 + ///
524 + /// Multi-thread flavor because `write_streaming_backup` bridges the fetches
525 + /// back onto the runtime from a blocking-pool thread.
526 + #[tokio::test(flavor = "multi_thread")]
527 + async fn stream_matches_full_export() {
528 + let (state, user_id) = crate::test_utils::setup_test_state().await;
529 + let dir = tempfile::tempdir().unwrap();
530 + let path = dir.path().join("parity.json.gz");
531 +
532 + write_streaming_backup(
533 + &state,
534 + user_id,
535 + dir.path().to_path_buf(),
536 + path.clone(),
537 + Utc::now(),
538 + )
539 + .await
540 + .expect("streaming backup writes");
541 +
542 + let streamed = keys_on_disk(&path);
543 +
544 + let reference = serde_json::to_value(collect_full_export(&state, user_id).await.unwrap())
545 + .unwrap()
546 + .as_object()
547 + .unwrap()
548 + .keys()
549 + .cloned()
550 + .collect::<std::collections::BTreeSet<String>>();
551 +
552 + assert_eq!(
553 + streamed,
554 + reference,
555 + "streaming backup and FullExport disagree about which keys a backup \
556 + contains. Missing from the stream: {:?}; extra in the stream: {:?}",
557 + reference.difference(&streamed).collect::<Vec<_>>(),
558 + streamed.difference(&reference).collect::<Vec<_>>()
559 + );
560 + }
561 +
562 + /// A populated database streamed out and read back through `read_backup`,
563 + /// with one row seeded in every collection so a dropped table shows up as a
564 + /// zero count rather than passing silently.
565 + #[tokio::test(flavor = "multi_thread")]
566 + async fn streamed_backup_round_trips_every_collection() {
567 + use chrono::NaiveDate;
568 + use goingson_core::models::{
569 + NewAttachment, NewEmail, NewMilestone, NewProblem, NewSavedView, ViewFilters, ViewType,
570 + };
571 + use goingson_core::{NewContact, NewEvent, NewTask};
572 +
573 + let (state, user_id) = crate::test_utils::setup_test_state().await;
574 + let project_id = crate::test_utils::create_test_project(&state, user_id).await;
575 +
576 + let task = state
577 + .tasks
578 + .create(
579 + user_id,
580 + NewTask::builder("Streamed task")
581 + .project_id(project_id)
582 + .build(),
583 + )
584 + .await
585 + .unwrap();
586 + state
587 + .events
588 + .create(
589 + user_id,
590 + NewEvent::builder("Streamed event", Utc::now()).build(),
591 + )
592 + .await
593 + .unwrap();
594 + state
595 + .emails
596 + .create(
597 + user_id,
598 + NewEmail {
599 + project_id: None,
600 + from_address: "a@example.com".into(),
601 + to_address: "b@example.com".into(),
602 + subject: "Streamed email".into(),
603 + body: "body".into(),
604 + is_read: false,
605 + received_at: None,
606 + },
607 + )
608 + .await
609 + .unwrap();
610 + state
611 + .contacts
612 + .create(
613 + user_id,
614 + NewContact {
615 + display_name: "Streamed contact".into(),
616 + nickname: None,
617 + company: None,
618 + title: None,
619 + notes: String::new(),
620 + tags: vec![],
621 + birthday: None,
622 + timezone: None,
623 + is_implicit: false,
624 + },
625 + )
626 + .await
627 + .unwrap();
628 + state.tasks.start_timer(task.id, user_id).await.unwrap();
629 + state
630 + .milestones
631 + .create(
632 + user_id,
633 + NewMilestone {
634 + project_id,
635 + name: "Streamed milestone".into(),
636 + description: String::new(),
637 + position: 1,
638 + target_date: None,
639 + },
640 + )
641 + .await
642 + .unwrap();
643 + state
644 + .daily_notes
645 + .upsert(
646 + user_id,
647 + NaiveDate::from_ymd_opt(2026, 6, 15).unwrap(),
648 + "went well",
649 + "could improve",
650 + true,
651 + )
652 + .await
653 + .unwrap();
654 + state
655 + .attachments
656 + .create(
657 + user_id,
658 + NewAttachment {
659 + task_id: Some(task.id),
660 + project_id: None,
661 + filename: "notes.txt".into(),
662 + file_size: 4,
663 + mime_type: "text/plain".into(),
664 + blob_hash: "0".repeat(64),
665 + source_email_id: None,
666 + },
667 + )
668 + .await
669 + .unwrap();
670 + state
671 + .sync_accounts
672 + .create(user_id, "caldav", "Streamed account", Some("s@example.com"))
673 + .await
674 + .unwrap();
675 + state
676 + .saved_views
677 + .create(
678 + user_id,
679 + NewSavedView {
680 + name: "Streamed view".into(),
681 + view_type: ViewType::Tasks,
682 + filters: ViewFilters::default(),
683 + sort_by: None,
684 + sort_order: None,
685 + is_pinned: Some(true),
686 + },
687 + )
688 + .await
689 + .unwrap();
690 + state
691 + .weekly_reviews
692 + .upsert(
693 + user_id,
694 + NaiveDate::from_ymd_opt(2026, 6, 15).unwrap(),
695 + "Good week",
696 + )
697 + .await
698 + .unwrap();
699 + state
700 + .monthly_reviews
701 + .upsert_goal(user_id, "2026-06", "Ship it", 1)
702 + .await
703 + .unwrap();
704 + state
705 + .monthly_reviews
706 + .upsert_reflection(user_id, "2026-06", "Highlight", "Change")
707 + .await
708 + .unwrap();
709 + let now = Utc::now();
710 + state
711 + .problems
712 + .ingest(
713 + user_id,
714 + NewProblem {
715 + source: "test".into(),
716 + source_ref: "p1".into(),
717 + title: "Streamed problem".into(),
718 + body: String::new(),
719 + pain: 3,
720 + scale: 3,
721 + project_id: None,
722 + tags: vec![],
723 + created_at: now,
724 + updated_at: now,
Lines truncated
M src-tauri/src/export/backup.rs +121 -1
@@ -10,6 +10,7 @@
10 10 use flate2::Compression;
11 11 use flate2::read::GzDecoder;
12 12 use flate2::write::GzEncoder;
13 + use serde::ser::{SerializeMap, Serializer as _};
13 14 use serde::{Deserialize, Serialize};
14 15
15 16 use goingson_core::{
@@ -149,7 +150,12 @@
149 150 }
150 151 }
151 152
152 - /// Writes a full export to a gzip-compressed JSON file.
153 + /// Writes an already-assembled full export to a gzip-compressed JSON file.
154 + ///
155 + /// The backup commands use [`write_backup_streaming`] instead, which produces
156 + /// the same bytes without holding every collection at once. This form stays as
157 + /// the reference for that format and for callers that already have a
158 + /// [`FullExport`] in hand.
153 159 ///
154 160 /// # Arguments
155 161 ///
@@ -178,6 +184,112 @@
178 184 Ok(metadata.len())
179 185 }
180 186
187 + /// The map serializer serde_json hands back for `&mut Serializer<W>`.
188 + ///
189 + /// Named through the trait so nothing here depends on `serde_json`'s
190 + /// `Compound` type by name.
191 + type JsonMap<'a, W> = <&'a mut serde_json::Serializer<W> as serde::Serializer>::SerializeMap;
192 +
193 + /// The open top-level object of a streaming backup.
194 + ///
195 + /// Handed to the callback of [`write_backup_streaming`], which appends one
196 + /// collection at a time. Each value is serialized straight into the compressor
197 + /// and can then be dropped, so the caller never has to hold more than one
198 + /// collection at once.
199 + pub struct ExportSink<'a, W: Write> {
200 + map: JsonMap<'a, W>,
201 + item_count: usize,
202 + }
203 +
204 + impl<W: Write> ExportSink<'_, W> {
205 + /// Serializes `items` under `key` as one entry of the export object.
206 + ///
207 + /// `key` is the on-disk (camelCase) name and must match the corresponding
208 + /// [`FullExport`] field's serde name; the parity test in `backup_scheduler`
209 + /// fails if the two key sets ever diverge.
210 + pub fn collection<T: Serialize>(
211 + &mut self,
212 + key: &'static str,
213 + items: &[T],
214 + ) -> Result<(), BackupError> {
215 + self.map.serialize_entry(key, items)?;
216 + self.item_count += items.len();
217 + Ok(())
218 + }
219 + }
220 +
221 + /// Outcome of a streaming backup write.
222 + #[derive(Debug, Clone, Copy)]
223 + pub struct StreamedBackup {
224 + /// Size of the finished compressed file in bytes.
225 + pub size_bytes: u64,
226 + /// Total number of rows written across every collection.
227 + pub item_count: usize,
228 + }
229 +
230 + /// Writes a full export to a gzip-compressed JSON file without ever holding the
231 + /// whole export in memory.
232 + ///
233 + /// Drives serde's map serializer directly: this function writes the `version`
234 + /// and `exportedAt` header entries, then calls `fill` once with an
235 + /// [`ExportSink`] that appends the collections. A caller that fetches, appends,
236 + /// and drops one collection per call makes peak memory the largest single
237 + /// collection instead of the sum of all of them, which is the whole point of
238 + /// this path over [`write_backup`].
239 + ///
240 + /// The emitted bytes are structurally identical to what [`write_backup`]
241 + /// produces for the equivalent [`FullExport`] (serde_json serializes a struct as
242 + /// a map with the same compact formatter), so the result round-trips through
243 + /// [`read_backup`]. Like [`write_backup`] it writes to a `.tmp` sibling and
244 + /// renames, so a crash mid-write cannot leave a truncated backup in place.
245 + ///
246 + /// # Arguments
247 + ///
248 + /// * `path` - Destination file path
249 + /// * `exported_at` - Timestamp recorded in the export header
250 + /// * `fill` - Appends every collection, in [`FullExport`] field order
251 + pub fn write_backup_streaming<P, F>(
252 + path: P,
253 + exported_at: DateTime<Utc>,
254 + fill: F,
255 + ) -> Result<StreamedBackup, BackupError>
256 + where
257 + P: AsRef<Path>,
258 + F: FnOnce(&mut ExportSink<'_, GzEncoder<File>>) -> Result<(), BackupError>,
259 + {
260 + let dest = path.as_ref();
261 + let tmp_path = dest.with_extension("tmp");
262 +
263 + let file = File::create(&tmp_path)?;
264 + let encoder = GzEncoder::new(file, Compression::default());
265 + let mut serializer = serde_json::Serializer::new(encoder);
266 +
267 + // serde_json ignores the declared length for maps, and the struct path
268 + // (serialize_struct) delegates to serialize_map, so the framing matches
269 + // write_backup byte for byte.
270 + let mut sink = ExportSink {
271 + map: (&mut serializer).serialize_map(None)?,
272 + item_count: 0,
273 + };
274 + sink.map
275 + .serialize_entry("version", FullExport::CURRENT_VERSION)?;
276 + sink.map.serialize_entry("exportedAt", &exported_at)?;
277 + fill(&mut sink)?;
278 + let item_count = sink.item_count;
279 + sink.map.end()?;
280 +
281 + serializer.into_inner().finish()?;
282 +
283 + // Prevents corrupt backups if the process crashes mid-write
284 + std::fs::rename(&tmp_path, dest)?;
285 +
286 + let metadata = std::fs::metadata(dest)?;
287 + Ok(StreamedBackup {
288 + size_bytes: metadata.len(),
289 + item_count,
290 + })
291 + }
292 +
181 293 /// Reads a full export from a gzip-compressed JSON file.
182 294 ///
183 295 /// # Arguments
@@ -241,6 +353,12 @@
241 353 Json(serde_json::Error),
242 354 /// Backup version is not compatible
243 355 IncompatibleVersion { found: String, expected: String },
356 + /// A collection source failed while streaming (a database read, typically).
357 + ///
358 + /// Only [`write_backup_streaming`] produces this: its `fill` callback pulls
359 + /// each collection as it goes, so a fetch failure surfaces mid-write instead
360 + /// of before it.
361 + Collect(Box<dyn std::error::Error + Send + Sync>),
244 362 }
245 363
246 364 impl std::fmt::Display for BackupError {
@@ -254,6 +372,7 @@
254 372 "Incompatible backup version: found {found}, expected {expected}"
255 373 )
256 374 }
375 + BackupError::Collect(e) => write!(f, "Failed to collect export data: {e}"),
257 376 }
258 377 }
259 378 }
@@ -263,6 +382,7 @@
263 382 match self {
264 383 BackupError::Io(e) => Some(e),
265 384 BackupError::Json(e) => Some(e),
385 + BackupError::Collect(e) => Some(&**e),
266 386 BackupError::IncompatibleVersion { .. } => None,
267 387 }
268 388 }