Merge pull request #946 from warmbly/devin/1791705885-monitoring-source-clock

feat: compare monitoring evidence against a statement-local database clock

.github/workflows/ci.yml

@@ -228,6 +228,23 @@ jobs:
           go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults|DispatchDefersInvalidHours)|WarmupSenderDistinguishes)' -count=1
           go test -race ./internal/app/consumer -run '^TestLiveStoredSendResult' -count=1
 
+      - name: Monitoring database observation regressions (PostgreSQL)
+        env:
+          WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
+        run: |
+          set -euo pipefail
+          go test -race -v ./internal/repository -run '^TestLiveMonitoringDatabaseClock' -count=1 | tee monitoring-clock.log
+          if grep -q -- '--- SKIP:' monitoring-clock.log; then
+            echo "Monitoring database observation regressions must execute without skips"
+            exit 1
+          fi
+          for test in TestLiveMonitoringDatabaseClockAfterRequestStart \
+            TestLiveMonitoringDatabaseClockWarningSemantics \
+            TestLiveMonitoringDatabaseClockPredicateBoundaries \
+            TestLiveMonitoringDatabaseClockSingleSnapshot; do
+            grep -q "^--- PASS: $test " monitoring-clock.log
+          done
+
       - name: Warmup charge and authoritative result regressions (PostgreSQL)
         env:
           WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable

docs/content/docs/api/endpoints.mdx

