Skip to main content

max / makenotwork

Report which SyncKit versions are in the field SyncKit is client-side only, so there is no deployed version to poll. The server now aggregates the SDK version off each sync request's User-Agent; this reads that aggregate on a per-target interval and records it, so target_status answers which SyncKit is actually out there. Informational by design. The distribution cannot fail a target: a user still on an old SDK is a fact, not an incident, and no threshold would make a version mix wrong. A readout PoM cannot take is degraded, because otherwise a token rotation that misses PoM leaves it reporting green while knowing nothing. Not failed: whether the platform is up belongs to the health condition. So the result type carries no status field, and the task fires no alerts. Reuses alerts.alerts_ingest_token rather than taking a token of its own, so a rotation has one place to miss instead of two. Without it the check logs and does not spawn.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-30 00:55 UTC
Signed with PGP, not checked
Commit: c95d98ca815421e378ec829a79e3066a594cf6a2
Parent: 1243c05
17 files changed, +875 insertions, -20 deletions
M pom/README.md +1
@@ -57,6 +57,7 @@
57 57 - **TLS**: hosts to probe for certificate expiry warnings
58 58 - **Tests**: SSH targets and commands for remote test suite execution
59 59 - **Repo**: per-target local checkout (`path`, optional `subdir`) that `pom versions` counts the live build against. Optional; without it the commits-behind column is blank
60 + - **SyncKit fleet**: per-target readout of which SyncKit SDK versions are syncing (`base_url`, `window_days`, intervals). Reuses the alerts ingest token; informational, so an old version in the field never degrades a target
60 61
61 62 ## Module Overview
62 63
M pom/deploy/pom-hetzner.toml +12
@@ -68,6 +68,18 @@
68 68 interval_secs = 300
69 69 timeout_secs = 10
70 70
71 + [targets.mnw.synckit_fleet]
72 + # Which SyncKit SDK versions are actually syncing. SyncKit is client-side
73 + # only, so there is no deployed version to poll; the server aggregates the
74 + # version off each sync request's User-Agent and this reads that. Authed
75 + # with alerts.alerts_ingest_token, so the check does not spawn without one.
76 + # Informational: an old version in the field never degrades the target, only
77 + # a readout PoM cannot take does. Localhost for the same reason as above.
78 + base_url = "http://127.0.0.1:3000"
79 + window_days = 30
80 + interval_secs = 3600
81 + timeout_secs = 10
82 +
71 83 [targets.mnw-cli]
72 84 label = "MNW CLI SSH Server"
73 85
M pom/src/api.rs +17
@@ -642,6 +642,22 @@
642 642 error: s.error,
643 643 });
644 644
645 + let synckit_fleet = db::get_latest_synckit_fleet_check(pool, &name)
646 + .await
647 + .ok()
648 + .flatten()
649 + .map(|f| crate::status::SyncKitFleetView {
650 + versions: f
651 + .version_list()
652 + .into_iter()
653 + .map(|v| (v.client_version, v.devices))
654 + .collect(),
655 + devices: f.devices,
656 + window_days: f.window_days,
657 + checked_at: f.checked_at,
658 + error: f.error,
659 + });
660 +
645 661 // Tests: the latest run plus PoM's staleness verdict, sourced exactly as
646 662 // build_target_status does (version at test time vs current version).
647 663 let tests = if let Some(tests_config) = &target_config.tests {
@@ -725,6 +741,7 @@
725 741 backups,
726 742 scan_pipeline,
727 743 systemd,
744 + synckit_fleet,
728 745 tests,
729 746 dns,
730 747 cors,
M pom/src/config.rs +38
@@ -270,6 +270,9 @@
270 270 /// Scan-pipeline health check against `<base_url>/admin/uploads/health.json`.
271 271 /// `None` disables.
272 272 pub scan_pipeline: Option<ScanPipelineConfig>,
273 + /// SyncKit field-version readout against
274 + /// `<base_url>/api/internal/synckit/client-versions`. `None` disables.
275 + pub synckit_fleet: Option<SyncKitFleetConfig>,
273 276 /// Local checkout of the code this target runs, used by `pom versions` to
274 277 /// count how far the live build is behind. `None` leaves that column blank.
275 278 pub repo: Option<RepoConfig>,
@@ -359,6 +362,41 @@
359 362 10
360 363 }
361 364
365 + /// SyncKit field-version readout for a target.
366 + ///
367 + /// Carries no token of its own: the endpoint is authed with the same
368 + /// `alerts.alerts_ingest_token` PoM already holds for pushing alerts to MNW, and
369 + /// copying that secret into a second config block would mean a rotation has two
370 + /// places to miss. Without an `alerts_ingest_token`, this check does not spawn.
371 + #[derive(Debug, Clone, Deserialize)]
372 + pub struct SyncKitFleetConfig {
373 + /// Base URL of the makenotwork instance (e.g. "https://makenot.work").
374 + pub base_url: String,
375 + /// Activity window handed to the server as `?days=`. Defaults to 30, the
376 + /// server's own default; it clamps anything outside 1..=365.
377 + #[serde(default = "default_synckit_fleet_window_days")]
378 + pub window_days: u32,
379 + /// Check interval. Defaults to 3600s.
380 + #[serde(default = "default_synckit_fleet_interval")]
381 + pub interval_secs: u64,
382 + /// HTTP request timeout. Defaults to 10s.
383 + #[serde(default = "default_synckit_fleet_timeout")]
384 + pub timeout_secs: u64,
385 + }
386 +
387 + fn default_synckit_fleet_window_days() -> u32 {
388 + 30
389 + }
390 + fn default_synckit_fleet_interval() -> u64 {
391 + // Hourly. A fleet's version mix moves when users update, which is days-scale;
392 + // polling it as often as a liveness check would buy nothing and put a
393 + // GROUP BY over sync_devices on a five-minute timer.
394 + 3600
395 + }
396 + fn default_synckit_fleet_timeout() -> u64 {
397 + 10
398 + }
399 +
362 400 #[derive(Debug, Clone, Deserialize)]
363 401 pub struct DnsRecord {
364 402 /// Hostname to resolve (e.g. "makenot.work").
M pom/src/display.rs +6 -3
@@ -345,7 +345,7 @@
345 345 /// Format prune results for CLI display.
346 346 pub fn format_prune(result: &PruneResult, days: i64) -> String {
347 347 format!(
348 - "Pruned {} health checks, {} test runs, {} test details, {} peer heartbeats, {} alerts, {} TLS checks, {} incidents, {} route checks, {} DNS checks, {} WHOIS checks, {} backup checks, {} systemd checks older than {} days.\n",
348 + "Pruned {} health checks, {} test runs, {} test details, {} peer heartbeats, {} alerts, {} TLS checks, {} incidents, {} route checks, {} DNS checks, {} WHOIS checks, {} backup checks, {} systemd checks, {} synckit fleet checks older than {} days.\n",
349 349 result.health,
350 350 result.tests,
351 351 result.test_details,
@@ -358,6 +358,7 @@
358 358 result.whois,
359 359 result.backups,
360 360 result.systemd,
361 + result.synckit_fleet,
361 362 days
362 363 )
363 364 }
@@ -1230,11 +1231,12 @@
1230 1231 whois: 2,
1231 1232 backups: 1,
1232 1233 systemd: 6,
1234 + synckit_fleet: 7,
1233 1235 };
1234 1236 let out = format_prune(&result, 30);
1235 1237 assert_eq!(
1236 1238 out,
1237 - "Pruned 5 health checks, 3 test runs, 15 test details, 10 peer heartbeats, 2 alerts, 1 TLS checks, 4 incidents, 0 route checks, 8 DNS checks, 2 WHOIS checks, 1 backup checks, 6 systemd checks older than 30 days.\n"
1239 + "Pruned 5 health checks, 3 test runs, 15 test details, 10 peer heartbeats, 2 alerts, 1 TLS checks, 4 incidents, 0 route checks, 8 DNS checks, 2 WHOIS checks, 1 backup checks, 6 systemd checks, 7 synckit fleet checks older than 30 days.\n"
1238 1240 );
1239 1241 }
1240 1242
@@ -1253,9 +1255,10 @@
1253 1255 whois: 0,
1254 1256 backups: 0,
1255 1257 systemd: 0,
1258 + synckit_fleet: 0,
1256 1259 };
1257 1260 let out = format_prune(&result, 7);
1258 - assert!(out.contains("Pruned 0 health checks, 0 test runs, 0 test details, 0 peer heartbeats, 0 alerts, 0 TLS checks, 0 incidents, 0 route checks, 0 DNS checks, 0 WHOIS checks, 0 backup checks, 0 systemd checks older than 7 days."));
1261 + assert!(out.contains("Pruned 0 health checks, 0 test runs, 0 test details, 0 peer heartbeats, 0 alerts, 0 TLS checks, 0 incidents, 0 route checks, 0 DNS checks, 0 WHOIS checks, 0 backup checks, 0 systemd checks, 0 synckit fleet checks older than 7 days."));
1259 1262 }
1260 1263
1261 1264 // format_mesh
M pom/src/status.rs +210
@@ -80,6 +80,8 @@
80 80 pub scan_pipeline: Option<ScanView>,
81 81 /// Latest local systemd daemon-health check. `None` if not monitored.
82 82 pub systemd: Option<SystemdView>,
83 + /// Latest SyncKit field-version readout. `None` if not monitored.
84 + pub synckit_fleet: Option<SyncKitFleetView>,
83 85 /// Latest test run and PoM's staleness verdict. `None` if the target has no
84 86 /// test config.
85 87 pub tests: Option<TestsView>,
@@ -144,6 +146,21 @@
144 146 pub error: Option<String>,
145 147 }
146 148
149 + /// A SyncKit field-version readout. No status field: the reading cannot be
150 + /// unhealthy, only unavailable. See [`crate::checks::synckit_fleet`].
151 + pub(crate) struct SyncKitFleetView {
152 + /// Devices that synced within the server's activity window.
153 + pub devices: i64,
154 + /// The window that count covers, in days.
155 + pub window_days: i64,
156 + /// Version and device count, most devices first. `None` version means
157 + /// "syncing, version unknown".
158 + pub versions: Vec<(Option<String>, i64)>,
159 + pub checked_at: String,
160 + /// Why the readout could not be taken. `None` on a successful reading.
161 + pub error: Option<String>,
162 + }
163 +
147 164 pub(crate) struct TestsView {
148 165 /// Whether any run has ever been recorded. Separates "no evidence"
149 166 /// (pending) from a run that failed.
@@ -221,6 +238,9 @@
221 238 if let Some(sd) = &target.systemd {
222 239 conditions.push(systemd_condition(sd));
223 240 }
241 + if let Some(fleet) = &target.synckit_fleet {
242 + conditions.push(synckit_fleet_condition(fleet));
243 + }
224 244 if let Some(tests) = &target.tests {
225 245 conditions.push(tests_condition(tests));
226 246 }
@@ -463,6 +483,59 @@
463 483 }
464 484 }
465 485
486 + /// The SyncKit field-version readout as a condition. Only two outcomes, and
487 + /// neither of them depends on which versions came back: `ok` when the reading was
488 + /// taken (the detail is the distribution), `degraded` when it could not be. No
489 + /// version mix can fail this condition, because a user still on an old SDK is a
490 + /// fact about the world rather than an incident PoM can page anyone about. A
491 + /// broken readout *is* worth yellow: PoM has stopped being able to answer "which
492 + /// SyncKit is in the field" and would otherwise go on reporting green while
493 + /// knowing nothing. It stays out of red because whether the platform is up is the
494 + /// `health` condition's job.
495 + fn synckit_fleet_condition(fleet: &SyncKitFleetView) -> Condition {
496 + let since = parse_instant(&fleet.checked_at);
497 + let (status, detail) = match &fleet.error {
498 + Some(error) => (
499 + Status::Degraded,
500 + format!("fleet readout unavailable: {error}"),
501 + ),
502 + None if fleet.devices == 0 => (
503 + Status::Ok,
504 + format!("no devices synced in {}d", fleet.window_days),
505 + ),
506 + None => (
507 + Status::Ok,
508 + format!(
509 + "{} device{} in {}d: {}",
510 + fleet.devices,
511 + plural(fleet.devices as usize),
512 + fleet.window_days,
513 + fleet_summary(&fleet.versions)
514 + ),
515 + ),
516 + };
517 + Condition {
518 + condition_type: "synckit_fleet".into(),
519 + status,
520 + since,
521 + detail: Some(detail),
522 + }
523 + }
524 +
525 + /// The distribution as one line, e.g. `0.6.0 x12, unknown x3`. An absent version
526 + /// is spelled out rather than dropped: a fleet that is half unknown is a real
527 + /// reading and hiding the unknowns would overstate what PoM knows.
528 + fn fleet_summary(versions: &[(Option<String>, i64)]) -> String {
529 + if versions.is_empty() {
530 + return "no versions reported".into();
531 + }
532 + versions
533 + .iter()
534 + .map(|(version, devices)| format!("{} x{devices}", version.as_deref().unwrap_or("unknown")))
535 + .collect::<Vec<_>>()
536 + .join(", ")
537 + }
538 +
466 539 /// Test state as a condition. A never-run target is `pending`: evidence of
467 540 /// nothing, like one never health-checked. A failing or stale run is `degraded`,
468 541 /// not `failed`: a red suite is a real regression signal, but the running
@@ -626,6 +699,28 @@
626 699 },
627 700 ));
628 701 }
702 + // The fleet distribution is a field, not just a condition detail: it is the
703 + // answer to a question an operator asks on purpose ("which SyncKit is out
704 + // there"), not a by-product of something being wrong. A successful readout of
705 + // zero devices still reports, so the number is visibly zero rather than
706 + // missing.
707 + if let Some(fleet) = &target.synckit_fleet
708 + && fleet.error.is_none()
709 + {
710 + fields.push(Field::new(
711 + "synckit fleet",
712 + Value::Text {
713 + value: fleet_summary(&fleet.versions),
714 + },
715 + ));
716 + fields.push(Field::new(
717 + "synckit devices",
718 + Value::Quantity {
719 + value: fleet.devices as f64,
720 + unit: None,
721 + },
722 + ));
723 + }
629 724 if let Some(checked) = target
630 725 .health
631 726 .as_ref()
@@ -675,6 +770,7 @@
675 770 backups: Vec::new(),
676 771 scan_pipeline: None,
677 772 systemd: None,
773 + synckit_fleet: None,
678 774 tests: None,
679 775 dns: None,
680 776 cors: None,
@@ -1106,6 +1202,120 @@
1106 1202 assert!(scan.detail.as_deref().unwrap().contains("502 Bad Gateway"));
1107 1203 }
1108 1204
1205 + fn fleet_view(versions: Vec<(Option<&str>, i64)>) -> SyncKitFleetView {
1206 + SyncKitFleetView {
1207 + devices: versions.iter().map(|(_, d)| d).sum(),
1208 + window_days: 30,
1209 + versions: versions
1210 + .into_iter()
1211 + .map(|(v, d)| (v.map(str::to_string), d))
1212 + .collect(),
1213 + checked_at: checked_at(),
1214 + error: None,
1215 + }
1216 + }
1217 +
1218 + #[test]
1219 + fn a_fleet_readout_reports_the_distribution_without_degrading() {
1220 + let mut t = healthy("mnw");
1221 + t.synckit_fleet = Some(fleet_view(vec![(Some("0.6.0"), 12), (None, 3)]));
1222 + let p = payload(&[t], now());
1223 +
1224 + let n = node(&p, "target:mnw");
1225 + assert_eq!(n.status, Status::Ok);
1226 + let fleet = n
1227 + .conditions
1228 + .iter()
1229 + .find(|c| c.condition_type == "synckit_fleet")
1230 + .unwrap();
1231 + assert_eq!(fleet.status, Status::Ok);
1232 + let detail = fleet.detail.as_deref().unwrap();
1233 + assert!(detail.contains("15 devices in 30d"), "got {detail}");
1234 + assert!(detail.contains("0.6.0 x12"), "got {detail}");
1235 + assert!(detail.contains("unknown x3"), "got {detail}");
1236 + }
1237 +
1238 + #[test]
1239 + fn an_ancient_version_in_the_field_does_not_degrade_the_target() {
1240 + // The whole design call: version age is a fact, not an incident. A fleet
1241 + // entirely on a year-old SDK must still read green.
1242 + let mut t = healthy("mnw");
1243 + t.synckit_fleet = Some(fleet_view(vec![(Some("0.1.0"), 200)]));
1244 + let p = payload(&[t], now());
1245 +
1246 + let n = node(&p, "target:mnw");
1247 + assert_eq!(n.status, Status::Ok);
1248 + }
1249 +
1250 + #[test]
1251 + fn an_empty_fleet_is_ok_and_says_so() {
1252 + let mut t = healthy("mnw");
1253 + t.synckit_fleet = Some(fleet_view(vec![]));
1254 + let p = payload(&[t], now());
1255 +
1256 + let n = node(&p, "target:mnw");
1257 + assert_eq!(n.status, Status::Ok);
1258 + let fleet = n
1259 + .conditions
1260 + .iter()
1261 + .find(|c| c.condition_type == "synckit_fleet")
1262 + .unwrap();
1263 + assert!(
1264 + fleet
1265 + .detail
1266 + .as_deref()
1267 + .unwrap()
1268 + .contains("no devices synced in 30d")
1269 + );
1270 + }
1271 +
1272 + #[test]
1273 + fn an_unavailable_fleet_readout_degrades_but_never_fails() {
1274 + let mut t = healthy("mnw");
1275 + let mut fleet = fleet_view(vec![]);
1276 + fleet.error = Some("HTTP 401 (alerts ingest token rejected)".into());
1277 + t.synckit_fleet = Some(fleet);
1278 + let p = payload(&[t], now());
1279 +
1280 + let n = node(&p, "target:mnw");
1281 + assert_eq!(
1282 + n.status,
1283 + Status::Degraded,
1284 + "a readout PoM cannot take is yellow, not red: health owns whether MNW is up"
1285 + );
1286 + let fleet = n
1287 + .conditions
1288 + .iter()
1289 + .find(|c| c.condition_type == "synckit_fleet")
1290 + .unwrap();
1291 + assert!(fleet.detail.as_deref().unwrap().contains("401"));
1292 + }
1293 +
1294 + #[test]
1295 + fn a_successful_readout_becomes_target_fields() {
1296 + let mut t = healthy("mnw");
1297 + t.synckit_fleet = Some(fleet_view(vec![(Some("0.6.0"), 12)]));
1298 + let p = payload(&[t], now());
1299 +
1300 + let n = node(&p, "target:mnw");
1301 + assert!(n.fields.iter().any(|f| f.label == "synckit fleet"));
1302 + assert!(n.fields.iter().any(|f| f.label == "synckit devices"));
1303 + }
1304 +
1305 + #[test]
1306 + fn an_unavailable_readout_contributes_no_fields() {
1307 + // A blank column beats a field reading "0 devices" that is really "PoM
1308 + // could not ask".
1309 + let mut t = healthy("mnw");
1310 + let mut fleet = fleet_view(vec![]);
1311 + fleet.error = Some("request: timed out".into());
1312 + t.synckit_fleet = Some(fleet);
1313 + let p = payload(&[t], now());
1314 +
1315 + let n = node(&p, "target:mnw");
1316 + assert!(!n.fields.iter().any(|f| f.label.starts_with("synckit")));
1317 + }
1318 +
1109 1319 fn tests_view(passed: bool, stale: bool) -> TestsView {
1110 1320 TestsView {
1111 1321 ran: true,
M pom/src/types.rs +35
@@ -507,6 +507,41 @@
507 507 pub error: Option<String>,
508 508 }
509 509
510 + /// One SDK version and how many devices are on it.
511 + #[derive(Debug, Clone, Serialize, Deserialize)]
512 + pub struct SyncKitVersionSnapshot {
513 + /// `None` means "syncing, version unknown": a client from before the SDK
514 + /// sent a User-Agent, or a value the server would not vouch for. Not a
515 + /// missing reading, an answer.
516 + pub client_version: Option<String>,
517 + pub devices: i64,
518 + /// Most recent sync from any device on this version, RFC 3339.
519 + pub last_seen_at: Option<String>,
520 + }
521 +
522 + /// Which SyncKit SDK versions are actually syncing against a target.
523 + ///
524 + /// Carries no `status` field, unlike every other check result, because there is
525 + /// no bad state to be in: see [`crate::checks::synckit_fleet`]. `error` being set
526 + /// is the only thing that can be wrong, and it means PoM could not take the
527 + /// reading, not that the fleet is unhealthy.
528 + #[derive(Debug, Clone, Serialize, Deserialize)]
529 + pub struct SyncKitFleetCheckResult {
530 + /// Config key identifying the monitored target.
531 + pub target: String,
532 + /// Activity window the server aggregated over, echoed back from its reply.
533 + pub window_days: i32,
534 + /// Devices that synced within the window, across all versions.
535 + pub devices: i64,
536 + /// Distribution, server-ordered (most devices first).
537 + pub versions: Vec<SyncKitVersionSnapshot>,
538 + /// When this check was performed, in RFC 3339 format (UTC).
539 + pub checked_at: String,
540 + /// Why the readout could not be produced. `None` on a successful reading,
541 + /// including a reading of zero devices.
542 + pub error: Option<String>,
543 + }
544 +
510 545 #[derive(Debug, Clone, Serialize, Deserialize)]
511 546 pub struct SystemdUnitSnapshot {
512 547 /// Unit name (e.g. "sandod.service").
M pom/tests/integration.rs +124 -13
@@ -413,7 +413,7 @@
413 413 // A fresh in-memory DB should run all migrations and reach the latest version.
414 414 let pool = db::connect_in_memory().await.unwrap();
415 415 let version = db::get_schema_version(&pool).await.unwrap();
416 - assert_eq!(version, 12);
416 + assert_eq!(version, 13);
417 417
418 418 // Verify the schema_version table has entries for each migration
419 419 let rows = sqlx::query_as::<_, (i64, String)>(
@@ -422,7 +422,7 @@
422 422 .fetch_all(&pool)
423 423 .await
424 424 .unwrap();
425 - assert_eq!(rows.len(), 12);
425 + assert_eq!(rows.len(), 13);
426 426 assert_eq!(rows[0].0, 1);
427 427 assert_eq!(rows[0].1, "initial schema");
428 428 assert_eq!(rows[1].0, 2);
@@ -447,6 +447,8 @@
447 447 assert_eq!(rows[10].1, "add scan_pipeline_checks table");
448 448 assert_eq!(rows[11].0, 12);
449 449 assert_eq!(rows[11].1, "add systemd_checks table");
450 + assert_eq!(rows[12].0, 13);
451 + assert_eq!(rows[12].1, "add synckit_fleet_checks table");
450 452
451 453 // Verify actual tables were created by inserting data
452 454 let snapshot = HealthSnapshot {
@@ -466,18 +468,18 @@
466 468 async fn migration_already_current_is_idempotent() {
467 469 // Running migrations on an already-migrated DB should be a no-op.
468 470 let pool = db::connect_in_memory().await.unwrap();
469 - assert_eq!(db::get_schema_version(&pool).await.unwrap(), 12);
471 + assert_eq!(db::get_schema_version(&pool).await.unwrap(), 13);
470 472
471 473 // Run migrations again
472 474 db::run_migrations(&pool).await.unwrap();
473 - assert_eq!(db::get_schema_version(&pool).await.unwrap(), 12);
475 + assert_eq!(db::get_schema_version(&pool).await.unwrap(), 13);
474 476
475 - // schema_version should still have exactly twelve entries (not duplicated)
477 + // schema_version should still have exactly thirteen entries (not duplicated)
476 478 let count = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM schema_version")
477 479 .fetch_one(&pool)
478 480 .await
479 481 .unwrap();
480 - assert_eq!(count.0, 12);
482 + assert_eq!(count.0, 13);
481 483 }
482 484
483 485 #[tokio::test]
@@ -536,8 +538,8 @@
536 538 // Now run migrations, should detect existing tables, stamp as v1, then run v2+v3+v4+v5+v6
537 539 db::run_migrations(&pool).await.unwrap();
538 540
539 - // Version should be 12 (stamped v1 + ran v2..v12)
540 - assert_eq!(db::get_schema_version(&pool).await.unwrap(), 12);
541 + // Version should be 13 (stamped v1 + ran v2..v13)
542 + assert_eq!(db::get_schema_version(&pool).await.unwrap(), 13);
541 543
542 544 // Description should indicate pre-existing
543 545 let row =
@@ -858,7 +860,7 @@
858 860 async fn migration_v2_creates_alerts_table() {
859 861 let pool = db::connect_in_memory().await.unwrap();
860 862 let version = db::get_schema_version(&pool).await.unwrap();
861 - assert_eq!(version, 12);
863 + assert_eq!(version, 13);
862 864
863 865 // Verify alerts table exists by inserting
864 866 let id = db::insert_alert(
@@ -944,7 +946,7 @@
944 946 async fn migration_v3_creates_tls_checks_table() {
945 947 let pool = db::connect_in_memory().await.unwrap();
946 948 let version = db::get_schema_version(&pool).await.unwrap();
947 - assert_eq!(version, 12);
949 + assert_eq!(version, 13);
948 950
949 951 // Verify tls_checks table exists by inserting
950 952 let status = pom::types::TlsStatus {
@@ -1182,7 +1184,7 @@
1182 1184 async fn migration_v4_creates_incidents_table() {
1183 1185 let pool = db::connect_in_memory().await.unwrap();
1184 1186 let version = db::get_schema_version(&pool).await.unwrap();
1185 - assert_eq!(version, 12);
1187 + assert_eq!(version, 13);
1186 1188
1187 1189 // Verify incidents table exists by inserting
1188 1190 let id = db::insert_incident(&pool, "mnw", "operational", "degraded")
@@ -1322,7 +1324,7 @@
1322 1324 async fn migration_v5_creates_route_checks_table() {
1323 1325 let pool = db::connect_in_memory().await.unwrap();
1324 1326 let version = db::get_schema_version(&pool).await.unwrap();
1325 - assert_eq!(version, 12);
1327 + assert_eq!(version, 13);
1326 1328
1327 1329 // Verify route_checks table exists by inserting
1328 1330 let result = pom::checks::routes::RouteCheckResult {
@@ -2471,7 +2473,7 @@
2471 2473 async fn migration_v6_creates_dns_and_whois_tables() {
2472 2474 let pool = db::connect_in_memory().await.unwrap();
2473 2475 let version = db::get_schema_version(&pool).await.unwrap();
2474 - assert_eq!(version, 12);
2476 + assert_eq!(version, 13);
2475 2477
2476 2478 // Verify dns_checks table exists
2477 2479 let dns_result = DnsCheckResult {
@@ -2648,6 +2650,115 @@
2648 2650 assert_eq!(row.days_remaining, Some(261));
2649 2651 }
2650 2652
2653 + #[tokio::test]
2654 + async fn synckit_fleet_check_insert_and_query() {
2655 + let pool = db::connect_in_memory().await.unwrap();
2656 +
2657 + let result = SyncKitFleetCheckResult {
2658 + target: "mnw".to_string(),
2659 + window_days: 30,
2660 + devices: 15,
2661 + versions: vec![
2662 + SyncKitVersionSnapshot {
2663 + client_version: Some("0.6.0".to_string()),
2664 + devices: 12,
2665 + last_seen_at: Some("2026-07-29T12:00:00Z".to_string()),
2666 + },
2667 + SyncKitVersionSnapshot {
2668 + client_version: None,
2669 + devices: 3,
2670 + last_seen_at: Some("2026-07-20T09:00:00Z".to_string()),
2671 + },
2672 + ],
2673 + checked_at: "2026-07-29T13:00:00Z".to_string(),
2674 + error: None,
2675 + };
2676 + db::insert_synckit_fleet_check(&pool, &result)
2677 + .await
2678 + .unwrap();
2679 +
2680 + let row = db::get_latest_synckit_fleet_check(&pool, "mnw")
2681 + .await
2682 + .unwrap()
2683 + .expect("readout should be stored");
2684 + assert_eq!(row.devices, 15);
2685 + assert_eq!(row.window_days, 30);
2686 + assert!(row.error.is_none());
2687 +
2688 + let versions = row.version_list();
2689 + assert_eq!(versions.len(), 2);
2690 + assert_eq!(versions[0].client_version.as_deref(), Some("0.6.0"));
2691 + assert_eq!(versions[0].devices, 12);
2692 + // The unknown bucket must survive the JSON round trip as null, not as the
2693 + // string "unknown": the readout distinguishes "no version reported" from a
2694 + // client that literally called itself that.
2695 + assert!(versions[1].client_version.is_none());
2696 + assert_eq!(versions[1].devices, 3);
2697 + }
2698 +
2699 + #[tokio::test]
2700 + async fn synckit_fleet_check_returns_latest() {
2701 + let pool = db::connect_in_memory().await.unwrap();
2702 +
2703 + for (devices, checked_at) in [(3, "2026-07-28T00:00:00Z"), (9, "2026-07-29T00:00:00Z")] {
2704 + let result = SyncKitFleetCheckResult {
2705 + target: "mnw".to_string(),
2706 + window_days: 30,
2707 + devices,
2708 + versions: Vec::new(),
2709 + checked_at: checked_at.to_string(),
2710 + error: None,
2711 + };
2712 + db::insert_synckit_fleet_check(&pool, &result)
2713 + .await
2714 + .unwrap();
2715 + }
2716 +
2717 + // History is kept rather than overwritten, so the read must be the newest row.
2718 + let row = db::get_latest_synckit_fleet_check(&pool, "mnw")
2719 + .await
2720 + .unwrap()
2721 + .unwrap();
2722 + assert_eq!(row.devices, 9);
2723 + }
2724 +
2725 + #[tokio::test]
2726 + async fn synckit_fleet_check_stores_an_unavailable_readout() {
2727 + let pool = db::connect_in_memory().await.unwrap();
2728 +
2729 + // A failed readout is persisted, not dropped: otherwise the ledger keeps
2730 + // serving the last good distribution and nothing shows that PoM went blind.
2731 + let result = SyncKitFleetCheckResult {
2732 + target: "mnw".to_string(),
2733 + window_days: 0,
2734 + devices: 0,
2735 + versions: Vec::new(),
2736 + checked_at: "2026-07-29T13:00:00Z".to_string(),
2737 + error: Some("HTTP 401 (alerts ingest token rejected)".to_string()),
2738 + };
2739 + db::insert_synckit_fleet_check(&pool, &result)
2740 + .await
2741 + .unwrap();
2742 +
2743 + let row = db::get_latest_synckit_fleet_check(&pool, "mnw")
2744 + .await
2745 + .unwrap()
2746 + .unwrap();
2747 + assert!(row.error.as_deref().unwrap().contains("401"));
2748 + assert!(row.version_list().is_empty());
2749 + }
2750 +
2751 + #[tokio::test]
2752 + async fn synckit_fleet_check_returns_none_for_unknown_target() {
2753 + let pool = db::connect_in_memory().await.unwrap();
2754 + assert!(
2755 + db::get_latest_synckit_fleet_check(&pool, "nope")
2756 + .await
2757 + .unwrap()
2758 + .is_none()
2759 + );
2760 + }
2761 +
2651 2762 #[tokio::test]
2652 2763 async fn whois_check_returns_latest() {
2653 2764 let pool = db::connect_in_memory().await.unwrap();
M pom/src/checks/mod.rs +3 -2
@@ -1,6 +1,6 @@
1 1 //! Check implementations, one module per probe kind: health, TLS, DNS, WHOIS,
2 - //! routes, CORS, backups, SSH, port scans, local systemd units, plus
3 - //! latency-drift analysis.
2 + //! routes, CORS, backups, SSH, port scans, local systemd units, the SyncKit
3 + //! field-version readout, plus latency-drift analysis.
4 4
5 5 pub mod backup;
6 6 pub mod cors;
@@ -12,6 +12,7 @@
12 12 pub mod scan_pipeline;
13 13 pub mod ssh;
14 14 pub mod ssh_banner;
15 + pub mod synckit_fleet;
15 16 pub mod systemd;
16 17 pub mod tls;
17 18 pub mod whois;
M pom/src/cli/serve.rs +2
@@ -115,6 +115,8 @@
115 115 &token,
116 116 alerter.as_ref(),
117 117 ));
118 + // No alerter: the fleet readout has nothing to alert on. See the task module.
119 + handles.extend(tasks::spawn_synckit_fleet_tasks(config, pool, &token));
118 120 handles.push(tasks::spawn_prune_task(pool, prune_days, &token));
119 121
120 122 // Spawn peer heartbeat tasks
M pom/src/db/maintenance.rs +8
@@ -18,6 +18,7 @@
18 18 pub whois: u64,
19 19 pub backups: u64,
20 20 pub systemd: u64,
21 + pub synckit_fleet: u64,
21 22 }
22 23
23 24 /// Delete records older than `days` from all tables.
@@ -40,6 +41,7 @@
40 41 whois: 0,
41 42 backups: 0,
42 43 systemd: 0,
44 + synckit_fleet: 0,
43 45 });
44 46 }
45 47
@@ -121,6 +123,11 @@
121 123 .execute(pool)
122 124 .await?;
123 125
126 + let synckit_fleet_result = sqlx::query("DELETE FROM synckit_fleet_checks WHERE checked_at < ?")
127 + .bind(&cutoff_str)
128 + .execute(pool)
129 + .await?;
130 +
124 131 Ok(PruneResult {
125 132 health: health_result.rows_affected(),
126 133 tests: test_result.rows_affected(),
@@ -134,5 +141,6 @@
134 141 whois: whois_result.rows_affected(),
135 142 backups: backups_result.rows_affected(),
136 143 systemd: systemd_result.rows_affected(),
144 + synckit_fleet: synckit_fleet_result.rows_affected(),
137 145 })
138 146 }
M pom/src/db/migrations.rs +16
@@ -262,6 +262,22 @@
262 262 CREATE INDEX IF NOT EXISTS idx_systemd_checks_target ON systemd_checks(target, id DESC);
263 263 ",
264 264 ),
265 + (
266 + 13,
267 + "add synckit_fleet_checks table",
268 + r"
269 + CREATE TABLE IF NOT EXISTS synckit_fleet_checks (
270 + id INTEGER PRIMARY KEY AUTOINCREMENT,
271 + target TEXT NOT NULL,
272 + window_days INTEGER NOT NULL,
273 + devices INTEGER NOT NULL,
274 + versions TEXT NOT NULL, -- JSON array of version snapshots
275 + checked_at TEXT NOT NULL,
276 + error TEXT
277 + );
278 + CREATE INDEX IF NOT EXISTS idx_synckit_fleet_checks_target ON synckit_fleet_checks(target, id DESC);
279 + ",
280 + ),
265 281 ];
266 282
267 283 #[instrument(skip_all)]
M pom/src/db/mod.rs +4 -2
M pom/src/cli/tasks/mod.rs +2
A pom/src/checks/synckit_fleet.rs +236
A pom/src/cli/tasks/synckit_fleet.rs +95
A pom/src/db/synckit_fleet.rs +66