Skip to main content

max / goingson

perf: selective sync invalidation, poison-safe sync lock, windowed threads, IMAP batching Axis 5 of the Run #26 ultra-fuzz remediation. UF-3: sync:changes-applied now carries the DB tables a pull touched, threaded from pull_changes (PullOutcome.changed_tables) through SyncResult. The frontend invalidates only the affected cache entities (SYNC_TABLE_TO_ENTITIES), falling back to a full bust for an unknown/absent payload, so a remote task edit no longer forces the compose screen to re-hydrate every contact. A new list_contact_email_directory command (one JOIN, name+email only) replaces the full contact hydration in the compose autocomplete. UF-4 (CHRONIC-A): the sync client's RwLock is sealed inside a SyncClientCell newtype with a private field, so a raw .read().expect("poisoned") is unwritable. AppState::read_recovering() is the only accessor and recovers from poisoning. UF-9: the threaded email summary is a single ROW_NUMBER() window pass instead of a per-thread correlated subquery, binding the filter set once. UF-10: both IMAP fetch loops batch the UID set (50/command) to bound memory and command size; the duplicated per-message parse is now one build_parsed_email. UF-11: backup directory creation, gzip, and pruning run in spawn_blocking so a large-DB backup doesn't stall the reactor. Medium: email_sync_scheduler prunes failure_counts for deleted/disabled accounts each pass so the map can't grow unbounded. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-16 05:41 UTC
Commit: 731894cda774b53e985dd65e42348f480208d2bb
Parent: 735059d
20 files changed, +440 insertions, -234 deletions
M crates/core/src/contact.rs +12
@@ -77,6 +77,18 @@ impl Contact {
77 77
78 78 // ============ Sub-collection Entities ============
79 79
80 + /// A flattened (name, email) pair for compose autocomplete.
81 + ///
82 + /// Returned by `list_email_directory` — a single JOIN that skips the per-contact
83 + /// sub-collection hydration the compose screen does not need.
84 + #[derive(Debug, Clone, Serialize, Deserialize)]
85 + #[serde(rename_all = "camelCase")]
86 + pub struct ContactEmailEntry {
87 + pub name: String,
88 + pub email: String,
89 + pub is_implicit: bool,
90 + }
91 +
80 92 /// An email address belonging to a contact.
81 93 #[derive(Debug, Clone, Serialize, Deserialize)]
82 94 #[serde(rename_all = "camelCase")]
M crates/core/src/lib.rs +1 -1
@@ -53,7 +53,7 @@ pub mod validation;
53 53 pub mod weekly_review;
54 54
55 55 pub use contact::{
56 - Contact, ContactCustomField, ContactEmail, ContactPhone, NewContact, NewContactCustomField,
56 + Contact, ContactCustomField, ContactEmail, ContactEmailEntry, ContactPhone, NewContact, NewContactCustomField,
57 57 NewContactEmail, NewContactPhone, NewSocialHandle, SocialHandle, UpdateContact,
58 58 };
59 59 pub use error::CoreError;
M crates/core/src/repository.rs +5 -1
@@ -14,7 +14,7 @@ use crate::id_types::{
14 14 use uuid::Uuid;
15 15
16 16 use crate::contact::{
17 - Contact, ContactCustomField, ContactEmail, ContactPhone, NewContact, NewContactCustomField,
17 + Contact, ContactCustomField, ContactEmail, ContactEmailEntry, ContactPhone, NewContact, NewContactCustomField,
18 18 NewContactEmail, NewContactPhone, NewSocialHandle, SocialHandle, UpdateContact,
19 19 };
20 20 use crate::error::CoreError;
@@ -937,6 +937,10 @@ pub trait ContactRepository: Send + Sync {
937 937 /// Searches across display_name, nickname, company, title, notes, and email addresses.
938 938 async fn list_filtered(&self, user_id: UserId, search: Option<&str>, tag: Option<&str>, include_implicit: bool) -> Result<Vec<Contact>>;
939 939
940 + /// Flat (name, email) directory for compose autocomplete — one JOIN, no
941 + /// per-contact sub-collection hydration. One row per contact email address.
942 + async fn list_email_directory(&self, user_id: UserId, include_implicit: bool) -> Result<Vec<ContactEmailEntry>>;
943 +
940 944 /// Finds a contact by email address.
941 945 async fn find_by_email(&self, user_id: UserId, email: &str) -> Result<Option<Contact>>;
942 946
M crates/db-sqlite/src/repository/contact_repo.rs +28 -1
@@ -7,7 +7,7 @@ use async_trait::async_trait;
7 7 use sqlx::SqlitePool;
8 8 use std::collections::{HashMap, HashSet};
9 9 use goingson_core::{
10 - Contact, ContactCustomField, ContactEmail, ContactEmailId, ContactId, ContactPhone,
10 + Contact, ContactCustomField, ContactEmail, ContactEmailEntry, ContactEmailId, ContactId, ContactPhone,
11 11 ContactPhoneId, ContactRepository, CoreError, CustomFieldId, NewContact,
12 12 NewContactCustomField, NewContactEmail, NewContactPhone, NewSocialHandle, Result,
13 13 SocialHandle, SocialHandleId, UpdateContact, UserId,
@@ -584,6 +584,33 @@ impl ContactRepository for SqliteContactRepository {
584 584 }
585 585
586 586 #[tracing::instrument(skip_all)]
587 + async fn list_email_directory(&self, user_id: UserId, include_implicit: bool) -> Result<Vec<ContactEmailEntry>> {
588 + // One JOIN, one row per email address — no per-contact sub-collection
589 + // hydration (the compose autocomplete only needs name + address).
590 + let implicit_filter = if include_implicit { "" } else { "AND c.is_implicit = 0" };
591 + let sql = format!(
592 + "SELECT c.display_name, ce.address, c.is_implicit \
593 + FROM contacts c JOIN contact_emails ce ON ce.contact_id = c.id \
594 + WHERE c.user_id = ? {} \
595 + ORDER BY c.display_name ASC, ce.is_primary DESC",
596 + implicit_filter
597 + );
598 + let rows: Vec<(String, String, i64)> = sqlx::query_as(&sql)
599 + .bind(user_id.to_string())
600 + .fetch_all(&self.pool)
601 + .await
602 + .map_err(CoreError::database)?;
603 + Ok(rows
604 + .into_iter()
605 + .map(|(name, email, is_implicit)| ContactEmailEntry {
606 + name,
607 + email,
608 + is_implicit: is_implicit != 0,
609 + })
610 + .collect())
611 + }
612 +
613 + #[tracing::instrument(skip_all)]
587 614 async fn find_by_email(&self, user_id: UserId, email: &str) -> Result<Option<Contact>> {
588 615 let row = sqlx::query_as::<_, ContactRow>(
589 616 r#"
M crates/db-sqlite/src/repository/email_repo.rs +28 -16
@@ -154,31 +154,43 @@ impl EmailRepository for SqliteEmailRepository {
154 154 latest_email_id: String,
155 155 }
156 156
157 + // Rank emails within each thread by recency in a single pass with window
158 + // functions, then keep the latest row per thread. This replaces a
159 + // correlated subquery that rescanned `emails` once per thread group
160 + // (O(threads x emails)); the partition's MAX(received_at) is the rn = 1
161 + // row, and the thread-wide counts come from window aggregates.
157 162 let summary_sql = format!(
158 - r#"SELECT
159 - COALESCE(e.thread_id, e.id) AS thread_key,
160 - MAX(e.received_at) AS latest_received_at,
161 - COUNT(*) AS thread_count,
162 - SUM(CASE WHEN e.is_read = 0 THEN 1 ELSE 0 END) AS unread_count,
163 - (SELECT e2.id FROM emails e2
164 - WHERE COALESCE(e2.thread_id, e2.id) = COALESCE(e.thread_id, e.id)
165 - AND e2.user_id = ? AND e2.is_draft = 0 {} {} {}
166 - ORDER BY e2.received_at DESC LIMIT 1) AS latest_email_id
167 - FROM emails e
168 - WHERE e.user_id = ? AND e.is_draft = 0 {} {} {}
169 - GROUP BY COALESCE(e.thread_id, e.id)
163 + r#"WITH ranked AS (
164 + SELECT
165 + e.id AS email_id,
166 + COALESCE(e.thread_id, e.id) AS thread_key,
167 + e.received_at AS received_at,
168 + ROW_NUMBER() OVER (
169 + PARTITION BY COALESCE(e.thread_id, e.id)
170 + ORDER BY e.received_at DESC, e.id DESC
171 + ) AS rn,
172 + COUNT(*) OVER (PARTITION BY COALESCE(e.thread_id, e.id)) AS thread_count,
173 + SUM(CASE WHEN e.is_read = 0 THEN 1 ELSE 0 END)
174 + OVER (PARTITION BY COALESCE(e.thread_id, e.id)) AS unread_count
175 + FROM emails e
176 + WHERE e.user_id = ? AND e.is_draft = 0 {} {} {}
177 + )
178 + SELECT
179 + thread_key,
180 + received_at AS latest_received_at,
181 + thread_count,
182 + unread_count,
183 + email_id AS latest_email_id
184 + FROM ranked
185 + WHERE rn = 1
170 186 ORDER BY latest_received_at DESC
171 187 LIMIT ? OFFSET ?"#,
172 188 archived_filter, folder_filter, label_filter,
173 - archived_filter, folder_filter, label_filter,
174 189 );
175 190
176 191 let mut summary_q = sqlx::query_as::<_, ThreadSummary>(&summary_sql).bind(&uid);
177 192 if let Some(f) = folder { summary_q = summary_q.bind(f); }
178 193 if let Some(l) = label { summary_q = summary_q.bind(l); }
179 - summary_q = summary_q.bind(&uid);
180 - if let Some(f) = folder { summary_q = summary_q.bind(f); }
181 - if let Some(l) = label { summary_q = summary_q.bind(l); }
182 194 let summaries = summary_q
183 195 .bind(limit_val)
184 196 .bind(offset_val)
M crates/db-sqlite/tests/contact_repo_tests.rs +42
@@ -807,3 +807,45 @@ async fn update_missing_subcollection_returns_none() {
807 807 .unwrap();
808 808 assert!(result.is_none());
809 809 }
810 +
811 + #[tokio::test]
812 + async fn list_email_directory_flattens_addresses_and_honors_implicit() {
813 + let pool = common::setup_test_db().await;
814 + let user_id = common::create_test_user(&pool).await;
815 + let repo = SqliteContactRepository::new(pool);
816 +
817 + let mk = |name: &str, implicit: bool| NewContact {
818 + display_name: name.to_string(),
819 + nickname: None,
820 + company: None,
821 + title: None,
822 + notes: String::new(),
823 + tags: vec![],
824 + birthday: None,
825 + timezone: None,
826 + is_implicit: implicit,
827 + };
828 +
829 + let alice = repo.create(user_id, mk("Alice", false)).await.unwrap();
830 + repo.add_email(alice.id, user_id, NewContactEmail {
831 + address: "alice@work.com".to_string(), label: "work".to_string(), is_primary: true,
832 + }).await.unwrap();
833 + repo.add_email(alice.id, user_id, NewContactEmail {
834 + address: "alice@home.com".to_string(), label: "home".to_string(), is_primary: false,
835 + }).await.unwrap();
836 +
837 + let ghost = repo.create(user_id, mk("Ghost", true)).await.unwrap();
838 + repo.add_email(ghost.id, user_id, NewContactEmail {
839 + address: "ghost@example.com".to_string(), label: "other".to_string(), is_primary: true,
840 + }).await.unwrap();
841 +
842 + // Implicit excluded: only Alice's two addresses.
843 + let explicit = repo.list_email_directory(user_id, false).await.unwrap();
844 + assert_eq!(explicit.len(), 2, "two addresses for the one explicit contact");
845 + assert!(explicit.iter().all(|e| e.name == "Alice" && !e.is_implicit));
846 +
847 + // Implicit included: Alice's two + Ghost's one.
848 + let all = repo.list_email_directory(user_id, true).await.unwrap();
849 + assert_eq!(all.len(), 3);
850 + assert!(all.iter().any(|e| e.email == "ghost@example.com" && e.is_implicit));
851 + }
M crates/db-sqlite/tests/email_repo_tests.rs +45
@@ -233,6 +233,51 @@ async fn test_email_ordering_by_received_at() {
233 233 }
234 234
235 235 #[tokio::test]
236 + async fn test_list_threaded_returns_latest_per_thread() {
237 + let pool = common::setup_test_db().await;
238 + let user_id = common::create_test_user(&pool).await;
239 + let repo = SqliteEmailRepository::new(pool.clone());
240 +
241 + let mk = |subject: &str, read: bool, hours_ago: i64| NewEmail {
242 + project_id: None,
243 + from_address: "a@example.com".to_string(),
244 + to_address: "me@example.com".to_string(),
245 + subject: subject.to_string(),
246 + body: "body".to_string(),
247 + is_read: read,
248 + received_at: Some(Utc::now() - Duration::hours(hours_ago)),
249 + };
250 +
251 + // Two emails that will share a thread (older read, newer unread) + a standalone.
252 + let older = repo.create(user_id, mk("Thread root", true, 2)).await.unwrap();
253 + let newer = repo.create(user_id, mk("Thread reply", false, 1)).await.unwrap();
254 + let _standalone = repo.create(user_id, mk("Lonely", true, 0)).await.unwrap();
255 +
256 + // Group the pair into one thread keyed by the root id.
257 + sqlx::query("UPDATE emails SET thread_id = ? WHERE id IN (?, ?)")
258 + .bind(older.id.to_string())
259 + .bind(older.id.to_string())
260 + .bind(newer.id.to_string())
261 + .execute(&pool)
262 + .await
263 + .unwrap();
264 +
265 + let (threads, total) = repo
266 + .list_threaded(user_id, true, None, None, None, None)
267 + .await
268 + .expect("list_threaded");
269 +
270 + assert_eq!(total, 2, "the grouped pair counts as one thread, plus the standalone");
271 + let grouped = threads
272 + .iter()
273 + .find(|t| t.thread_count == 2)
274 + .expect("expected a 2-email thread");
275 + // The window function must surface the newest message as the thread's face.
276 + assert_eq!(grouped.most_recent_email.id, newer.id);
277 + assert!(grouped.has_unread, "thread contains an unread email");
278 + }
279 +
280 + #[tokio::test]
236 281 async fn test_count_unread() {
237 282 let pool = common::setup_test_db().await;
238 283 let user_id = common::create_test_user(&pool).await;
M src-tauri/frontend/compose.html +8 -14
@@ -172,20 +172,14 @@
172 172
173 173 async function loadContactEmails() {
174 174 try {
175 - const contacts = await invoke('list_contacts_filtered', { search: null, tag: null, includeImplicit: true });
176 - contactEmails = [];
177 - for (const c of contacts) {
178 - const name = c.displayName || c.display_name || '';
179 - const isImplicit = c.isImplicit || false;
180 - if (c.emails && c.emails.length > 0) {
181 - for (const e of c.emails) {
182 - contactEmails.push({ name, email: e.address, isImplicit });
183 - }
184 - }
185 - if (c.primaryEmail && !c.emails?.some(e => e.address === c.primaryEmail)) {
186 - contactEmails.push({ name, email: c.primaryEmail, isImplicit });
187 - }
188 - }
175 + // Lightweight directory (one JOIN, name + address only) instead of
176 + // hydrating every contact's sub-collections just for autocomplete.
177 + const entries = await invoke('list_contact_email_directory', { includeImplicit: true });
178 + contactEmails = entries.map(e => ({
179 + name: e.name || '',
180 + email: e.email,
181 + isImplicit: e.isImplicit || false,
182 + }));
189 183 } catch (_) { /* contacts unavailable */ }
190 184 }
191 185
M src-tauri/frontend/js/app.js +47 -3
@@ -6,6 +6,29 @@
6 6 (function() {
7 7 'use strict';
8 8
9 + // Maps a synced DB table (from the `sync:changes-applied` payload) to the view
10 + // cache entities it affects. A table mapping to `[]` has no cached view to bust
11 + // (the re-render alone refetches it). Any table NOT present here triggers a full
12 + // cache bust — keep this in sync with the backend `UPSERT_ORDER` list.
13 + const SYNC_TABLE_TO_ENTITIES = {
14 + projects: ['projects'],
15 + milestones: ['projects'],
16 + tasks: ['tasks'],
17 + subtasks: ['tasks'],
18 + annotations: ['tasks'],
19 + time_sessions: ['tasks'],
20 + attachments: ['tasks', 'projects'],
21 + events: ['events'],
22 + contacts: ['contacts'],
23 + contact_emails: ['contacts'],
24 + contact_phones: ['contacts'],
25 + contact_social_handles: ['contacts'],
26 + contact_custom_fields: ['contacts'],
27 + email_accounts: ['emails'],
28 + sync_accounts: [],
29 + daily_notes: [],
30 + };
31 +
9 32 // ============ Application Initialization ============
10 33
11 34 document.addEventListener('DOMContentLoaded', async () => {
@@ -148,9 +171,30 @@ if (window.__TAURI__) {
148 171 refreshCurrentViewData();
149 172 });
150 173
151 - // Cloud sync: remote changes applied
152 - listen('sync:changes-applied', () => {
153 - GoingsOn.cache.invalidateAll();
174 + // Cloud sync: remote changes applied. The payload is the list of DB tables
175 + // the pull touched; invalidate only the affected cache entities so an
176 + // unrelated change (e.g. a task edit) doesn't force the compose screen to
177 + // re-hydrate every contact. Unknown/absent payload falls back to a full bust.
178 + listen('sync:changes-applied', (event) => {
179 + const tables = Array.isArray(event.payload) ? event.payload : null;
180 + if (!tables) {
181 + GoingsOn.cache.invalidateAll();
182 + } else {
183 + const entities = new Set();
184 + let unknown = false;
185 + for (const table of tables) {
186 + if (table in SYNC_TABLE_TO_ENTITIES) {
187 + for (const e of SYNC_TABLE_TO_ENTITIES[table]) entities.add(e);
188 + } else {
189 + unknown = true; // be safe about tables we don't have a mapping for
190 + }
191 + }
192 + if (unknown) {
193 + GoingsOn.cache.invalidateAll();
194 + } else if (entities.size) {
195 + GoingsOn.cache.invalidate(...entities);
196 + }
197 + }
154 198 refreshCurrentViewData();
155 199 });
156 200
M src-tauri/src/backup_scheduler.rs +19 -9
@@ -152,19 +152,32 @@ async fn check_and_backup(app: &tauri::AppHandle, state: &Arc<AppState>) -> Resu
152 152 .map_err(|e| format!("Failed to get app data dir: {}", e))?
153 153 .join("backups");
154 154
155 - std::fs::create_dir_all(&backup_dir)
156 - .map_err(|e| format!("Failed to create backup directory: {}", e))?;
157 -
158 155 let filename = backup_filename(now);
159 156 let file_path = backup_dir.join(&filename);
160 157
161 158 let export = collect_full_export(state).await.map_err(|e| e.to_string())?;
162 - let size = write_backup(&export, &file_path).map_err(|e| format!("Failed to write backup: {}", e))?;
159 + let item_count = export.total_count();
160 + let max_to_keep = settings.max_backups_to_keep as usize;
161 +
162 + // Directory creation, gzip serialization, and pruning are all blocking and
163 + // can take seconds on a large DB — run them on the blocking pool so the
164 + // async reactor isn't stalled (email_sync uses the same pattern).
165 + let log_path = file_path.clone();
166 + let size = tokio::task::spawn_blocking(move || -> Result<u64, String> {
167 + std::fs::create_dir_all(&backup_dir)
168 + .map_err(|e| format!("Failed to create backup directory: {}", e))?;
169 + let size = write_backup(&export, &file_path)
170 + .map_err(|e| format!("Failed to write backup: {}", e))?;
171 + prune_old_backups(&backup_dir, max_to_keep)?;
172 + Ok(size)
173 + })
174 + .await
175 + .map_err(|e| format!("Backup task panicked: {}", e))??;
163 176
164 177 info!(
165 - path = %file_path.display(),
178 + path = %log_path.display(),
166 179 size_bytes = size,
167 - items = export.total_count(),
180 + items = item_count,
168 181 "Automated backup completed"
169 182 );
170 183
@@ -175,9 +188,6 @@ async fn check_and_backup(app: &tauri::AppHandle, state: &Arc<AppState>) -> Resu
175 188 .await
176 189 .map_err(|e| format!("Failed to update last backup time: {}", e))?;
177 190
178 - // Prune old backups
179 - prune_old_backups(&backup_dir, settings.max_backups_to_keep as usize)?;
180 -
181 191 Ok(())
182 192 }
183 193
M src-tauri/src/commands/contact.rs +15
@@ -402,6 +402,21 @@ pub async fn list_contacts_filtered(
402 402 Ok(contacts.into_iter().map(ContactResponse::from).collect())
403 403 }
404 404
405 + /// Lightweight name+email directory for compose autocomplete.
406 + ///
407 + /// Avoids the full contact hydration (4 sub-queries per contact) that
408 + /// `list_contacts_filtered` performs; returns one entry per email address.
409 + #[tauri::command]
410 + #[instrument(skip_all)]
411 + pub async fn list_contact_email_directory(
412 + state: State<'_, Arc<AppState>>,
413 + include_implicit: Option<bool>,
414 + ) -> Result<Vec<goingson_core::ContactEmailEntry>, ApiError> {
415 + Ok(state.contacts
416 + .list_email_directory(DESKTOP_USER_ID, include_implicit.unwrap_or(false))
417 + .await?)
418 + }
419 +
405 420 /// Finds a contact by email address (case-insensitive).
406 421 #[tauri::command]
407 422 #[instrument(skip_all)]
M src-tauri/src/commands/sync.rs +3 -2
@@ -72,7 +72,7 @@ pub struct SyncSettingsInput {
72 72
73 73 /// Extract the sync client from state. Clones the Arc for use across await points.
74 74 fn get_sync_client(state: &AppState) -> Option<std::sync::Arc<synckit_client::SyncKitClient>> {
75 - state.sync_client.read().unwrap_or_else(|e| e.into_inner()).clone()
75 + state.read_recovering()
76 76 }
77 77
78 78 fn require_sync_client(state: &AppState) -> Result<std::sync::Arc<synckit_client::SyncKitClient>, ApiError> {
@@ -290,7 +290,8 @@ pub async fn sync_now(
290 290 };
291 291
292 292 if result.pulled > 0 {
293 - let _ = app.emit("sync:changes-applied", ());
293 + // Carry the changed tables so the UI invalidates selectively.
294 + let _ = app.emit("sync:changes-applied", &result.pulled_tables);
294 295 }
295 296
296 297 // Cleanup after manual sync too
M src-tauri/src/email/imap_client.rs +119 -178
@@ -14,6 +14,11 @@ type ImapSession = async_imap::Session<tokio_native_tls::TlsStream<TcpStream>>;
14 14 /// Maximum email size to download (25 MB). Emails exceeding this are skipped during sync.
15 15 const MAX_EMAIL_SIZE: u32 = 25 * 1024 * 1024;
16 16
17 + /// How many UIDs to body-fetch per IMAP command. Bounds the command length and
18 + /// the in-flight fetch pipeline so a large mailbox doesn't issue one enormous
19 + /// fetch.
20 + const IMAP_FETCH_BATCH: usize = 50;
21 +
17 22 /// Raw attachment data extracted during IMAP RFC822 parse.
18 23 #[derive(Debug, Clone)]
19 24 pub struct AttachmentPart {
@@ -227,121 +232,55 @@ impl ImapClient {
227 232 return Ok((Vec::new(), debug.join(", ")));
228 233 }
229 234
230 - let safe_uid_set = safe_seqs.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(",");
231 -
232 - // Fetch full bodies only for safe-sized emails
233 - let mut messages = session
234 - .uid_fetch(&safe_uid_set, "(UID FLAGS RFC822)")
235 - .await
236 - .map_err(|e| format!("Fetch error: {}", e))?;
237 -
238 235 let mut emails = Vec::new();
239 236 let folder_name = folder.to_string();
240 237 let mut msg_count = 0;
241 238 let mut body_count = 0;
242 239 let mut parse_errors = 0;
243 240
244 - while let Some(result) = messages.next().await {
245 - msg_count += 1;
246 - let message = match result {
247 - Ok(m) => m,
248 - Err(e) => {
249 - debug.push(format!("msg_err: {}", e));
250 - continue;
251 - }
252 - };
253 - let uid = match message.uid {
254 - Some(u) => u,
255 - None => {
256 - tracing::warn!(folder = %folder_name, "IMAP message missing UID, skipping");
257 - continue;
258 - }
259 - };
260 -
261 - // Check if \Seen flag is present (email has been read)
262 - let is_read = message.flags().any(|f| {
263 - matches!(f, async_imap::types::Flag::Seen)
264 - });
241 + // Fetch full bodies in bounded UID batches so neither the IMAP command
242 + // nor the in-flight fetch pipeline grows with the size of the mailbox.
243 + for chunk in safe_seqs.chunks(IMAP_FETCH_BATCH) {
244 + let safe_uid_set = chunk.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(",");
245 + let mut messages = session
246 + .uid_fetch(&safe_uid_set, "(UID FLAGS RFC822)")
247 + .await
248 + .map_err(|e| format!("Fetch error: {}", e))?;
265 249
266 - if let Some(body) = message.body() {
267 - body_count += 1;
268 - match mailparse::parse_mail(body) {
269 - Ok(parsed) => {
270 - let message_id = parsed
271 - .headers
272 - .iter()
273 - .find(|h| h.get_key().to_lowercase() == "message-id")
274 - .map(|h| h.get_value());
275 -
276 - let in_reply_to = parsed
277 - .headers
278 - .iter()
279 - .find(|h| h.get_key().to_lowercase() == "in-reply-to")
280 - .map(|h| h.get_value());
281 -
282 - let references_root = extract_references_root(&parsed.headers);
283 -
284 - let from = parsed
285 - .headers
286 - .iter()
287 - .find(|h| h.get_key().to_lowercase() == "from")
288 - .map(|h| h.get_value())
289 - .unwrap_or_default();
290 -
291 - let to = parsed
292 - .headers
293 - .iter()
294 - .find(|h| h.get_key().to_lowercase() == "to")
295 - .map(|h| h.get_value())
296 - .unwrap_or_default();
297 -
298 - let subject = parsed
299 - .headers
300 - .iter()
301 - .find(|h| h.get_key().to_lowercase() == "subject")
302 - .map(|h| h.get_value())
303 - .unwrap_or_default();
304 -
305 - let date_str = parsed
306 - .headers
307 - .iter()
308 - .find(|h| h.get_key().to_lowercase() == "date")
309 - .map(|h| h.get_value());
310 -
311 - let date = date_str
312 - .and_then(|s| mailparse::dateparse(&s).ok())
313 - .and_then(|ts| Utc.timestamp_opt(ts, 0).single())
314 - .unwrap_or(DateTime::UNIX_EPOCH);
315 -
316 - let (body_text, html_body) = Self::extract_body_with_html(&parsed);
317 - let attachments = extract_attachment_parts(&parsed);
318 -
319 - emails.push(ParsedEmail {
320 - message_id,
321 - in_reply_to,
322 - references_root,
323 - imap_uid: uid,
324 - source_folder: folder_name.clone(),
325 - from,
326 - to,
327 - subject,
328 - body: body_text,
329 - html_body,
330 - date,
331 - is_read,
332 - attachments,
333 - });
334 - }
250 + while let Some(result) = messages.next().await {
251 + msg_count += 1;
252 + let message = match result {
253 + Ok(m) => m,
335 254 Err(e) => {
336 - tracing::debug!(uid, folder = %folder_name, error = %e, "Failed to parse email");
337 - parse_errors += 1;
255 + debug.push(format!("msg_err: {}", e));
256 + continue;
257 + }
258 + };
259 + let uid = match message.uid {
260 + Some(u) => u,
261 + None => {
262 + tracing::warn!(folder = %folder_name, "IMAP message missing UID, skipping");
263 + continue;
264 + }
265 + };
266 +
267 + // Check if \Seen flag is present (email has been read)
268 + let is_read = message.flags().any(|f| matches!(f, async_imap::types::Flag::Seen));
269 +
270 + if let Some(body) = message.body() {
271 + body_count += 1;
272 + match mailparse::parse_mail(body) {
273 + Ok(parsed) => emails.push(Self::build_parsed_email(&parsed, uid, &folder_name, is_read)),
274 + Err(e) => {
275 + tracing::debug!(uid, folder = %folder_name, error = %e, "Failed to parse email");
276 + parse_errors += 1;
277 + }
338 278 }
339 279 }
340 280 }
341 281 }
342 282
343 283 debug.push(format!("msgs: {}, bodies: {}, parsed: {}, errs: {}", msg_count, body_count, emails.len(), parse_errors));
344 - drop(messages);
345 284 session.logout().await.ok();
346 285 Ok((emails, debug.join(", ")))
347 286 }
@@ -473,13 +412,6 @@ impl ImapClient {
473 412 });
474 413 }
475 414
476 - let safe_uid_set = safe_uids.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(",");
477 -
478 - let mut messages = session
479 - .uid_fetch(&safe_uid_set, "(UID FLAGS RFC822)")
480 - .await
481 - .map_err(|e| format!("UID fetch error: {}", e))?;
482 -
483 415 let mut emails = Vec::new();
484 416 let folder_name = folder.to_string();
485 417 let mut msg_count = 0;
@@ -488,84 +420,50 @@ impl ImapClient {
488 420 // Seed max_uid from all UIDs (including skipped large ones) so they aren't re-fetched
489 421 let mut max_uid: Option<u32> = uids.iter().copied().max();
490 422
491 - while let Some(result) = messages.next().await {
492 - msg_count += 1;
493 - let message = match result {
494 - Ok(m) => m,
495 - Err(e) => {
496 - debug.push(format!("msg_err: {}", e));
497 - continue;
498 - }
499 - };
500 - let uid = match message.uid {
501 - Some(u) => u,
502 - None => {
503 - tracing::warn!(folder = %folder_name, "IMAP message missing UID, skipping");
504 - continue;
505 - }
506 - };
423 + // Fetch full bodies in bounded UID batches so neither the IMAP command
424 + // nor the in-flight fetch pipeline grows with the size of the mailbox.
425 + for chunk in safe_uids.chunks(IMAP_FETCH_BATCH) {
426 + let safe_uid_set = chunk.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(",");
427 + let mut messages = session
428 + .uid_fetch(&safe_uid_set, "(UID FLAGS RFC822)")
429 + .await
430 + .map_err(|e| format!("UID fetch error: {}", e))?;
507 431
508 - max_uid = Some(max_uid.map_or(uid, |m: u32| m.max(uid)));
509 -
510 - let is_read = message.flags().any(|f| matches!(f, async_imap::types::Flag::Seen));
511 -
512 - if let Some(body) = message.body() {
513 - body_count += 1;
514 - match mailparse::parse_mail(body) {
515 - Ok(parsed) => {
516 - let message_id = parsed.headers.iter()
517 - .find(|h| h.get_key().to_lowercase() == "message-id")
518 - .map(|h| h.get_value());
519 - let in_reply_to = parsed.headers.iter()
520 - .find(|h| h.get_key().to_lowercase() == "in-reply-to")
521 - .map(|h| h.get_value());
522 - let references_root = extract_references_root(&parsed.headers);
523 - let from = parsed.headers.iter()
524 - .find(|h| h.get_key().to_lowercase() == "from")
525 - .map(|h| h.get_value()).unwrap_or_default();
526 - let to = parsed.headers.iter()
527 - .find(|h| h.get_key().to_lowercase() == "to")
528 - .map(|h| h.get_value()).unwrap_or_default();
529 - let subject = parsed.headers.iter()
530 - .find(|h| h.get_key().to_lowercase() == "subject")
531 - .map(|h| h.get_value()).unwrap_or_default();
532 - let date_str = parsed.headers.iter()
533 - .find(|h| h.get_key().to_lowercase() == "date")
534 - .map(|h| h.get_value());
535 - let date = date_str
536 - .and_then(|s| mailparse::dateparse(&s).ok())
537 - .and_then(|ts| Utc.timestamp_opt(ts, 0).single())
538 - .unwrap_or(DateTime::UNIX_EPOCH);
539 -
540 - let (body_text, html_body) = Self::extract_body_with_html(&parsed);
541 - let attachments = extract_attachment_parts(&parsed);
542 -
543 - emails.push(ParsedEmail {
544 - message_id,
545 - in_reply_to,
546 - references_root,
547 - imap_uid: uid,
548 - source_folder: folder_name.clone(),
549 - from,
550 - to,
551 - subject,
552 - body: body_text,
553 - html_body,
554 - date,
555 - is_read,
556 - attachments,
557 - });
558 - }
432 + while let Some(result) = messages.next().await {
433 + msg_count += 1;
434 + let message = match result {
435 + Ok(m) => m,
559 436 Err(e) => {
560 - tracing::debug!(uid, folder = %folder_name, error = %e, "Failed to parse email");
561 - parse_errors += 1;
437 + debug.push(format!("msg_err: {}", e));
438 + continue;
439 + }
440 + };
441 + let uid = match message.uid {
442 + Some(u) => u,
443 + None => {
444 + tracing::warn!(folder = %folder_name, "IMAP message missing UID, skipping");
445 + continue;
446 + }
447 + };
448 +
449 + max_uid = Some(max_uid.map_or(uid, |m: u32| m.max(uid)));
450 +
451 + let is_read = message.flags().any(|f| matches!(f, async_imap::types::Flag::Seen));
452 +
453 + if let Some(body) = message.body() {
454 + body_count += 1;
455 + match mailparse::parse_mail(body) {
456 + Ok(parsed) => emails.push(Self::build_parsed_email(&parsed, uid, &folder_name, is_read)),
457 + Err(e) => {
458 + tracing::debug!(uid, folder = %folder_name, error = %e, "Failed to parse email");
459 + parse_errors += 1;
460 + }
562 461 }
563 462 }
564 463 }
565 464 }
566 465
567 466 debug.push(format!("msgs: {}, bodies: {}, parsed: {}, errs: {}", msg_count, body_count, emails.len(), parse_errors));
568 - drop(messages);
569 467 session.logout().await.ok();
570 468
571 469 Ok(FolderFetchResult {
@@ -688,6 +586,49 @@ impl ImapClient {
688 586 Ok(format!("Folder '{}': exists={}, recent={}, search_all={}", folder, exists, recent, count))
689 587 }
690 588
589 + /// Build a `ParsedEmail` from a parsed message and its IMAP metadata.
590 + ///
591 + /// Shared by both fetch loops (full-folder and incremental) so the header
592 + /// extraction lives in one place.
593 + fn build_parsed_email(
594 + parsed: &mailparse::ParsedMail,
595 + uid: u32,
596 + folder_name: &str,
597 + is_read: bool,
598 + ) -> ParsedEmail {
599 + let header = |key: &str| {
600 + parsed
601 + .headers
602 + .iter()
603 + .find(|h| h.get_key().to_lowercase() == key)
604 + .map(|h| h.get_value())
605 + };
606 +
607 + let date = header("date")
608 + .and_then(|s| mailparse::dateparse(&s).ok())
609 + .and_then(|ts| Utc.timestamp_opt(ts, 0).single())
610 + .unwrap_or(DateTime::UNIX_EPOCH);
611 +
612 + let (body_text, html_body) = Self::extract_body_with_html(parsed);
613 + let attachments = extract_attachment_parts(parsed);
614 +
615 + ParsedEmail {
616 + message_id: header("message-id"),
617 + in_reply_to: header("in-reply-to"),
618 + references_root: extract_references_root(&parsed.headers),
619 + imap_uid: uid,
620 + source_folder: folder_name.to_string(),
621 + from: header("from").unwrap_or_default(),
622 + to: header("to").unwrap_or_default(),
623 + subject: header("subject").unwrap_or_default(),
624 + body: body_text,
625 + html_body,
626 + date,
627 + is_read,
628 + attachments,
629 + }
630 + }
631 +
691 632 /// Extracts body text and optionally HTML from a parsed email.
692 633 /// Returns (plain_text_body, optional_html_body).
693 634 fn extract_body_with_html(mail: &mailparse::ParsedMail) -> (String, Option<String>) {
M src-tauri/src/email_sync_scheduler.rs +9
@@ -90,6 +90,13 @@ async fn check_and_sync_accounts(
90 90
91 91 debug!("Email sync scheduler: {} account(s) need sync", accounts.len());
92 92
93 + // A failing account never updates last_sync_at, so it stays "due" and remains
94 + // in this list every tick; any failure_counts key absent here belongs to an
95 + // account that was deleted or had sync disabled. Prune them after the pass so
96 + // the map can't grow without bound over a long-running session.
97 + let due_account_ids: std::collections::HashSet<String> =
98 + accounts.iter().map(|a| a.id.to_string()).collect();
99 +
93 100 // Sync each account independently — a failure in one account (e.g. expired
94 101 // token, network issue) must not prevent other accounts from syncing.
95 102 for account in accounts {
@@ -151,6 +158,8 @@ async fn check_and_sync_accounts(
151 158 }
152 159 }
153 160
161 + failure_counts.retain(|account_id, _| due_account_ids.contains(account_id));
162 +
154 163 Ok(())
155 164 }
156 165
M src-tauri/src/lib.rs +1
@@ -159,6 +159,7 @@ macro_rules! all_commands {
159 159 $crate::commands::list_events_for_contact,
160 160 $crate::commands::list_emails_for_contact,
161 161 $crate::commands::list_contacts_filtered,
162 + $crate::commands::list_contact_email_directory,
162 163 $crate::commands::bulk_delete_contacts,
163 164 $crate::commands::bulk_tag_contacts,
164 165 // Snooze Options
M src-tauri/src/state.rs +31 -2
@@ -30,6 +30,28 @@ pub const SYNC_SERVER_URL: &str = "https://makenot.work";
30 30 /// The API key is a public client identifier, not a secret.
31 31 const SYNCKIT_TOML: &str = include_str!("../../synckit.toml");
32 32
33 + /// Holds the optional sync client behind a poison-recovering lock.
34 + ///
35 + /// The inner `RwLock` is private to this newtype, so no caller can write
36 + /// `.read().expect("poisoned")` and crash a background loop on a poisoned lock.
37 + /// The only access is [`SyncClientCell::get`], which recovers instead of
38 + /// panicking — the constructive fix for the recurring sync-scheduler poison nit.
39 + #[derive(Default)]
40 + pub struct SyncClientCell(RwLock<Option<Arc<SyncKitClient>>>);
41 +
42 + impl SyncClientCell {
43 + /// Wrap an initial (possibly absent) client.
44 + pub(crate) fn new(client: Option<Arc<SyncKitClient>>) -> Self {
45 + Self(RwLock::new(client))
46 + }
47 +
48 + /// Clone out the current client, recovering from a poisoned lock rather
49 + /// than panicking.
50 + pub(crate) fn get(&self) -> Option<Arc<SyncKitClient>> {
51 + self.0.read().unwrap_or_else(|e| e.into_inner()).clone()
52 + }
53 + }
54 +
33 55 /// Application state holding database connections and repositories
34 56 pub struct AppState {
35 57 pub pool: SqlitePool,
@@ -49,7 +71,7 @@ pub struct AppState {
49 71 pub monthly_reviews: Arc<dyn MonthlyReviewRepository>,
50 72 pub backup_settings: Arc<dyn BackupSettingsRepository>,
51 73 pub sync_accounts: Arc<dyn SyncAccountRepository>,
52 - pub sync_client: RwLock<Option<Arc<SyncKitClient>>>,
74 + pub sync_client: SyncClientCell,
53 75 pub sync_lock: Arc<TokioMutex<()>>,
54 76 /// Per-account email sync locks to prevent concurrent syncs on the same account.
55 77 pub email_sync_locks: Arc<Mutex<std::collections::HashSet<goingson_core::EmailAccountId>>>,
@@ -163,7 +185,7 @@ impl AppState {
163 185 monthly_reviews,
164 186 backup_settings,
165 187 sync_accounts,
166 - sync_client: RwLock::new(sync_client.map(Arc::new)),
188 + sync_client: SyncClientCell::new(sync_client.map(Arc::new)),
167 189 sync_lock: Arc::new(TokioMutex::new(())),
168 190 email_sync_locks: Arc::new(Mutex::new(std::collections::HashSet::new())),
169 191 token_refresh_locks: Arc::new(Mutex::new(std::collections::HashMap::new())),
@@ -173,6 +195,13 @@ impl AppState {
173 195 })
174 196 }
175 197
198 + /// Clone out the current sync client, recovering from a poisoned lock
199 + /// instead of panicking. Background loops must use this (never a raw
200 + /// `.read().expect(...)`), so a writer panic can't kill sync for the session.
201 + pub(crate) fn read_recovering(&self) -> Option<Arc<SyncKitClient>> {
202 + self.sync_client.get()
203 + }
204 +
176 205 /// Gets or creates a per-account token refresh lock.
177 206 pub fn token_refresh_lock(&self, account_id: uuid::Uuid) -> Arc<TokioMutex<()>> {
178 207 let mut locks = self.token_refresh_locks.lock().unwrap_or_else(|e| e.into_inner());
M src-tauri/src/sync_scheduler.rs +3 -2
@@ -72,7 +72,7 @@ pub async fn start_sync_scheduler(app: tauri::AppHandle, cancel: CancellationTok
72 72 }
73 73 };
74 74
75 - let client: Arc<synckit_client::SyncKitClient> = match state.sync_client.read().expect("sync_client poisoned").clone() {
75 + let client: Arc<synckit_client::SyncKitClient> = match state.read_recovering() {
76 76 Some(c) => c,
77 77 None => continue,
78 78 };
@@ -189,7 +189,8 @@ pub async fn start_sync_scheduler(app: tauri::AppHandle, cancel: CancellationTok
189 189 info!("Auto-sync: pushed {}, pulled {}", result.pushed, result.pulled);
190 190 }
191 191 if result.pulled > 0 {
192 - let _ = app.emit("sync:changes-applied", ());
192 + // Carry the changed tables so the UI invalidates selectively.
193 + let _ = app.emit("sync:changes-applied", &result.pulled_tables);
193 194 }
194 195 }
195 196 Err(e) => {
M src-tauri/src/sync_service/mod.rs +8 -2
@@ -101,6 +101,9 @@ pub(crate) const DELETE_ORDER: &[&str] = &[
101 101 pub struct SyncResult {
102 102 pub pushed: i64,
103 103 pub pulled: i64,
104 + /// Distinct DB tables touched by the pull, for selective UI cache invalidation.
105 + #[serde(skip)]
106 + pub pulled_tables: Vec<String>,
104 107 }
105 108
106 109 // -- High-level sync --
@@ -122,7 +125,10 @@ pub async fn perform_sync_with_blobs(
122 125
123 126 // Push first, then pull
124 127 let pushed = push_changes(pool, client, device_id).await?;
125 - let pulled = pull_changes(pool, client, device_id).await?;
128 + let pull_outcome = pull_changes(pool, client, device_id).await?;
129 + let pulled = pull_outcome.applied;
130 + let mut pulled_tables: Vec<String> = pull_outcome.changed_tables.into_iter().collect();
131 + pulled_tables.sort();
126 132
127 133 // Sync blobs after metadata (upload local, download missing)
128 134 if let Some(dir) = data_dir {
@@ -138,7 +144,7 @@ pub async fn perform_sync_with_blobs(
138 144 let now = Utc::now().to_rfc3339();
139 145 set_sync_state(pool, "last_sync_at", &now).await?;
140 146
141 - Ok(SyncResult { pushed, pulled })
147 + Ok(SyncResult { pushed, pulled, pulled_tables })
142 148 }
143 149
144 150 // -- Initial snapshot --
M src-tauri/src/sync_service/pull.rs +15 -2
@@ -18,10 +18,11 @@ pub async fn pull_changes(
18 18 pool: &SqlitePool,
19 19 client: &SyncKitClient,
20 20 device_id: Uuid,
21 - ) -> Result<i64, CoreError> {
21 + ) -> Result<PullOutcome, CoreError> {
22 22 let cursor_str = get_sync_state(pool, "pull_cursor").await?;
23 23 let mut cursor: i64 = cursor_str.parse().unwrap_or(0);
24 24 let mut total_applied: i64 = 0;
25 + let mut changed_tables: std::collections::HashSet<String> = std::collections::HashSet::new();
25 26
26 27 // Read local pending changes once (unpushed changelog entries)
27 28 let local_pending = read_local_pending(pool).await?;
@@ -45,6 +46,7 @@ pub async fn pull_changes(
45 46 if !clean.is_empty() {
46 47 let clean_entries: Vec<ChangeEntry> =
47 48 clean.into_iter().map(|p| p.entry).collect();
49 + changed_tables.extend(clean_entries.iter().map(|e| e.table.clone()));
48 50 apply_remote_changes(pool, clean_entries).await?;
49 51 }
50 52
@@ -76,6 +78,7 @@ pub async fn pull_changes(
76 78
77 79 resolved_count = resolved_entries.len() as i64;
78 80 if !resolved_entries.is_empty() {
81 + changed_tables.extend(resolved_entries.iter().map(|e| e.table.clone()));
79 82 apply_remote_changes(pool, resolved_entries).await?;
80 83 }
81 84
@@ -102,7 +105,17 @@ pub async fn pull_changes(
102 105 if total_applied > 0 {
103 106 debug!("Pulled and applied {} remote changes", total_applied);
104 107 }
105 - Ok(total_applied)
108 + Ok(PullOutcome { applied: total_applied, changed_tables })
109 + }
110 +
111 + /// Result of a pull: how many changes landed and which tables they touched.
112 + ///
113 + /// The table set drives selective cache invalidation in the UI so a pull that
114 + /// only touched, say, `tasks` doesn't force the compose screen to re-hydrate
115 + /// every contact.
116 + pub struct PullOutcome {
117 + pub applied: i64,
118 + pub changed_tables: std::collections::HashSet<String>,
106 119 }
107 120
108 121 /// Read unpushed changelog entries as ChangeEntry values for conflict detection.
M src-tauri/src/test_utils.rs +1 -1
@@ -72,7 +72,7 @@ pub async fn setup_test_state() -> (Arc<AppState>, UserId) {
72 72 monthly_reviews: Arc::new(SqliteMonthlyReviewRepository::new(pool.clone())),
73 73 backup_settings: Arc::new(SqliteBackupSettingsRepository::new(pool.clone())),
74 74 sync_accounts: Arc::new(SqliteSyncAccountRepository::new(pool.clone())),
75 - sync_client: std::sync::RwLock::new(None),
75 + sync_client: crate::state::SyncClientCell::default(),
76 76 sync_lock: Arc::new(tokio::sync::Mutex::new(())),
77 77 email_sync_locks: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())),
78 78 token_refresh_locks: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),