Skip to main content

max / makenotwork

synckit: indexed pull hot path, SSE cursor, deferred egress, sealed warning tick (ultra-fuzz --deep Perf A+) Performance A->A+ cold spots from the Run 3 SyncKit pass: - Pull/push device verification (P1): replaced get_sync_devices (fetch every device) + linear scan with sync_device_belongs, an indexed EXISTS point lookup, so verification stays O(1) as device count grows. - Pull round-trips (P2): folded the separate touch + cursor-update into one touch_and_advance_cursor UPDATE; removed the now-dead update_device_cursor. - SSE thundering herd (P2): the notify channel now carries the new max seq (Sender<()> -> Sender<i64>); the "changed" event ships {"seq":N} so a subscriber already at that cursor can skip a redundant pull. Additive: clients that ignore seq pull as before. - Egress counter (P4): the bytes_egress bump on blob download (a free dashboard metric, single hot-row UPDATE) now fires onto the bounded background pool instead of blocking the download and contending its lock. - Warning-tick CHRONIC seal: added a build-time guard test that fails if check_and_send_warnings ever names the email send inline — the serial-send-in- the-tick shape (Runs 1-3) can no longer reappear unnoticed; sends stay on state.bg in send_warning. Full synckit workflow suite (54) + lib units (12, incl. the seal) green; clippy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 20:23 UTC
Commit: 0fb3b363b9893e3ab135d30f6dad416ff0bb85b3
Parent: fc0659c
7 files changed, +109 insertions, -43 deletions
M server/src/db/synckit/devices.rs +43
@@ -57,6 +57,49 @@ pub async fn get_sync_devices(
57 57 Ok(devices)
58 58 }
59 59
60 + /// Does this device belong to this `(app, user)`? Indexed point lookup on the
61 + /// `sync_devices` PK + owner columns — the hot push/pull paths use this instead
62 + /// of fetching every device and linear-scanning, so device verification stays
63 + /// O(1) as a user's device count grows.
64 + #[tracing::instrument(skip_all)]
65 + pub async fn sync_device_belongs(
66 + pool: &PgPool,
67 + device_id: SyncDeviceId,
68 + app_id: SyncAppId,
69 + user_id: UserId,
70 + ) -> Result<bool> {
71 + let exists: bool = sqlx::query_scalar(
72 + "SELECT EXISTS(SELECT 1 FROM sync_devices WHERE id = $1 AND app_id = $2 AND user_id = $3)",
73 + )
74 + .bind(device_id)
75 + .bind(app_id)
76 + .bind(user_id)
77 + .fetch_one(pool)
78 + .await?;
79 + Ok(exists)
80 + }
81 +
82 + /// Mark a device seen and advance its pull cursor in one statement — the pull
83 + /// hot path used to issue a separate touch and a separate cursor update; this
84 + /// folds both into a single UPDATE. `GREATEST` keeps the cursor monotonic.
85 + #[tracing::instrument(skip_all)]
86 + pub async fn touch_and_advance_cursor(
87 + pool: &PgPool,
88 + device_id: SyncDeviceId,
89 + new_cursor: i64,
90 + ) -> Result<()> {
91 + sqlx::query(
92 + "UPDATE sync_devices
93 + SET last_seen_at = NOW(), last_pulled_seq = GREATEST(last_pulled_seq, $2)
94 + WHERE id = $1",
95 + )
96 + .bind(device_id)
97 + .bind(new_cursor)
98 + .execute(pool)
99 + .await?;
100 + Ok(())
101 + }
102 +
60 103 /// Delete a device by ID (only if owned by user within app).
61 104 #[tracing::instrument(skip_all)]
62 105 pub async fn delete_sync_device(
M server/src/db/synckit/keys.rs -17
@@ -132,23 +132,6 @@ pub async fn prune_sync_log(pool: &PgPool, retain_days: i64) -> Result<u64> {
132 132 Ok(result.rows_affected())
133 133 }
134 134
135 - /// Update a device's last-pulled cursor position.
136 - #[tracing::instrument(skip_all)]
137 - pub async fn update_device_cursor(
138 - pool: &PgPool,
139 - device_id: SyncDeviceId,
140 - seq: i64,
141 - ) -> Result<()> {
142 - sqlx::query(
143 - "UPDATE sync_devices SET last_pulled_seq = GREATEST(last_pulled_seq, $1) WHERE id = $2",
144 - )
145 - .bind(seq)
146 - .bind(device_id)
147 - .execute(pool)
148 - .await?;
149 - Ok(())
150 - }
151 -
152 135 /// Compact the sync log by removing entries that all devices for a given
153 136 /// (app_id, user_id) have already pulled. Keeps a safety margin of entries
154 137 /// newer than `min_age_days` regardless of cursor positions.
M server/src/lib.rs +4 -2
@@ -118,8 +118,10 @@ pub struct AppState {
118 118 /// Set by the deploy script via the internal API before uploading a new binary.
119 119 pub restart_at: Arc<std::sync::atomic::AtomicI64>,
120 120 /// SSE push notification channels for SyncKit subscribers.
121 - /// Key: (app_id, user_id), Value: broadcast sender that SSE connections subscribe to.
122 - pub sync_notify: Arc<DashMap<(SyncAppId, UserId), tokio::sync::broadcast::Sender<()>>>,
121 + /// Key: (app_id, user_id), Value: broadcast sender carrying the new max
122 + /// `seq` after each push, so a subscriber already at that cursor can skip
123 + /// a redundant pull.
124 + pub sync_notify: Arc<DashMap<(SyncAppId, UserId), tokio::sync::broadcast::Sender<i64>>>,
123 125 /// Concurrent SSE connection count per user (for rate limiting).
124 126 pub sse_connections: Arc<DashMap<UserId, std::sync::atomic::AtomicUsize>>,
125 127 /// Prometheus metrics handle for rendering the admin dashboard. `None` in tests.
M server/src/routes/synckit/blobs.rs +11 -6
@@ -248,12 +248,17 @@ pub(super) async fn blob_download_url(
248 248 // Count egress optimistically at presign time. The client may not
249 249 // actually download (retries that hit dedup-cached content, for
250 250 // example), so this overcounts slightly. Acceptable for a free
251 - // dashboard metric.
252 - if let Err(e) = synckit_billing::add_bytes_egress(
253 - &state.db, sync_user.app_id, blob.size_bytes,
254 - ).await {
255 - tracing::error!(error = ?e, app_id = %sync_user.app_id, "failed to bump bytes_egress_period");
256 - }
251 + // dashboard metric. Deferred onto the bounded background pool: it's a
252 + // single hot-row UPDATE that the download path shouldn't wait on or
253 + // contend its lock against (Perf P4).
254 + let db = state.db.clone();
255 + let app_id = sync_user.app_id;
256 + let egress_bytes = blob.size_bytes;
257 + state.bg.spawn("synckit-egress-bump", async move {
258 + if let Err(e) = synckit_billing::add_bytes_egress(&db, app_id, egress_bytes).await {
259 + tracing::error!(error = ?e, app_id = %app_id, "failed to bump bytes_egress_period");
260 + }
261 + });
257 262 }
258 263
259 264 let download_url = synckit_s3
M server/src/routes/synckit/subscribe.rs +4 -2
@@ -28,7 +28,7 @@ use crate::{
28 28 /// Drop guard that decrements the per-user SSE connection counter.
29 29 struct SseConnectionGuard {
30 30 sse_connections: std::sync::Arc<dashmap::DashMap<UserId, std::sync::atomic::AtomicUsize>>,
31 - sync_notify: std::sync::Arc<dashmap::DashMap<(SyncAppId, UserId), tokio::sync::broadcast::Sender<()>>>,
31 + sync_notify: std::sync::Arc<dashmap::DashMap<(SyncAppId, UserId), tokio::sync::broadcast::Sender<i64>>>,
32 32 user_id: UserId,
33 33 app_id: SyncAppId,
34 34 }
@@ -134,7 +134,9 @@ pub(super) async fn sync_subscribe(
134 134 };
135 135
136 136 let stream = BroadcastStream::new(rx).filter_map(|result| match result {
137 - Ok(()) => Some(Ok(Event::default().event("changed").data("{}"))),
137 + // Carry the new max seq so a client already at this cursor can skip the
138 + // pull. Clients that don't read `seq` just pull as before.
139 + Ok(seq) => Some(Ok(Event::default().event("changed").data(format!("{{\"seq\":{seq}}}")))),
138 140 Err(_) => None, // Lagged — skip missed events, client will pull anyway
139 141 });
140 142
M server/src/routes/synckit/sync.rs +13 -16
@@ -79,9 +79,9 @@ pub(super) async fn sync_push(
79 79 }
80 80 }
81 81
82 - // Verify device belongs to this user + app
83 - let devices = db::synckit::get_sync_devices(&state.db, app_id, user_id).await?;
84 - if !devices.iter().any(|d| d.id == req.device_id) {
82 + // Verify device belongs to this user + app (indexed point lookup, not a
83 + // full device fetch + linear scan).
84 + if !db::synckit::sync_device_belongs(&state.db, req.device_id, app_id, user_id).await? {
85 85 return Err(AppError::BadRequest("Unknown device".to_string()));
86 86 }
87 87
@@ -113,9 +113,11 @@ pub(super) async fn sync_push(
113 113 )
114 114 .await?;
115 115
116 - // Notify SSE subscribers that new changes are available
116 + // Notify SSE subscribers, carrying the new max seq so a device already at
117 + // (or past) this cursor can skip a redundant pull — avoids the thundering
118 + // herd where every subscriber pulls on every push.
117 119 if let Some(sender) = state.sync_notify.get(&(app_id, user_id)) {
118 - let _ = sender.send(()); // Ignore errors (no subscribers = ok)
120 + let _ = sender.send(cursor); // Ignore errors (no subscribers = ok)
119 121 }
120 122
121 123 Ok(Json(PushResponse { cursor }).into_response())
@@ -138,15 +140,12 @@ pub(super) async fn sync_pull(
138 140 tracing::Span::current().record("app_id", tracing::field::display(&app_id));
139 141 tracing::Span::current().record("user_id", tracing::field::display(&user_id));
140 142
141 - // Verify device belongs to this user + app
142 - let devices = db::synckit::get_sync_devices(&state.db, app_id, user_id).await?;
143 - if !devices.iter().any(|d| d.id == req.device_id) {
143 + // Verify device belongs to this user + app (indexed point lookup, not a
144 + // full device fetch + linear scan).
145 + if !db::synckit::sync_device_belongs(&state.db, req.device_id, app_id, user_id).await? {
144 146 return Err(AppError::BadRequest("Unknown device".to_string()));
145 147 }
146 148
147 - // Touch device
148 - db::synckit::touch_sync_device(&state.db, req.device_id).await?;
149 -
150 149 // Validate table name filters if provided
151 150 if let Some(ref tables) = req.tables {
152 151 if tables.len() > 50 {
@@ -174,11 +173,9 @@ pub(super) async fn sync_pull(
174 173 let has_more = entries.len() as i64 == page_size;
175 174 let new_cursor = entries.last().map(|e| e.seq).unwrap_or(req.cursor);
176 175
177 - // Track this device's cursor position for sync log compaction.
178 - // GREATEST in the query ensures we never regress the cursor.
179 - if new_cursor > req.cursor {
180 - db::synckit::update_device_cursor(&state.db, req.device_id, new_cursor).await?;
181 - }
176 + // Mark the device seen and advance its compaction cursor in one statement.
177 + // GREATEST keeps the cursor monotonic even if `new_cursor == req.cursor`.
178 + db::synckit::touch_and_advance_cursor(&state.db, req.device_id, new_cursor).await?;
182 179
183 180 let changes: Vec<PullChangeEntry> = entries
184 181 .into_iter()
M server/src/scheduler/synckit_warnings.rs +34
@@ -109,6 +109,40 @@ async fn send_warning(email: EmailClient, db: sqlx::PgPool, host_url: Arc<str>,
109 109 mod tests {
110 110 use crate::db::synckit_billing::highest_breached_threshold;
111 111
112 + /// Constructive seal for the warning-loop CHRONIC (ultra-fuzz Runs 1-3).
113 + /// The send must only ever happen on the bounded background pool, inside
114 + /// `send_warning` — never inline in the scheduler tick. This fails the build
115 + /// if `check_and_send_warnings` ever names the email send directly, so the
116 + /// drifted "serial send inside the tick" shape can't reappear unnoticed.
117 + #[test]
118 + fn tick_never_sends_inline() {
119 + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
120 + .join("src/scheduler/synckit_warnings.rs");
121 + let src = std::fs::read_to_string(&path).expect("read warnings source");
122 +
123 + let start = src
124 + .find("async fn check_and_send_warnings")
125 + .expect("tick fn present");
126 + let rest = &src[start..];
127 + // The tick body ends where the next top-level fn (send_warning) begins.
128 + let end = rest[1..]
129 + .find("\nasync fn ")
130 + .or_else(|| rest[1..].find("\nfn "))
131 + .map(|i| i + 1)
132 + .unwrap_or(rest.len());
133 + let tick_body = &rest[..end];
134 +
135 + assert!(
136 + !tick_body.contains("send_synckit_usage_warning"),
137 + "CHRONIC seal violated: check_and_send_warnings must not perform an \
138 + email send inline — enqueue onto state.bg and send in send_warning."
139 + );
140 + assert!(
141 + tick_body.contains("state.bg.spawn"),
142 + "warning tick must dispatch sends onto the bounded background pool",
143 + );
144 + }
145 +
112 146 #[test]
113 147 fn no_breach_below_first_threshold() {
114 148 // 50% used, no warning yet