Skip to main content

max / makenotwork

Decompose AppState Phase 5: slice methods + worker projections Clears the two optional Phase 3 leftovers. Sync slice gains the orchestration that handlers were doing by hand: notify_push (SSE push fan-out) and subscribe_channel (get-or-create the per-(app,user) broadcast channel). Handlers no longer touch sync.sync_notify directly; the channel depth is the named SYNC_NOTIFY_CHANNEL_DEPTH const. Trim the Scanning slice to just { scanner }. No handler that extracts State<Scanning> ever read scan_semaphore -- it is worker-only (WorkerContext, wired from AppLimiters at startup), so carrying it in the slice misstated every scan handler's dependencies. Narrow the two background workers off &AppState via hand-written FromRef projections: monitor::MonitorCtx and build_runner::BuildCtx. Field names mirror AppState so the loop/runner bodies are an unchanged ctx. rename. spawn_monitor and dispatch_pending_build take the ctx; main.rs and the scheduler project it with FromRef::from_ref(&state) at the call site, same seam as the stripe dispatcher. Both ctx types join the from_ref_slices guard test. No non-handler &AppState now remains outside the composition roots and the FromRequestParts extractor machinery. Crate + tests compile clean, clippy clean, guard test green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-14 14:26 UTC
Commit: 01e6a857d21a1b35791dd6034a5e111727db8471
Parent: 8b97cb2
7 files changed, +201 insertions, -88 deletions
M server/src/build_runner.rs +78 -39
@@ -4,12 +4,51 @@
4 4 //! running and one is pending, it spawns a `tokio::spawn` task that SSHes to
5 5 //! the appropriate build host, clones, builds, signs, and uploads artifacts.
6 6
7 + use std::sync::Arc;
7 8 use std::time::Duration;
8 9
10 + use axum::extract::FromRef;
11 +
12 + use crate::config::Config;
9 13 use crate::constants::{BUILD_MAX_LOG_BYTES, BUILD_TIMEOUT_SECS};
10 14 use crate::db::{self, BuildStatus, DbBuild, DbBuildConfig};
15 + use crate::scanning::ScanPipeline;
16 + use crate::storage::StorageBackend;
17 + use crate::wam_client::WamClient;
11 18 use crate::AppState;
12 19
20 + /// The slice of [`AppState`] the OTA build runner needs: the DB pool it drives
21 + /// every query through, the config for build-host lookup, the optional scanner
22 + /// it enqueues uploaded artifacts onto, the SyncKit blob bucket it uploads to,
23 + /// and the optional WAM client for build-failure tickets.
24 + ///
25 + /// Field names mirror [`AppState`] (`ctx.db`, `ctx.config`, `ctx.scanner`,
26 + /// `ctx.wam`) so the runner body reads like the pre-decomposition code. This is a
27 + /// projection view (like the handler slices in [`crate`]); [`dispatch_pending_build`]
28 + /// takes it instead of the whole `AppState`, so the runner's dependencies are
29 + /// stated, not ambient. `synckit_s3` is flattened out of `AppStorage` here since
30 + /// the runner touches only that one bucket.
31 + #[derive(Clone)]
32 + pub struct BuildCtx {
33 + pub db: sqlx::PgPool,
34 + pub config: Config,
35 + pub scanner: Option<Arc<ScanPipeline>>,
36 + pub synckit_s3: Option<Arc<dyn StorageBackend>>,
37 + pub wam: Option<WamClient>,
38 + }
39 +
40 + impl FromRef<AppState> for BuildCtx {
41 + fn from_ref(s: &AppState) -> Self {
42 + Self {
43 + db: s.db.clone(),
44 + config: s.config.clone(),
45 + scanner: s.scanner.clone(),
46 + synckit_s3: s.storage.synckit_s3.clone(),
47 + wam: s.wam.clone(),
48 + }
49 + }
50 + }
51 +
13 52 /// Wall-clock cap for an artifact SCP download. Only the build ssh was
14 53 /// timeout-wrapped; a build host that stalled mid-transfer wedged the single
15 54 /// build slot forever (Run 15 Resilience). Artifacts can be large, so this is
@@ -117,9 +156,9 @@ fn build_host_for_target<'a>(config: &'a crate::config::Config, os: &str) -> Opt
117 156 ///
118 157 /// Called from the scheduler loop. Non-blocking — spawns the build task and returns.
119 158 #[tracing::instrument(skip_all, name = "build_runner::dispatch")]
120 - pub async fn dispatch_pending_build(state: &AppState) {
159 + pub async fn dispatch_pending_build(ctx: &BuildCtx) {
121 160 // Recover from stale running builds (e.g. server crashed mid-build)
122 - match db::builds::fail_stale_running_builds(&state.db, BUILD_TIMEOUT_SECS as i64).await {
161 + match db::builds::fail_stale_running_builds(&ctx.db, BUILD_TIMEOUT_SECS as i64).await {
123 162 Ok(n) if n > 0 => {
124 163 tracing::warn!(count = n, "marked stale running builds as failed");
125 164 }
@@ -129,7 +168,7 @@ pub async fn dispatch_pending_build(state: &AppState) {
129 168 _ => {}
130 169 }
131 170
132 - let build = match db::builds::claim_pending_build(&state.db).await {
171 + let build = match db::builds::claim_pending_build(&ctx.db).await {
133 172 Ok(Some(b)) => b,
134 173 Ok(None) => return,
135 174 Err(e) => {
@@ -138,12 +177,12 @@ pub async fn dispatch_pending_build(state: &AppState) {
138 177 }
139 178 };
140 179
141 - let config = match db::builds::get_build_config_by_app(&state.db, build.app_id).await {
180 + let config = match db::builds::get_build_config_by_app(&ctx.db, build.app_id).await {
142 181 Ok(Some(c)) => c,
143 182 Ok(None) => {
144 183 tracing::error!(build_id = %build.id, "build config not found for pending build");
145 184 if let Err(e) = db::builds::update_build_status(
146 - &state.db,
185 + &ctx.db,
147 186 build.id,
148 187 BuildStatus::Failed,
149 188 Some("Build config not found"),
@@ -159,9 +198,9 @@ pub async fn dispatch_pending_build(state: &AppState) {
159 198 }
160 199 };
161 200
162 - let state = state.clone();
201 + let ctx = ctx.clone();
163 202 tokio::spawn(async move {
164 - run_build(&state, &build, &config).await;
203 + run_build(&ctx, &build, &config).await;
165 204 });
166 205 }
167 206
@@ -183,7 +222,7 @@ type GroupOutput = (Vec<TargetArtifact>, Vec<TargetError>);
183 222
184 223 /// Execute a full build: iterate targets, SSH to hosts, build, upload artifacts.
185 224 #[tracing::instrument(skip_all, name = "build_runner::run_build", fields(build_id = %build.id, version = %build.version))]
186 - async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
225 + async fn run_build(ctx: &BuildCtx, build: &DbBuild, config: &DbBuildConfig) {
187 226 let mut artifact_keys: Vec<(String, String, String, String)> = Vec::new(); // (target_os, arch, s3_key, signature)
188 227 let mut failed_count: usize = 0;
189 228 let mut first_error: Option<String> = None;
@@ -197,7 +236,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
197 236 for target_str in &config.targets {
198 237 let Some((target_os, arch)): Option<(&str, &str)> = target_str.split_once('/') else {
199 238 let msg = format!("invalid target format: {target_str}\n");
200 - let _ = append_log_bounded(state, build.id, &msg).await;
239 + let _ = append_log_bounded(ctx, build.id, &msg).await;
201 240 failed_count += 1;
202 241 if first_error.is_none() {
203 242 first_error = Some(format!("invalid target format: {target_str}"));
@@ -205,12 +244,12 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
205 244 continue;
206 245 };
207 246
208 - let host = match build_host_for_target(&state.config, target_os) {
247 + let host = match build_host_for_target(&ctx.config, target_os) {
209 248 Some(h) => h,
210 249 None => {
211 250 let msg = format!("no build host for {target_os}, skipping {target_str}\n");
212 251 tracing::warn!("{}", msg.trim());
213 - let _ = append_log_bounded(state, build.id, &msg).await;
252 + let _ = append_log_bounded(ctx, build.id, &msg).await;
214 253 failed_count += 1;
215 254 if first_error.is_none() {
216 255 first_error = Some(format!("no build host for {target_os}"));
@@ -232,14 +271,14 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
232 271 let group_count = groups.len();
233 272 let mut set: tokio::task::JoinSet<(usize, GroupOutput)> = tokio::task::JoinSet::new();
234 273 for (idx, (host, targets)) in groups.into_iter().enumerate() {
235 - let state = state.clone();
274 + let ctx = ctx.clone();
236 275 let build = build.clone();
237 276 let config = config.clone();
238 277 set.spawn(async move {
239 278 let mut oks: Vec<TargetArtifact> = Vec::new();
240 279 let mut errs: Vec<TargetError> = Vec::new();
241 280 for (target_os, arch) in &targets {
242 - match execute_target(&state, &build, &config, &host, target_os, arch).await {
281 + match execute_target(&ctx, &build, &config, &host, target_os, arch).await {
243 282 Ok((s3_key, signature)) => {
244 283 oks.push((target_os.clone(), arch.clone(), s3_key, signature));
245 284 }
@@ -271,7 +310,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
271 310 for (target_str, e) in errs {
272 311 let msg = format!("target {target_str} failed: {e}\n");
273 312 tracing::error!("{}", msg.trim());
274 - let _ = append_log_bounded(state, build.id, &msg).await;
313 + let _ = append_log_bounded(ctx, build.id, &msg).await;
275 314 failed_count += 1;
276 315 if first_error.is_none() {
277 316 first_error = Some(e);
@@ -282,7 +321,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
282 321 if artifact_keys.is_empty() || failed_count > 0 {
283 322 let err_msg = build_failure_message(artifact_keys.len(), failed_count, first_error.as_deref());
284 323 if let Err(e) = db::builds::update_build_status(
285 - &state.db,
324 + &ctx.db,
286 325 build.id,
287 326 BuildStatus::Failed,
288 327 Some(&err_msg),
@@ -290,7 +329,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
290 329 .await {
291 330 tracing::error!(build_id = %build.id, error = ?e, "failed to mark build as failed");
292 331 }
293 - if let Some(ref wam) = state.wam {
332 + if let Some(ref wam) = ctx.wam {
294 333 let title = format!("Build failed: {} v{}", build.tag, build.version);
295 334 wam.create_ticket(&title, Some(&err_msg), "high", "build-failed", Some(&build.id.to_string())).await;
296 335 }
@@ -310,7 +349,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
310 349 None => {
311 350 let msg = "build produced no signed artifact; refusing to publish an unsigned OTA release";
312 351 if let Err(e) =
313 - db::builds::update_build_status(&state.db, build.id, BuildStatus::Failed, Some(msg))
352 + db::builds::update_build_status(&ctx.db, build.id, BuildStatus::Failed, Some(msg))
314 353 .await
315 354 {
316 355 tracing::error!(build_id = %build.id, error = ?e, "failed to mark build failed (missing signature)");
@@ -321,7 +360,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
321 360
322 361 // Create OTA release (only for fully successful builds)
323 362 let release = match db::ota::create_release(
324 - &state.db,
363 + &ctx.db,
325 364 build.app_id,
326 365 &build.version,
327 366 &format!("Automated build from tag {}", build.tag),
@@ -333,7 +372,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
333 372 Err(e) => {
334 373 let msg = format!("failed to create OTA release: {e}");
335 374 if let Err(e) = db::builds::update_build_status(
336 - &state.db,
375 + &ctx.db,
337 376 build.id,
338 377 BuildStatus::Failed,
339 378 Some(&msg),
@@ -346,7 +385,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
346 385 };
347 386
348 387 // The app owner is the responsible identity for the artifact scans.
349 - let owner_id = db::synckit::get_sync_app_by_id(&state.db, build.app_id)
388 + let owner_id = db::synckit::get_sync_app_by_id(&ctx.db, build.app_id)
350 389 .await
351 390 .ok()
352 391 .flatten()
@@ -357,19 +396,19 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
357 396 // channel.
358 397 for (target_os, arch, s3_key, _signature) in &artifact_keys {
359 398 // Get file size from S3 via HEAD request (best-effort, use 0 if unavailable)
360 - let file_size = if let Some(s3) = state.storage.synckit_s3.as_ref() {
399 + let file_size = if let Some(s3) = ctx.synckit_s3.as_ref() {
361 400 s3.object_size(s3_key).await.ok().flatten().unwrap_or(0)
362 401 } else {
363 402 0
364 403 };
365 404
366 - match db::ota::create_artifact(&state.db, release.id, target_os, arch, s3_key, file_size)
405 + match db::ota::create_artifact(&ctx.db, release.id, target_os, arch, s3_key, file_size)
367 406 .await
368 407 {
369 408 Ok(artifact) => {
370 409 if let Some(owner_id) = owner_id
371 410 && let Err(e) = crate::routes::ota::enqueue_ota_artifact_scan(
372 - &state.db, state.scanner.as_ref(), artifact.id, s3_key, owner_id, file_size,
411 + &ctx.db, ctx.scanner.as_ref(), artifact.id, s3_key, owner_id, file_size,
373 412 )
374 413 .await
375 414 {
@@ -381,13 +420,13 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
381 420 }
382 421
383 422 // Link build to release
384 - if let Err(e) = db::builds::set_build_release(&state.db, build.id, release.id).await {
423 + if let Err(e) = db::builds::set_build_release(&ctx.db, build.id, release.id).await {
385 424 tracing::error!(build_id = %build.id, release_id = %release.id, error = ?e, "failed to link build to release");
386 425 }
387 426
388 427 // All targets succeeded (partial failures return early above)
389 428 if let Err(e) = db::builds::update_build_status(
390 - &state.db,
429 + &ctx.db,
391 430 build.id,
392 431 BuildStatus::Succeeded,
393 432 None,
@@ -406,7 +445,7 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
406 445
407 446 /// Execute a single target: SSH to host, clone, build, upload artifact.
408 447 async fn execute_target(
409 - state: &AppState,
448 + ctx: &BuildCtx,
410 449 build: &DbBuild,
411 450 config: &DbBuildConfig,
412 451 host: &str,
@@ -418,17 +457,17 @@ async fn execute_target(
418 457 .ok_or_else(|| format!("unsupported target: {target}"))?;
419 458
420 459 // Look up repo for clone URL
421 - let repo = db::git_repos::get_repo_by_id(&state.db, config.repo_id)
460 + let repo = db::git_repos::get_repo_by_id(&ctx.db, config.repo_id)
422 461 .await
423 462 .map_err(|e| format!("failed to look up repo: {e}"))?
424 463 .ok_or("repo not found")?;
425 464
426 - let repo_owner = db::users::get_user_by_id(&state.db, repo.user_id)
465 + let repo_owner = db::users::get_user_by_id(&ctx.db, repo.user_id)
427 466 .await
428 467 .map_err(|e| format!("failed to look up repo owner: {e}"))?
429 468 .ok_or("repo owner not found")?;
430 469
431 - let git_root = state
470 + let git_root = ctx
432 471 .config
433 472 .build.git_repos_path
434 473 .as_deref()
@@ -472,7 +511,7 @@ async fn execute_target(
472 511 );
473 512
474 513 let log_msg = format!("[{target}] building on {host}...\n");
475 - let _ = append_log_bounded(state, build.id, &log_msg).await;
514 + let _ = append_log_bounded(ctx, build.id, &log_msg).await;
476 515
477 516 // Execute via SSH with timeout
478 517 let ssh_result = tokio::time::timeout(
@@ -494,7 +533,7 @@ async fn execute_target(
494 533 }
495 534 };
496 535
497 - let _ = append_log_bounded(state, build.id, &format!("[{target}] {}\n", output.trim())).await;
536 + let _ = append_log_bounded(ctx, build.id, &format!("[{target}] {}\n", output.trim())).await;
498 537
499 538 // SCP artifact back and upload to S3
500 539 let s3_key = crate::storage::S3Client::generate_ota_artifact_key(
@@ -545,8 +584,8 @@ async fn execute_target(
545 584 // `upload_multipart` reads the file in chunks and lets the S3 SDK do
546 585 // parallel part uploads, keeping memory bounded regardless of artifact
547 586 // size.
548 - let synckit_s3 = state
549 - .storage.synckit_s3
587 + let synckit_s3 = ctx
588 + .synckit_s3
550 589 .as_ref()
551 590 .ok_or("SyncKit storage not configured")?;
552 591
@@ -567,14 +606,14 @@ async fn execute_target(
567 606
568 607 if !signature.is_empty() {
569 608 let _ = append_log_bounded(
570 - state,
609 + ctx,
571 610 build.id,
572 611 &format!("[{target}] uploaded to {s3_key} (signed)\n"),
573 612 )
574 613 .await;
575 614 } else {
576 615 let _ = append_log_bounded(
577 - state,
616 + ctx,
578 617 build.id,
579 618 &format!("[{target}] uploaded to {s3_key}\n"),
580 619 )
@@ -736,23 +775,23 @@ async fn run_scp_download(
736 775 /// Probes `octet_length(log)` instead of fetching the whole row (the log
737 776 /// column tops out at 5 MiB and is read on every line append).
738 777 async fn append_log_bounded(
739 - state: &AppState,
778 + ctx: &BuildCtx,
740 779 build_id: db::BuildId,
741 780 line: &str,
742 781 ) -> crate::error::Result<()> {
743 782 const TRUNCATED: &str = "[log truncated]\n";
744 783 if let Some((current_len, already_truncated)) =
745 - db::builds::get_build_log_size(&state.db, build_id, TRUNCATED).await?
784 + db::builds::get_build_log_size(&ctx.db, build_id, TRUNCATED).await?
746 785 && (current_len as usize) + line.len() > BUILD_MAX_LOG_BYTES
747 786 {
748 787 if !already_truncated {
749 788 tracing::warn!(build_id = %build_id, "Build log exceeded {} bytes, truncating", BUILD_MAX_LOG_BYTES);
750 - db::builds::append_build_log(&state.db, build_id, TRUNCATED).await?;
789 + db::builds::append_build_log(&ctx.db, build_id, TRUNCATED).await?;
751 790 }
752 791 return Ok(());
753 792 }
754 793 let sanitized = strip_ansi_escapes(line);
755 - db::builds::append_build_log(&state.db, build_id, &sanitized).await
794 + db::builds::append_build_log(&ctx.db, build_id, &sanitized).await
756 795 }
757 796
758 797 /// Strip ANSI escape sequences (e.g. color codes) from build output before
M server/src/lib.rs +45 -5
@@ -239,20 +239,23 @@ impl FromRef<AppState> for Integrations {
239 239 }
240 240 }
241 241
242 - /// File-scanning slice: the scan pipeline plus the semaphore that bounds
243 - /// concurrent in-memory scans. Draws `scanner` from the top level and
244 - /// `scan_semaphore` from [`AppLimiters`].
242 + /// File-scanning slice: the scan pipeline that handlers enqueue work onto.
243 + ///
244 + /// Just `scanner`. The `scan_semaphore` that bounds concurrent in-memory scans
245 + /// is *not* here: it is consumed only inside the scan worker
246 + /// ([`scanning::worker::WorkerContext`], wired directly from [`AppLimiters`] at
247 + /// startup), never by a request handler. Every handler that extracts
248 + /// `State<Scanning>` reads only `scanner`, so bundling the worker-only semaphore
249 + /// in would be dead weight that misstates the handler's dependencies.
245 250 #[derive(Clone)]
246 251 pub struct Scanning {
247 252 pub scanner: Option<Arc<ScanPipeline>>,
248 - pub scan_semaphore: Arc<tokio::sync::Semaphore>,
249 253 }
250 254
251 255 impl FromRef<AppState> for Scanning {
252 256 fn from_ref(s: &AppState) -> Self {
253 257 Self {
254 258 scanner: s.scanner.clone(),
255 - scan_semaphore: s.limiters.scan_semaphore.clone(),
256 259 }
257 260 }
258 261 }
@@ -276,6 +279,38 @@ impl FromRef<AppState> for Sync {
276 279 }
277 280 }
278 281
282 + /// Depth of each per-(app,user) SSE broadcast channel. A subscriber that lags
283 + /// past this many un-received `seq` values gets a `Lagged` error and skips to
284 + /// the latest (the client pulls anyway), so the buffer only needs to cover a
285 + /// brief scheduling gap between a push and the subscriber's poll.
286 + const SYNC_NOTIFY_CHANNEL_DEPTH: usize = 16;
287 +
288 + impl Sync {
289 + /// Notify SSE subscribers of a push to `(app_id, user_id)`, carrying the new
290 + /// max `seq` so a device already at (or past) this cursor can skip a
291 + /// redundant pull. No-op when nobody is subscribed.
292 + pub fn notify_push(&self, app_id: SyncAppId, user_id: UserId, cursor: i64) {
293 + if let Some(sender) = self.sync_notify.get(&(app_id, user_id)) {
294 + let _ = sender.send(cursor); // Err = no subscribers, which is fine.
295 + }
296 + }
297 +
298 + /// Get-or-create the broadcast channel for `(app_id, user_id)` and return a
299 + /// fresh receiver. The channel is created lazily on first subscribe; the
300 + /// SSE connection guard prunes it once the last receiver drops.
301 + pub fn subscribe_channel(
302 + &self,
303 + app_id: SyncAppId,
304 + user_id: UserId,
305 + ) -> tokio::sync::broadcast::Receiver<i64> {
306 + self.sync_notify
307 + .entry((app_id, user_id))
308 + .or_insert_with(|| tokio::sync::broadcast::channel(SYNC_NOTIFY_CHANNEL_DEPTH).0)
309 + .value()
310 + .subscribe()
311 + }
312 + }
313 +
279 314 /// Git source-browser slice: the syntax highlighter (blob rendering) plus the
280 315 /// semaphore that bounds concurrent smart-HTTP clone/fetch child processes.
281 316 /// Draws `syntax` from the top level and `git_smart_http_semaphore` from
@@ -746,6 +781,11 @@ mod from_ref_slices {
746 781 _assert_from_ref::<Sync>();
747 782 _assert_from_ref::<Git>();
748 783 _assert_from_ref::<Ops>();
784 + // Background-worker projection views (hand-written FromRef): not
785 + // extracted as axum `State<T>`, but built from `AppState` at spawn/
786 + // dispatch so each worker states its deps instead of holding the struct.
787 + _assert_from_ref::<monitor::MonitorCtx>();
788 + _assert_from_ref::<build_runner::BuildCtx>();
749 789 }
750 790 }
751 791
M server/src/main.rs +4 -1
@@ -499,7 +499,10 @@ async fn main() {
499 499 );
500 500
501 501 // Start background health monitor and scheduler
502 - let _monitor_handle = makenotwork::monitor::spawn_monitor(state.clone(), shutdown_rx);
502 + let _monitor_handle = makenotwork::monitor::spawn_monitor(
503 + axum::extract::FromRef::from_ref(&state),
504 + shutdown_rx,
505 + );
503 506 let scheduler_shutdown_rx = shutdown_tx.subscribe();
504 507 let scheduler_handle = makenotwork::scheduler::spawn_scheduler(state.clone(), scheduler_shutdown_rx);
505 508
M server/src/monitor.rs +65 -29
@@ -11,9 +11,45 @@ use std::time::Instant;
11 11 use tokio::sync::watch;
12 12 use tokio::task::JoinHandle;
13 13
14 - use crate::AppState;
14 + use axum::extract::FromRef;
15 +
16 + use crate::config::Config;
15 17 use crate::constants;
16 18 use crate::db;
19 + use crate::email::EmailClient;
20 + use crate::wam_client::WamClient;
21 + use crate::{AppCaches, AppState, AppStorage};
22 +
23 + /// The slice of [`AppState`] the background health monitor needs: the DB pool it
24 + /// probes, the S3 backend it pings, the derived caches it reports/prunes, the
25 + /// mailer + config for alert emails, and the optional WAM client for tickets.
26 + ///
27 + /// Field names mirror [`AppState`] so the loop body reads `ctx.db`,
28 + /// `ctx.storage.s3`, `ctx.caches.session_cache`, etc. This is a projection view
29 + /// (like the handler slices in [`crate`]); [`spawn_monitor`] takes it instead of
30 + /// the whole `AppState`, so the monitor's dependencies are stated, not ambient.
31 + #[derive(Clone)]
32 + pub struct MonitorCtx {
33 + pub db: sqlx::PgPool,
34 + pub storage: AppStorage,
35 + pub caches: AppCaches,
36 + pub email: EmailClient,
37 + pub config: Config,
38 + pub wam: Option<WamClient>,
39 + }
40 +
41 + impl FromRef<AppState> for MonitorCtx {
42 + fn from_ref(s: &AppState) -> Self {
43 + Self {
44 + db: s.db.clone(),
45 + storage: s.storage.clone(),
46 + caches: s.caches.clone(),
47 + email: s.email.clone(),
48 + config: s.config.clone(),
49 + wam: s.wam.clone(),
50 + }
51 + }
52 + }
17 53
18 54 /// Overall service status.
19 55 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -43,17 +79,17 @@ pub struct HealthSnapshot {
43 79 }
44 80
45 81 /// Run a lightweight health probe (no HTTP self-call, no table counts).
46 - pub async fn run_health_check(state: &AppState) -> HealthSnapshot {
82 + pub async fn run_health_check(ctx: &MonitorCtx) -> HealthSnapshot {
47 83 let start = Instant::now();
48 84
49 85 // 1. Database: SELECT 1
50 86 let db_ok = sqlx::query_scalar::<_, i32>("SELECT 1")
51 - .fetch_one(&state.db)
87 + .fetch_one(&ctx.db)
52 88 .await
53 89 .is_ok();
54 90
55 91 // 2. S3 connectivity (skip if not configured)
56 - let s3_ok = match &state.storage.s3 {
92 + let s3_ok = match &ctx.storage.s3 {
57 93 Some(s3) => match s3.check_connectivity().await {
58 94 Ok(()) => true,
59 95 Err(e) => {
@@ -67,7 +103,7 @@ pub async fn run_health_check(state: &AppState) -> HealthSnapshot {
67 103 // 3. Sessions: probe the session table
68 104 let sessions_ok =
69 105 sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM tower_sessions.session)")
70 - .fetch_one(&state.db)
106 + .fetch_one(&ctx.db)
71 107 .await
72 108 .is_ok();
73 109
@@ -92,14 +128,14 @@ pub async fn run_health_check(state: &AppState) -> HealthSnapshot {
92 128 }
93 129
94 130 /// Spawn the background monitor loop. Drop `shutdown_tx` to stop it.
95 - pub fn spawn_monitor(state: AppState, shutdown_rx: watch::Receiver<()>) -> JoinHandle<()> {
131 + pub fn spawn_monitor(ctx: MonitorCtx, shutdown_rx: watch::Receiver<()>) -> JoinHandle<()> {
96 132 // Supervise the loop (Perf-S3): a panic in the loop body would otherwise kill
97 133 // the task and silently stop all health checks, alerting, and daily
98 134 // maintenance — with no alert, since the alerting lives in the same loop.
99 135 // Re-spawn on panic so monitoring survives; a clean shutdown ends supervision.
100 136 tokio::spawn(async move {
101 137 loop {
102 - match tokio::spawn(run_monitor_loop(state.clone(), shutdown_rx.clone())).await {
138 + match tokio::spawn(run_monitor_loop(ctx.clone(), shutdown_rx.clone())).await {
103 139 Ok(()) => return, // clean shutdown
104 140 Err(e) if e.is_panic() => {
105 141 tracing::error!(error = ?e, "health monitor loop panicked; restarting in 5s");
@@ -114,7 +150,7 @@ pub fn spawn_monitor(state: AppState, shutdown_rx: watch::Receiver<()>) -> JoinH
114 150 /// The health-monitor loop body. Extracted from `spawn_monitor` so a panic here is
115 151 /// caught at the task boundary by the supervisor (Perf-S3) instead of unwinding the
116 152 /// whole monitor task and silently halting background maintenance.
117 - async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>) {
153 + async fn run_monitor_loop(ctx: MonitorCtx, mut shutdown_rx: watch::Receiver<()>) {
118 154 let alert_email = std::env::var("ALERT_EMAIL").ok();
119 155 match &alert_email {
120 156 Some(email) => tracing::info!(alert_email = %email, "health monitor started"),
@@ -151,13 +187,13 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
151 187 }
152 188 }
153 189
154 - let snap = run_health_check(&state).await;
190 + let snap = run_health_check(&ctx).await;
155 191
156 192 // Heartbeat (Perf-S3): stamp this tick so an external watcher (the PoM
157 193 // health contract / dashboard) can detect a stalled monitor — a stale
158 194 // `health_monitor` last_ran_at means the loop died and the supervisor
159 195 // couldn't bring it back.
160 - if let Err(e) = db::scheduler_jobs::record_job_run(&state.db, "health_monitor", 0).await {
196 + if let Err(e) = db::scheduler_jobs::record_job_run(&ctx.db, "health_monitor", 0).await {
161 197 tracing::warn!(error = ?e, "failed to record health-monitor heartbeat");
162 198 }
163 199
@@ -168,8 +204,8 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
168 204 // tier caps — sub-second today but grows linearly with paying
169 205 // creator count, so we cache the result for 5 minutes rather
170 206 // than burning a multi-second query 2880×/day at 10k+ creators.
171 - crate::metrics::record_db_pool_stats(&state.db);
172 - crate::metrics::record_domain_cache_size(state.caches.domain_cache.len());
207 + crate::metrics::record_db_pool_stats(&ctx.db);
208 + crate::metrics::record_domain_cache_size(ctx.caches.domain_cache.len());
173 209 static STORAGE_FILL_LAST: std::sync::OnceLock<std::sync::Mutex<std::time::Instant>> =
174 210 std::sync::OnceLock::new();
175 211 const STORAGE_FILL_TTL: std::time::Duration = std::time::Duration::from_secs(300);
@@ -189,9 +225,9 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
189 225 }
190 226 };
191 227 if should_refresh {
192 - crate::metrics::record_storage_fill_stats(&state.db).await;
193 - crate::metrics::record_custom_pages_stats(&state.db).await;
194 - crate::metrics::record_scan_queue_stats(&state.db).await;
228 + crate::metrics::record_storage_fill_stats(&ctx.db).await;
229 + crate::metrics::record_custom_pages_stats(&ctx.db).await;
230 + crate::metrics::record_scan_queue_stats(&ctx.db).await;
195 231 }
196 232
197 233 // Log status changes. Skip the bootstrap None->Operational transition
@@ -237,7 +273,7 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
237 273 if cooldown_elapsed {
238 274 if let Some(ref to) = alert_email {
239 275 let (subject, body) = build_alert(previous_status, &snap);
240 - match state.email.send_alert(to, &subject, &body).await {
276 + match ctx.email.send_alert(to, &subject, &body).await {
241 277 Ok(()) => tracing::info!(recipient = %to, "alert email sent"),
242 278 Err(e) => tracing::error!(error = ?e, "failed to send alert email"),
243 279 }
@@ -250,10 +286,10 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
250 286 // into a ~100s single task holding the email-client clone.
251 287 // The subscriber list is bounded by the query's LIMIT.
252 288 {
253 - let pool = state.db.clone();
254 - let email_client = state.email.clone();
255 - let host_url = state.config.host_url.clone();
256 - let signing_secret = state.config.signing_secret.clone();
289 + let pool = ctx.db.clone();
290 + let email_client = ctx.email.clone();
291 + let host_url = ctx.config.host_url.clone();
292 + let signing_secret = ctx.config.signing_secret.clone();
257 293 let current_status = snap.status.as_str().to_string();
258 294 let prev_status = previous_status
259 295 .map_or("unknown", |s| s.as_str())
@@ -305,7 +341,7 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
305 341
306 342 // Create WAM ticket on degradation/error transitions
307 343 if snap.status != MonitorStatus::Operational
308 - && let Some(ref wam) = state.wam
344 + && let Some(ref wam) = ctx.wam
309 345 {
310 346 let priority = match snap.status {
311 347 MonitorStatus::Error => "critical",
@@ -330,8 +366,8 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
330 366 // threshold (e.g. each scheduler tick briefly maxes the pool)
331 367 // spams a ticket every ALERT_COOLDOWN_SECS window.
332 368 {
333 - let pool_size = state.db.size();
334 - let pool_idle = state.db.num_idle() as u32;
369 + let pool_size = ctx.db.size();
370 + let pool_idle = ctx.db.num_idle() as u32;
335 371 let active = pool_size.saturating_sub(pool_idle);
336 372 let pct = (active * 100).checked_div(pool_size).unwrap_or(0);
337 373 let high = 80u32;
@@ -353,7 +389,7 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
353 389 .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
354 390 if cooldown_ok
355 391 && pool_pressure_armed
356 - && let Some(ref wam) = state.wam
392 + && let Some(ref wam) = ctx.wam
357 393 {
358 394 let title = format!("DB pool pressure: {active}/{pool_size} active");
359 395 wam.create_ticket(&title, None, "high", "db-pool-pressure", None)
@@ -380,7 +416,7 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
380 416 // — a 30s refresh on connection-utilization is the right tradeoff.
381 417 // The probe itself is a single-row query against a system view;
382 418 // its cost is on the order of microseconds.
383 - if let Some((active, max_conn)) = crate::metrics::record_pg_stat_activity(&state.db).await {
419 + if let Some((active, max_conn)) = crate::metrics::record_pg_stat_activity(&ctx.db).await {
384 420 let pct = active * 100 / max_conn;
385 421 if pct > 80 {
386 422 tracing::warn!(
@@ -391,7 +427,7 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
391 427 );
392 428 let cooldown_ok = last_pg_activity_alert_at
393 429 .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
394 - if cooldown_ok && let Some(ref wam) = state.wam {
430 + if cooldown_ok && let Some(ref wam) = ctx.wam {
395 431 let title = format!(
396 432 "Postgres saturation: {active}/{max_conn} client backends ({pct}%)"
397 433 );
@@ -419,7 +455,7 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
419 455
420 456 // Persist snapshot (best-effort)
421 457 if let Err(e) = db::monitor::insert_health_history(
422 - &state.db,
458 + &ctx.db,
423 459 snap.status.as_str(),
424 460 snap.db_ok,
425 461 snap.s3_ok,
@@ -434,8 +470,8 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
434 470
435 471 // Prune expired session cache entries every cycle
436 472 let cache_ttl = std::time::Duration::from_secs(constants::SESSION_TOUCH_CACHE_SECS);
437 - state
438 - .caches.session_cache
473 + ctx.caches
474 + .session_cache
439 475 .retain(|_, validated_at| validated_at.elapsed() < cache_ttl);
440 476
441 477 // Daily prune/compaction (health history, sync log, OAuth cleanup) used
M server/src/routes/synckit/subscribe.rs +1 -9
@@ -114,8 +114,6 @@ pub(super) async fn sync_subscribe(
114 114 ));
115 115 }
116 116
117 - let key = (sync_user.app_id, sync_user.user_id);
118 -
119 117 let guard = SseConnectionGuard {
120 118 sse_connections: sync.sse_connections.clone(),
121 119 sync_notify: sync.sync_notify.clone(),
@@ -124,13 +122,7 @@ pub(super) async fn sync_subscribe(
124 122 };
125 123
126 124 // Get or create the broadcast channel for this app+user
127 - let rx = {
128 - let entry = sync.sync_notify.entry(key).or_insert_with(|| {
129 - let (tx, _) = tokio::sync::broadcast::channel(16);
130 - tx
131 - });
132 - entry.value().subscribe()
133 - };
125 + let rx = sync.subscribe_channel(sync_user.app_id, sync_user.user_id);
134 126
135 127 let stream = BroadcastStream::new(rx).filter_map(|result| match result {
136 128 // Carry the new max seq so a client already at this cursor can skip the
M server/src/routes/synckit/sync.rs +1 -3
@@ -119,9 +119,7 @@ pub(super) async fn sync_push(
119 119 // Notify SSE subscribers, carrying the new max seq so a device already at
120 120 // (or past) this cursor can skip a redundant pull — avoids the thundering
121 121 // herd where every subscriber pulls on every push.
122 - if let Some(sender) = sync.sync_notify.get(&(app_id, user_id)) {
123 - let _ = sender.send(cursor); // Ignore errors (no subscribers = ok)
124 - }
122 + sync.notify_push(app_id, user_id, cursor);
125 123
126 124 Ok(Json(PushResponse { cursor }).into_response())
127 125 }
M server/src/scheduler/mod.rs +7 -2
@@ -200,8 +200,13 @@ pub fn spawn_scheduler(
200 200 // Send onboarding drip emails
201 201 announcements::send_onboarding_emails(&state.db, &state.email, &state.config).await;
202 202
203 - // Dispatch pending builds
204 - crate::build_runner::dispatch_pending_build(&state).await;
203 + // Dispatch pending builds. Project the runner's slice from the
204 + // scheduler's `AppState` at the call site (same seam as the stripe
205 + // dispatcher below) so the runner declares only what it needs.
206 + crate::build_runner::dispatch_pending_build(
207 + &axum::extract::FromRef::from_ref(&state),
208 + )
209 + .await;
205 210
206 211 // Publish scheduled blog posts — same off-lock pattern as items.
207 212 match db::blog_posts::publish_scheduled_blog_posts(&state.db).await {