@@ -635,6 +635,7 @@ Alongside them, `api_url` is this API's own public base (for a copyable example
 - All of `/pool-link/*` and `/cloud-link/*` (the self-hosted warmup pool link: approving an instance's code, listing and unlinking instances, reading the pool plan's price and opening its checkout, and on a self-hosted instance the connect flow and mailbox enrollment). `/cloud-link/*` exists only on a self-hosted instance. Connections belong to the selected local workspace, but linking and unlinking them (`/connect`, `/connect/poll`, `DELETE /cloud-link`) and the linked cloud workspace's mailboxes (`/workspace-mailboxes`, `.../adopt`) take a platform admin holding `manage_settings` whose session presented a second factor (`admin_mfa_required` otherwise). Reading the link and putting a workspace's own mailboxes on it (`GET /cloud-link`, `/mailboxes`, enroll, pause, resume and the Google and Microsoft sign-in under `/oauth/*`) take the workspace's `MANAGE_EMAILS`. `GET /cloud-link` includes `legacy_connected`; mailbox rows include `legacy` when they still use the old connection. A normal `DELETE /cloud-link` only disconnects the selected workspace's scoped link. `DELETE /cloud-link?legacy=true` explicitly disconnects the legacy link across all workspaces and uses the same platform-admin permission. New connections and enrollments require workspace scope, and a Cloud workspace accepts only one active link (legacy links count). `POST /pool-link/codes` requires `remote_organization_id` and returns `workspace_scoped: true` for supported clients. `POST /pool-link/codes` and `POST /pool-link/poll` are public and per-IP rate limited: they are the device-code handshake an instance uses before it has a token, and `/pool-link/instance/*` accepts only an instance token. `/pool-link/instance/oauth/*`, `/pool-link/instance/mailboxes/:id/token`, `/pool-link/instance/workspace-mailboxes` and `/pool-link/instance/mailboxes/adopt` are the cloud-managed mailbox surface (Google and Microsoft sign-in on Warmbly's OAuth apps, brokered access tokens); their instance-side counterparts are `/cloud-link/oauth/*` and `/cloud-link/workspace-mailboxes/*`. `/pool-link/instance/redirects` (list), and `GET`, `PUT`, `DELETE` and `POST .../verify` on `/pool-link/instance/redirects/:domain`, are the root redirects Warmbly Cloud serves for the calling instance; the instance drives them through `PUT /emails/domains/:domain/redirect` with `served_by: "cloud"`, and `GET /pool-link/instance` carries `redirects` (`available`, `host`, `limit`, `used`)
 - All of `/admin/*`. `DELETE /admin/instance/invitations/expired` requires platform admin `manage_organizations` and an MFA-verified session. It removes only expired invitation records across all workspaces, leaves active invitations and membership unchanged, records an admin audit entry, and returns `{ "cleaned": true }`. The admin panel confirms before calling it.
   - `GET /admin/instance/health` requires `view_analytics` and an MFA-verified admin session. Returns configuration `checks`, `summary`, additive `execution_coverage` (currently `unreported`) and `operational_monitoring_path`. Empty findings do not establish execution coverage or operational health.
+  - Aggregate PostgreSQL monitoring sources, including sync and worker health samples, use one database reference for SQL thresholds, windows and source/metric observation clocks. That observation can be later than the snapshot's collection-start `checked_at`; cached snapshots keep their original clocks. Database observation is not provider recovery evidence.
   - `GET /admin/instance/monitoring` requires `view_analytics` and an MFA-verified admin session, not an organization API-key scope. Returns the read-only version `"1"` snapshot with `checked_at`, `refresh_after`, `coverage` and permission-shaped `sources`; HTTP responses use `Cache-Control: no-store`, while the backend shares a bounded one-minute collection cache. Each source has availability, coverage, nullable observation/scope measurements and metrics with nullable counts, affected mailboxes/organizations, evidence clocks, conditions and optional investigation paths. Missing permissions return unavailable source placeholders with `permission_denied`, not private metrics. Source bits and interpretation are listed under [operational monitoring](/development/instance-health/#operational-monitoring). Unavailable is not zero or an outage; a positive broker lag/pending observation remains `unknown`/`info`, not proof of failed throughput. This route never subscribes, replays, reconnects, clears holds or changes broker offsets/assignments.
   - `GET /admin/analytics/overview` requires `view_analytics`. `total_emails_sent` and `emails_sent_today` count confirmed warmup and other outbound sends, not dispatches. Today uses the UTC calendar day. A send-count database error fails the request instead of returning zero counters. See [platform admin send counters](/guides/analytics/#platform-admin-send-counters).
   - `GET /admin/organizations` and `GET /admin/organizations/:id` require `view_organizations`. Their acquisition fields include the nullable `referrer_host` saved at signup, alongside UTM source, medium, campaign and landing path. The list accepts an exact `referrer_host` filter as well as its existing acquisition filters. An absent source is unknown, not proof of a direct visit. See [signup source attribution](/guides/analytics/#signup-source-attribution-platform-admins).

docs/content/docs/development/instance-health.mdx

@@ -93,6 +93,8 @@ Worker heartbeat totals and up to 20 worker details share one PostgreSQL snapsho
 
 ### Reading a snapshot
 
+Aggregate sources collected in one database query, including sync and worker health samples, capture one database reference time for all their thresholds, windows and source/metric observations. A normal completion or receipt visible when the query starts is compared with that reference, not the earlier refresh request start. Genuinely future-dated evidence still warns, with no clock-skew tolerance. The snapshot's checked time and cache expiry keep their collection-start meaning; the database observation does not confirm provider recovery.
+
 - **Snapshot checked** and **Collection attempted** are collection clocks. They do not confirm delivery, sync or recovery. **Source observed** and **Observed** are the measurement clocks; **Evidence since**, **Latest evidence**, their ages and any window are distinct. The UI retains the original clocks after a failed refresh and marks retained measurements stale. A successful browser fetch of old data does not make it fresh.
 - `fresh`, `stale` and `unavailable` describe measurement availability. The UI also marks observations older than two minutes or implausibly future-dated as stale/unknown. Coverage can be complete only for the measured sources; it does not mean every customer operation succeeded.
 - A nullable count is **Not measured**, not zero. A successful zero means no matching rows in that metric's scope/window. Zero recent sends means **No matching recent evidence**, not healthy throughput. A completed task, heartbeat, low memory usage or sync tick does not prove delivery or recovery.

internal/app/monitoring/service_test.go

@@ -69,6 +69,37 @@ func TestMonitoringPreservesSourceObservationAfterRefreshStart(t *testing.T) {
 	}
 }
 
+func TestMonitoringDatabaseObservationKeepsCacheAndAttemptClocks(t *testing.T) {
+	start := time.Date(2026, 10, 10, 16, 0, 59, 0, time.UTC)
+	now := start
+	observed := start.Add(650 * time.Millisecond)
+	windowStart := observed.Add(-time.Hour)
+	count := int64(3)
+	calls := 0
+	fail := false
+	s := New([]Source{{ID: "worker_samples", Permission: models.AdminPermViewWorkers, Collect: func(_ context.Context, _ time.Time) (models.MonitoringSource, error) {
+		calls++
+		if fail {
+			return models.MonitoringSource{}, errors.New("collection unavailable")
+		}
+		return models.MonitoringSource{ObservedAt: &observed, Metrics: []models.MonitoringMetric{{ID: "worker_sample_auth_errors_1h", Count: &count, ObservedAt: &observed, WindowStart: &windowStart, WindowEnd: &observed, Availability: models.MonitoringFresh, Condition: models.MonitoringRecentFailure}}}, nil
+	}}})
+	s.now = func() time.Time { return now }
+	first := s.Snapshot(t.Context(), models.AdminPermViewWorkers)
+	now = start.Add(30 * time.Second)
+	cached := s.Snapshot(t.Context(), models.AdminPermViewWorkers)
+	if calls != 1 || !cached.CheckedAt.Equal(start) || !cached.RefreshAfter.Equal(first.RefreshAfter) || !cached.Sources[0].CheckedAt.Equal(start) || !cached.Sources[0].ObservedAt.Equal(observed) || !cached.Sources[0].Metrics[0].WindowEnd.Equal(observed) {
+		t.Fatal("cache changed the source observation or checked-at clocks", cached)
+	}
+	fail = true
+	now = start.Add(CacheTTL + time.Second)
+	retained := s.Snapshot(t.Context(), models.AdminPermViewWorkers)
+	source := retained.Sources[0]
+	if calls != 2 || !retained.CheckedAt.Equal(now) || !source.CheckedAt.Equal(now) || source.Availability != models.MonitoringStale || !source.ObservedAt.Equal(observed) || !source.Metrics[0].ObservedAt.Equal(observed) || !source.Metrics[0].WindowStart.Equal(windowStart) || !source.Metrics[0].WindowEnd.Equal(observed) || *source.Metrics[0].Count != count {
+		t.Fatal("failed refresh replaced retained source evidence with a collection clock", retained)
+	}
+}
+
 func TestMonitoringFailedRefreshRetainsOriginalEvidenceThenExpires(t *testing.T) {
 	now := time.Date(2026, 10, 9, 12, 0, 0, 0, time.UTC)
 	fail := false

internal/repository/pg_monitoring.go

@@ -69,14 +69,7 @@ func (r *MonitoringRepository) collect(ctx context.Context, at time.Time, id str
 			return out, nil
 		}
 	}
-	queries := ""
-	for _, q := range section.queries {
-		if queries != "" {
-			queries += " UNION ALL "
-		}
-		queries += strings.Replace(strings.ReplaceAll(q.query, "$1", "$1::timestamptz"), "SELECT ", "SELECT '"+q.id+"', ", 1) + " AND $1::timestamptz IS NOT NULL"
-	}
-	rows, err := r.pool.Query(ctx, queries, at)
+	rows, err := r.pool.Query(ctx, monitoringStatement(section))
 	if err != nil {
 		return out, err
 	}
@@ -87,14 +80,18 @@ func (r *MonitoringRepository) collect(ctx context.Context, at time.Time, id str
 			return out, errors.New("unexpected monitoring rows")
 		}
 		var id string
+		var observed time.Time
 		m := models.MonitoringMetric{}
-		if err := rows.Scan(&id, &m.Count, &m.AffectedMailboxes, &m.AffectedOrganizations, &m.UnknownOrganizationRows, &m.EvidenceAt, &m.LatestEvidenceAt, &m.NextEligibleAt); err != nil {
+		if err := rows.Scan(&id, &observed, &m.Count, &m.AffectedMailboxes, &m.AffectedOrganizations, &m.UnknownOrganizationRows, &m.EvidenceAt, &m.LatestEvidenceAt, &m.NextEligibleAt); err != nil {
 			return out, err
 		}
+		if index == 0 {
+			out.ObservedAt = &observed
+		}
 		found := false
 		for _, q := range section.queries {
 			if q.id == id {
-				v := monitoringMetric(q, at, section.investigate)
+				v := monitoringMetric(q, observed, section.investigate)
 				v.Count = m.Count
 				v.AffectedMailboxes = m.AffectedMailboxes
 				v.AffectedOrganizations = m.AffectedOrganizations
@@ -123,6 +120,16 @@ func (r *MonitoringRepository) collect(ctx context.Context, at time.Time, id str
 	return out, nil
 }
 
+func monitoringStatement(section monitoringSection) string {
+	queries := make([]string, 0, len(section.queries))
+	for _, q := range section.queries {
+		query := strings.ReplaceAll(q.query, "$1", "(SELECT at FROM observation)")
+		queries = append(queries, strings.Replace(query, "SELECT ", "SELECT '"+q.id+"', (SELECT at FROM observation), ", 1))
+	}
+	// One database reference for every predicate and observation in the statement snapshot.
+	return "WITH observation AS MATERIALIZED (SELECT clock_timestamp() AS at) " + strings.Join(queries, " UNION ALL ")
+}
+
 func monitoringMetric(q monitoringQuery, at time.Time, path string) models.MonitoringMetric {
 	m := models.MonitoringMetric{ID: q.id, Title: q.title, Unit: q.unit, Condition: q.condition, Severity: q.severity, Note: q.note, Investigate: path, ObservedAt: &at, Availability: models.MonitoringFresh}
 	if seconds, ok := monitoringThresholds[q.id]; ok {

internal/repository/pg_monitoring_clock_live_test.go

@@ -0,0 +1,347 @@
+package repository
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+type monitoringClockTrace struct {
+	queries atomic.Int64
+}
+
+type monitoringClockTraceKey struct{}
+
+func (tr *monitoringClockTrace) TraceQueryStart(ctx context.Context, _ *pgx.Conn, data pgx.TraceQueryStartData) context.Context {
+	return context.WithValue(ctx, monitoringClockTraceKey{}, strings.Contains(data.SQL, "SELECT '"))
+}
+
+func (tr *monitoringClockTrace) TraceQueryEnd(ctx context.Context, _ *pgx.Conn, _ pgx.TraceQueryEndData) {
+	if measured, _ := ctx.Value(monitoringClockTraceKey{}).(bool); measured {
+		tr.queries.Add(1)
+	}
+}
+
+func monitoringClockPool(t *testing.T, pool *pgxpool.Pool, trace *monitoringClockTrace) *pgxpool.Pool {
+	t.Helper()
+	cfg := pool.Config().Copy()
+	cfg.MaxConns = 1
+	cfg.MinConns = 0
+	cfg.ConnConfig.RuntimeParams["default_transaction_read_only"] = "on"
+	cfg.ConnConfig.RuntimeParams["statement_timeout"] = "3000"
+	cfg.ConnConfig.RuntimeParams["timezone"] = "America/Los_Angeles"
+	cfg.ConnConfig.Tracer = trace
+	ro, err := pgxpool.NewWithConfig(t.Context(), cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(ro.Close)
+	var readOnly bool
+	if err := ro.QueryRow(t.Context(), `SELECT current_setting('transaction_read_only')::boolean`).Scan(&readOnly); err != nil || !readOnly {
+		t.Fatal("collection connection is not read-only", err)
+	}
+	return ro
+}
+
+func requireMonitoringClock(t *testing.T, source models.MonitoringSource, before, after time.Time) {
+	t.Helper()
+	if source.ObservedAt == nil || source.ObservedAt.Before(before) || source.ObservedAt.After(after) {
+		t.Fatalf("source clock outside database collection: observed=%v before=%v after=%v", source.ObservedAt, before, after)
+	}
+	for _, metric := range source.Metrics {
+		if metric.ObservedAt == nil || !metric.ObservedAt.Equal(*source.ObservedAt) {
+			t.Fatalf("metric %s has a different source clock: %v", metric.ID, metric.ObservedAt)
+		}
+		if metric.WindowEnd != nil && (!metric.WindowEnd.Equal(*source.ObservedAt) || metric.WindowStart == nil) {
+			t.Fatalf("metric %s window has a different source clock", metric.ID)
+		}
+	}
+}
+
+func TestLiveMonitoringDatabaseClockAfterRequestStart(t *testing.T) {
+	for _, source := range []string{"sync", "worker_samples"} {
+		t.Run(source, func(t *testing.T) {
+			_, pool := liveContactDB(t)
+			requireSchemaVersion(t, pool, 287)
+			f := newWarmupUsageFixture(t, pool)
+			worker := monitoringWorkerNode(t, pool, "worker", true, nil)
+			f.exec(`UPDATE email_accounts SET status='active',worker_id=$2 WHERE id=$1`, f.account, worker)
+			requested := monitoringWorkerClock(t, pool)
+			var receipt time.Time
+			metric := "sync_future_clock"
+			if source == "sync" {
+				if err := pool.QueryRow(t.Context(), `INSERT INTO email_sync_state(email_id,user_id,last_synced_at) VALUES($1,$2,clock_timestamp()) RETURNING last_synced_at`, f.account, f.user).Scan(&receipt); err != nil {
+					t.Fatal(err)
+				}
+			} else {
+				metric = "worker_samples_missing"
+				if err := pool.QueryRow(t.Context(), `INSERT INTO worker_health_samples(worker_id,observed_at,auth_errors) VALUES($1,clock_timestamp(),3) RETURNING observed_at`, worker).Scan(&receipt); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if !receipt.After(requested) {
+				t.Fatal("fixture receipt did not follow request start")
+			}
+			trace := &monitoringClockTrace{}
+			ro := monitoringClockPool(t, pool, trace)
+			before := monitoringWorkerClock(t, pool)
+			got := collectMonitoring(t, NewMonitoringRepository(ro), source, requested)
+			after := monitoringWorkerClock(t, pool)
+			m := measuredMonitoring(t, got, metric)
+			t.Logf("request=%s database_receipt=%s collection_before=%s warning_count=%d observation=%v", requested.Format(time.RFC3339Nano), receipt.Format(time.RFC3339Nano), before.Format(time.RFC3339Nano), *m.Count, m.ObservedAt)
+			if *m.Count != 0 {
+				t.Fatalf("normal database receipt after request start was classified future/missing: %s=%d", metric, *m.Count)
+			}
+			requireMonitoringClock(t, got, before, after)
+			if trace.queries.Load() != 1 || ro.Stat().AcquiredConns() != 0 {
+				t.Fatal("source did not use one released connection and statement", trace.queries.Load())
+			}
+			if source == "worker_samples" {
+				window := measuredMonitoring(t, got, "worker_sample_auth_errors_1h")
+				if *window.Count != 3 || window.WindowStart == nil || window.WindowEnd.Sub(*window.WindowStart) != time.Hour {
+					t.Fatal("new receipt was excluded from the measured window", window)
+				}
+			}
+		})
+	}
+}
+
+func TestLiveMonitoringDatabaseClockWarningSemantics(t *testing.T) {
+	_, pool := liveContactDB(t)
+	requireSchemaVersion(t, pool, 287)
+	f := newWarmupUsageFixture(t, pool)
+	worker := monitoringWorkerNode(t, pool, "worker", true, nil)
+	f.exec(`UPDATE email_accounts SET status='active',worker_id=$2 WHERE id=$1`, f.account, worker)
+	now := monitoringWorkerClock(t, pool)
+	stale, future, expired := now.Add(-time.Hour), now.Add(time.Hour), now.Add(-time.Second)
+	r := NewMonitoringRepository(monitoringClockPool(t, pool, &monitoringClockTrace{}))
+	for _, tc := range []struct {
+		name, provider, metric string
+		last, hold             *time.Time
+		missing                bool
+		condition              models.MonitoringCondition
+	}{
+		{"future smtp", "smtp_imap", "sync_future_clock", &future, nil, false, models.MonitoringUnknown},
+		{"future gmail", "gmail", "sync_future_clock", &future, nil, false, models.MonitoringUnknown},
+		{"future outlook", "outlook", "sync_future_clock", &future, nil, false, models.MonitoringUnknown},
+		{"stale", "smtp_imap", "sync_stale", &stale, nil, false, models.MonitoringTelemetryStale},
+		{"null completion", "smtp_imap", "sync_stale", nil, nil, false, models.MonitoringTelemetryStale},
+		{"missing state", "smtp_imap", "sync_missing", nil, nil, true, models.MonitoringUnknown},
+		{"held stale", "smtp_imap", "sync_held_stale", &stale, &future, false, models.MonitoringBudget},
+		{"expired hold", "smtp_imap", "sync_stale", &stale, &expired, false, models.MonitoringTelemetryStale},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f.exec(`UPDATE email_accounts SET provider=$2 WHERE id=$1`, f.account, tc.provider)
+			f.exec(`DELETE FROM email_sync_state WHERE email_id=$1`, f.account)
+			if !tc.missing {
+				f.exec(`INSERT INTO email_sync_state(email_id,user_id,last_synced_at,throttled_until,throttle_reason) VALUES($1,$2,$3,$4,'org_daily')`, f.account, f.user, tc.last, tc.hold)
+			}
+			before := monitoringWorkerClock(t, pool)
+			got := collectMonitoring(t, r, "sync", now.Add(24*time.Hour))
+			requireMonitoringClock(t, got, before, monitoringWorkerClock(t, pool))
+			m := measuredMonitoring(t, got, tc.metric)
+			if *m.Count != 1 || m.Condition != tc.condition || m.Severity != "warning" && tc.metric != "sync_held_stale" {
+				t.Fatal("real warning was hidden or reclassified", m)
+			}
+			if tc.metric == "sync_future_clock" && (m.LatestEvidenceAt == nil || !m.LatestEvidenceAt.Equal(future)) {
+				t.Fatal("future evidence clock was changed", m)
+			}
+			if tc.metric == "sync_held_stale" {
+				if *measuredMonitoring(t, got, "sync_stale").Count != 0 || *measuredMonitoring(t, got, "sync_hold_org_daily").Count != 1 {
+					t.Fatal("active hold did not retain its separate evidence", got)
+				}
+			}
+			if tc.name == "expired hold" && *measuredMonitoring(t, got, "sync_hold_org_daily").Count != 0 {
+				t.Fatal("expired hold stayed active", got)
+			}
+		})
+	}
+	for _, tc := range []struct {
+		name string
+		at   *time.Time
+	}{
+		{"future receipt", &future}, {"stale receipt", &stale}, {"missing receipt", nil},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f.exec(`DELETE FROM worker_health_samples WHERE worker_id=$1`, worker)
+			if tc.at != nil {
+				f.exec(`INSERT INTO worker_health_samples(worker_id,observed_at) VALUES($1,$2)`, worker, tc.at)
+			}
+			got := collectMonitoring(t, r, "worker_samples", now.Add(24*time.Hour))
+			m := measuredMonitoring(t, got, "worker_samples_missing")
+			if *m.Count != 1 || m.Condition != models.MonitoringTelemetryStale || m.Severity != "warning" {
+				t.Fatal("future/stale/missing worker receipt was hidden", m)
+			}
+		})
+	}
+	f.exec(`UPDATE email_accounts SET auth_state='passing',auth_checked_at=$2,send_recovery_hold=true,send_recovery_reason='unknown' WHERE id=$1`, f.account, future)
+	domain := collectMonitoring(t, r, "domain_auth", now)
+	if *measuredMonitoring(t, domain, "domain_future_clock").Count != 1 || *measuredMonitoring(t, domain, "domain_passing_fresh").Count != 0 {
+		t.Fatal("provider future observation became passing evidence", domain)
+	}
+	if *measuredMonitoring(t, collectMonitoring(t, r, "send_safety", now), "holds_unknown").Count != 1 {
+		t.Fatal("unknown-send protection disappeared")
+	}
+}
+
+func TestLiveMonitoringDatabaseClockPredicateBoundaries(t *testing.T) {
+	_, pool := liveContactDB(t)
+	requireSchemaVersion(t, pool, 287)
+	f := newWarmupUsageFixture(t, pool)
+	worker := monitoringWorkerNode(t, pool, "worker", true, nil)
+	f.exec(`UPDATE email_accounts SET status='active',worker_id=$2 WHERE id=$1`, f.account, worker)
+	at := monitoringWorkerClock(t, pool)
+	for _, tc := range []struct {
+		name, section, metric string
+		offset                time.Duration
+		want                  int64
+	}{
+		{"sync just past", "sync", "sync_future_clock", -time.Microsecond, 0},
+		{"sync at reference", "sync", "sync_future_clock", 0, 0},
+		{"sync one microsecond future", "sync", "sync_future_clock", time.Microsecond, 1},
+		{"sync exactly stale cutoff", "sync", "sync_stale", -20 * time.Minute, 0},
+		{"sync just outside cutoff", "sync", "sync_stale", -20*time.Minute - time.Microsecond, 1},
+		{"sync just inside cutoff", "sync", "sync_stale", -20*time.Minute + time.Microsecond, 0},
+		{"hold exactly reference", "sync", "sync_hold_org_daily", 0, 0},
+		{"hold just expired", "sync", "sync_hold_org_daily", -time.Microsecond, 0},
+		{"hold one microsecond active", "sync", "sync_hold_org_daily", time.Microsecond, 1},
+		{"sample exactly cutoff", "worker_samples", "worker_samples_missing", -20 * time.Minute, 0},
+		{"sample just outside cutoff", "worker_samples", "worker_samples_missing", -20*time.Minute - time.Microsecond, 1},
+		{"sample just inside cutoff", "worker_samples", "worker_samples_missing", -20*time.Minute + time.Microsecond, 0},
+		{"sample at reference", "worker_samples", "worker_samples_missing", 0, 0},
+		{"sample one microsecond future", "worker_samples", "worker_samples_missing", time.Microsecond, 1},
+		{"window lower inclusive", "worker_samples", "worker_sample_auth_errors_1h", -time.Hour, 7},
+		{"window just outside", "worker_samples", "worker_sample_auth_errors_1h", -time.Hour - time.Microsecond, -1},
+		{"window just inside", "worker_samples", "worker_sample_auth_errors_1h", -time.Hour + time.Microsecond, 7},
+		{"window upper inclusive", "worker_samples", "worker_sample_auth_errors_1h", 0, 7},
+		{"window just future", "worker_samples", "worker_sample_auth_errors_1h", time.Microsecond, -1},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			stamp := at.Add(tc.offset)
+			f.exec(`DELETE FROM email_sync_state WHERE email_id=$1`, f.account)
+			f.exec(`INSERT INTO email_sync_state(email_id,user_id,last_synced_at,throttled_until,throttle_reason) VALUES($1,$2,$3,$3,'org_daily')`, f.account, f.user, stamp)
+			if tc.metric == "sync_stale" {
+				f.exec(`UPDATE email_sync_state SET throttled_until=NULL WHERE email_id=$1`, f.account)
+			}
+			f.exec(`DELETE FROM worker_health_samples WHERE worker_id=$1`, worker)
+			f.exec(`INSERT INTO worker_health_samples(worker_id,observed_at,auth_errors) VALUES($1,$2,7)`, worker, stamp)
+			var q monitoringQuery
+			for _, candidate := range monitoringSections[tc.section].queries {
+				if candidate.id == tc.metric {
+					q = candidate
+				}
+			}
+			if q.id == "" {
+				t.Fatal("missing production predicate")
+			}
+			scope := f.account
+			switch {
+			case tc.section == "sync":
+				q.query += " AND e.id=$2"
+			case tc.metric == "worker_samples_missing":
+				q.query += " AND n.id=$2"
+				scope = worker
+			default:
+				q.query += " AND h.worker_id=$2"
+				scope = worker
+			}
+			// Fix only the test clock to exercise exact native predicate boundaries.
+			query := strings.Replace(monitoringStatement(monitoringSection{queries: []monitoringQuery{q}}), "clock_timestamp()", "$1::timestamptz", 1)
+			var id string
+			var observed time.Time
+			var m models.MonitoringMetric
+			if err := pool.QueryRow(t.Context(), query, at, scope).Scan(&id, &observed, &m.Count, &m.AffectedMailboxes, &m.AffectedOrganizations, &m.UnknownOrganizationRows, &m.EvidenceAt, &m.LatestEvidenceAt, &m.NextEligibleAt); err != nil {
+				t.Fatal(err)
+			}
+			if id != q.id || !observed.Equal(at) || tc.want == -1 && m.Count != nil || tc.want >= 0 && (m.Count == nil || *m.Count != tc.want) {
+				t.Fatalf("native boundary changed: observed=%v metric=%+v want=%d", observed, m, tc.want)
+			}
+		})
+	}
+}
+
+func TestLiveMonitoringDatabaseClockSingleSnapshot(t *testing.T) {
+	_, pool := liveContactDB(t)
+	requireSchemaVersion(t, pool, 287)
+	f := newWarmupUsageFixture(t, pool)
+	worker := monitoringWorkerNode(t, pool, "worker", true, nil)
+	f.exec(`UPDATE email_accounts SET status='active',worker_id=$2 WHERE id=$1`, f.account, worker)
+	old := monitoringWorkerClock(t, pool).Add(-time.Hour)
+	f.exec(`INSERT INTO email_sync_state(email_id,user_id,last_synced_at) VALUES($1,$2,$3)`, f.account, f.user, old)
+	trace := &monitoringClockTrace{}
+	ro := monitoringClockPool(t, pool, trace)
+	var pid int
+	if err := ro.QueryRow(t.Context(), `SELECT pg_backend_pid()`).Scan(&pid); err != nil {
+		t.Fatal(err)
+	}
+	writer, err := pool.Acquire(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer writer.Release()
+	key := int64(uuid.New().ID())
+	if _, err := writer.Exec(t.Context(), `SELECT pg_advisory_lock($1)`, key); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _, _ = writer.Exec(context.Background(), `SELECT pg_advisory_unlock($1)`, key) }()
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	section := monitoringSections["sync"]
+	section.queries = append([]monitoringQuery(nil), section.queries...)
+	section.queries[0].query += " AND (SELECT pg_advisory_xact_lock(" + strconv.FormatInt(key, 10) + ")::text)=''"
+	result := make(chan struct {
+		source models.MonitoringSource
+		err    error
+	}, 1)
+	before := monitoringWorkerClock(t, pool)
+	go func() {
+		source, err := NewMonitoringRepository(ro).collect(ctx, before, "sync", section)
+		result <- struct {
+			source models.MonitoringSource
+			err    error
+		}{source, err}
+	}()
+	tick := time.NewTicker(time.Millisecond)
+	defer tick.Stop()
+	for {
+		var waiting bool
+		if err := writer.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_locks WHERE pid=$1 AND locktype='advisory' AND NOT granted)`, pid).Scan(&waiting); err != nil {
+			t.Fatal(err)
+		}
+		if waiting {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			t.Fatal("native statement did not reach the snapshot barrier", ctx.Err())
+		case <-tick.C:
+		}
+	}
+	if _, err := writer.Exec(ctx, `UPDATE email_sync_state SET last_synced_at=clock_timestamp() WHERE email_id=$1`, f.account); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := writer.Exec(ctx, `SELECT pg_advisory_unlock($1)`, key); err != nil {
+		t.Fatal(err)
+	}
+	got := <-result
+	if got.err != nil {
+		t.Fatal(got.err)
+	}
+	requireMonitoringClock(t, got.source, before, monitoringWorkerClock(t, pool))
+	stale := measuredMonitoring(t, got.source, "sync_stale")
+	rows := measuredMonitoring(t, got.source, "sync_state_rows")
+	if *stale.Count != 1 || rows.LatestEvidenceAt == nil || !rows.LatestEvidenceAt.Equal(old) || trace.queries.Load() != 1 || ro.Stat().AcquiredConns() != 0 {
+		t.Fatal("concurrent commit split the native statement snapshot", got.source, trace.queries.Load())
+	}
+	if *measuredMonitoring(t, collectMonitoring(t, NewMonitoringRepository(ro), "sync", before), "sync_stale").Count != 0 {
+		t.Fatal("later statement did not observe the committed completion")
+	}
+}

internal/repository/pg_monitoring_test.go

@@ -0,0 +1,22 @@
+package repository
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestMonitoringStatementUsesOneDatabaseObservation(t *testing.T) {
+	for id, section := range monitoringSections {
+		t.Run(id, func(t *testing.T) {
+			query := monitoringStatement(section)
+			if !strings.HasPrefix(query, "WITH observation AS MATERIALIZED (SELECT clock_timestamp() AS at) ") || strings.Count(query, "clock_timestamp()") != 1 || strings.Contains(query, "$1") || strings.Count(query, " UNION ALL ") != len(section.queries)-1 {
+				t.Fatal("source lost its single statement/database reference", query)
+			}
+			for _, metric := range section.queries {
+				if !strings.Contains(query, "SELECT '"+metric.id+"', (SELECT at FROM observation), ") {
+					t.Fatal("metric does not return the source reference", metric.id)
+				}
+			}
+		})
+	}
+}