max / goingson

Observability: instrument the sync core and every email/OAuth network call Add #[tracing::instrument(skip_all)] to the sync engine's mutating/outbound entry points and the previously-unspanned network clients, closing the gap between the fully-instrumented command layer and the silent protocol layer: - sync_service: pull_changes, push_changes, apply_upsert/apply_delete/ apply_email_account_upsert, assign_pending_hlcs, observe_remote, upload_pending_blobs, download_missing_blobs. - email: ImapClient connect/fetch_emails_incremental/move_message/ test_connection/list_folders/archive/unarchive; SmtpClient send_message/ test_connection. - oauth: provider exchange_code/refresh_token/revoke_token. Suite: 785 Rust + 55 JS pass; clippy --workspace --all-targets clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: Max Johnson <me@maxj.phd> · 2026-07-02 21:56 UTC

Commit: 17321630d2d5c12279dcc109d0db507783f9f628

Parent: 21b5072

8 files changed, +21 insertions, -0 deletions

M src-tauri/src/email/imap_client.rs +7

			@@ -124,6 +124,7 @@ impl ImapClient {
124	124		}
125	125
126	126		/// Connect to IMAP server and return an authenticated session
	127	+	#[tracing::instrument(skip_all)]
127	128		async fn connect(&self) -> Result<ImapSession, String> {
128	129		let addr = format!("{}:{}", self.server, self.port);
129	130
			@@ -315,6 +316,7 @@ impl ImapClient {
315	316
316	317		/// Fetch emails incrementally using IMAP UIDs when sync state is available,
317	318		/// falling back to SINCE-based search otherwise.
	319	+	#[tracing::instrument(skip_all)]
318	320		pub async fn fetch_emails_incremental(
319	321		&self,
320	322		folder: &str,
			@@ -506,6 +508,7 @@ impl ImapClient {
506	508		}
507	509
508	510		/// Move a message between folders using IMAP UID MOVE or COPY+DELETE
	511	+	#[tracing::instrument(skip_all)]
509	512		pub async fn move_message(
510	513		&self,
511	514		uid: u32,
			@@ -556,15 +559,18 @@ impl ImapClient {
556	559		}
557	560
558	561		/// Archive a message (move from INBOX to Archive folder)
	562	+	#[tracing::instrument(skip_all)]
559	563		pub async fn archive_message(&self, uid: u32, archive_folder: &str) -> Result<(), String> {
560	564		self.move_message(uid, "INBOX", archive_folder).await
561	565		}
562	566
563	567		/// Unarchive a message (move from Archive folder to INBOX)
	568	+	#[tracing::instrument(skip_all)]
564	569		pub async fn unarchive_message(&self, uid: u32, archive_folder: &str) -> Result<(), String> {
565	570		self.move_message(uid, archive_folder, "INBOX").await
566	571		}
567	572
	573	+	#[tracing::instrument(skip_all)]
568	574		pub async fn test_connection(&self) -> Result<(), String> {
569	575		let mut session = self.connect().await?;
570	576		session.logout().await.ok();
			@@ -572,6 +578,7 @@ impl ImapClient {
572	578		}
573	579
574	580		/// List all available folders on the IMAP server
	581	+	#[tracing::instrument(skip_all)]
575	582		pub async fn list_folders(&self) -> Result<Vec<String>, String> {
576	583		let mut session = self.connect().await?;
577	584

M src-tauri/src/email/smtp_client.rs +2

			@@ -228,6 +228,7 @@ impl SmtpClient {
228	228		Ok(mailer)
229	229		}
230	230
	231	+	#[tracing::instrument(skip_all)]
231	232		pub async fn send_message(&self, params: &SendParams<'_>) -> Result<String, String> {
232	233		// Generate message ID before building so sent email and local DB agree
233	234		let message_id = format!(
			@@ -311,6 +312,7 @@ impl SmtpClient {
311	312		Ok(message_id)
312	313		}
313	314
	315	+	#[tracing::instrument(skip_all)]
314	316		pub async fn test_connection(&self) -> Result<(), String> {
315	317		let mailer = self.build_mailer()?;
316	318

M src-tauri/src/oauth/provider.rs +3

			@@ -160,6 +160,7 @@ pub trait OAuthProvider: Send + Sync + 'static {
160	160		///
161	161		/// Default implementation handles standard OAuth2 token exchange with PKCE.
162	162		/// Respects `client_auth_method()` for credential handling.
	163	+	#[tracing::instrument(skip_all)]
163	164		async fn exchange_code(
164	165		&self,
165	166		code: &str,
			@@ -229,6 +230,7 @@ pub trait OAuthProvider: Send + Sync + 'static {
229	230		/// Refreshes an expired access token using a refresh token.
230	231		///
231	232		/// Default implementation handles standard OAuth2 token refresh.
	233	+	#[tracing::instrument(skip_all)]
232	234		async fn refresh_token(&self, refresh_token: &str) -> Result<TokenResult, String> {
233	235		let config = self.config();
234	236
			@@ -289,6 +291,7 @@ pub trait OAuthProvider: Send + Sync + 'static {
289	291		/// endpoint configured (e.g. Microsoft, which only supports session
290	292		/// logout). Used on account disconnect so a leaked refresh token can't be
291	293		/// used after the user removes the account locally.
	294	+	#[tracing::instrument(skip_all)]
292	295		async fn revoke_token(&self, token: &str) -> Result<(), String> {
293	296		let Some(revoke_url) = self.config().revoke_url.as_ref() else {
294	297		return Ok(());

M src-tauri/src/sync_service/apply.rs +3

			@@ -112,6 +112,7 @@ pub(crate) fn table_columns(table: &str) -> Option<&'static [&'static str]> {
112	112		/// would wipe them on any parent re-upsert. `ON CONFLICT DO UPDATE` mutates the
113	113		/// row in place, so children survive and referential integrity holds regardless
114	114		/// of FK enforcement. Every syncable table has `id` as its primary key.
	115	+	#[tracing::instrument(skip_all)]
115	116		pub(crate) async fn apply_upsert(
116	117		conn: &mut SqliteConnection,
117	118		table: &str,
			@@ -204,6 +205,7 @@ async fn not_null_columns(
204	205		/// leaving `password`, `oauth2_access_token`, `oauth2_refresh_token`, and
205	206		/// `oauth2_token_expires_at` untouched on existing rows. New rows get `password = ''`
206	207		/// to satisfy the NOT NULL constraint.
	208	+	#[tracing::instrument(skip_all)]
207	209		async fn apply_email_account_upsert(
208	210		conn: &mut SqliteConnection,
209	211		data: &serde_json::Value,
			@@ -274,6 +276,7 @@ pub(crate) fn bind_json_value<'q>(
274	276		}
275	277
276	278		/// Apply a DELETE for a remote change.
	279	+	#[tracing::instrument(skip_all)]
277	280		pub(crate) async fn apply_delete(conn: &mut SqliteConnection, table: &str, row_id: &str) -> Result<(), CoreError> {
278	281		// Validate table name is in our whitelist
279	282		if table_columns(table).is_none() {

M src-tauri/src/sync_service/blob_sync.rs +2

			@@ -22,6 +22,7 @@ fn short(hash: &str) -> &str {
22	22		///
23	23		/// Queries all distinct blob hashes from attachments, checks which ones have local
24	24		/// files, and uploads them via the SyncKit blob API (presigned S3 + E2E encryption).
	25	+	#[tracing::instrument(skip_all)]
25	26		pub async fn upload_pending_blobs(
26	27		pool: &SqlitePool,
27	28		data_dir: &Path,
			@@ -91,6 +92,7 @@ pub async fn upload_pending_blobs(
91	92		///
92	93		/// After metadata sync pulls attachment records from other devices, this function
93	94		/// downloads the actual blob data from SyncKit (presigned S3 + E2E decryption).
	95	+	#[tracing::instrument(skip_all)]
94	96		pub async fn download_missing_blobs(
95	97		pool: &SqlitePool,
96	98		data_dir: &Path,

M src-tauri/src/sync_service/hlc.rs +2

			@@ -32,6 +32,7 @@ async fn load_clock(pool: &SqlitePool, node: Uuid) -> Result<Hlc, CoreError> {
32	32		/// persistent clock once per row. Idempotent — already-stamped rows are skipped — so
33	33		/// push and conflict-detection observe identical stamps. Runs in one transaction so a
34	34		/// crash mid-stamp leaves the clock and the rows consistent (all-or-nothing).
	35	+	#[tracing::instrument(skip_all)]
35	36		pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(), CoreError> {
36	37		let ids: Vec<(i64, String, String)> = sqlx::query_as(
37	38		"SELECT id, table_name, row_id FROM sync_changelog \
			@@ -82,6 +83,7 @@ pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(
82	83		/// Advance the persistent clock past a remote HLC observed during pull, so that
83	84		/// subsequent local changes causally follow it. Monotonic: advancing past the batch
84	85		/// maximum is enough to cover every change in the batch.
	86	+	#[tracing::instrument(skip_all)]
85	87		pub async fn observe_remote(pool: &SqlitePool, remote: Hlc, device_id: Uuid) -> Result<(), CoreError> {
86	88		let clock = load_clock(pool, device_id).await?;
87	89		let advanced = Hlc::observe(clock, remote, Utc::now().timestamp_millis(), device_id);

M src-tauri/src/sync_service/pull.rs +1

			@@ -15,6 +15,7 @@ use super::hlc::{assign_pending_hlcs, load_committed_hlcs, observe_remote, recor
15	15		use super::{UPSERT_ORDER, DELETE_ORDER};
16	16		use super::apply::{apply_upsert, apply_delete};
17	17
	18	+	#[tracing::instrument(skip_all)]
18	19		pub async fn pull_changes(
19	20		pool: &SqlitePool,
20	21		client: &SyncKitClient,

M src-tauri/src/sync_service/push.rs +1

			@@ -14,6 +14,7 @@ use super::hlc::assign_pending_hlcs;
14	14		/// (id, table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter).
15	15		type PendingRow = (i64, String, String, String, String, Option<String>, Option<i64>, Option<i64>);
16	16
	17	+	#[tracing::instrument(skip_all)]
17	18		pub async fn push_changes(
18	19		pool: &SqlitePool,
19	20		client: &SyncKitClient,