release: keep stale rollups honest without losing fast path

.github/workflows/test.yml

@@ -7081,6 +7081,123 @@ jobs:
         run: |
           psql -h localhost -U postgres -d postgres -v ON_ERROR_STOP=1 -f devel/tests/query_attribution_assert.sql
 
+      - name: "Test 2.0: stale and rounded-up rollup source selection"
+        env:
+          PGPASSWORD: postgres
+        run: |
+          psql -h localhost -U postgres -d postgres -v ON_ERROR_STOP=1 << 'EOF'
+          begin;
+
+          -- When raw covers a wide aggregate window, rollup_1m may replace it
+          -- only if that rollup covers the requested end. Otherwise a lagging
+          -- worker would silently drop recent raw load. Completeness for a
+          -- stalled rollup older than physical raw coverage remains #122.
+          -- Conversely, a healthy rollup must keep the fast path when a
+          -- dashboard rounds until up to the next minute boundary.
+          do $$
+          declare
+            v_wait smallint;
+            v_start timestamptz := date_trunc('minute', now()) - interval '2 hours';
+            v_end timestamptz := date_trunc('minute', now());
+            v_historical_end timestamptz := v_end - interval '30 minutes';
+            v_start_ts int4;
+            v_recent_ts int4;
+            a record;
+          begin
+            truncate ash.sample_0, ash.sample_1, ash.sample_2;
+            truncate ash.rollup_1m, ash.rollup_1h;
+            update ash.config
+            set current_slot = 0,
+                num_partitions = 3,
+                sample_interval = interval '1 second',
+                rotation_period = interval '1 day',
+                rollup_1m_retention_days = 30,
+                rotated_at = v_end,
+                last_rollup_1m_ts = ash.ts_from_timestamptz(
+                  v_start + interval '1 minute'
+                ),
+                last_rollup_1h_ts = null
+            where singleton;
+
+            select ash._register_wait('active', 'IO', 'StaleRollupProof')
+            into v_wait;
+            v_start_ts := ash.ts_from_timestamptz(v_start);
+            v_recent_ts := ash.ts_from_timestamptz(v_end - interval '1 minute');
+
+            insert into ash.rollup_1m (
+              ts, datid, samples, peak_backends, wait_counts, query_counts
+            ) values (
+              v_start_ts, 0::oid, 1, 1,
+              array[v_wait, 1]::int4[], '{}'::int8[]
+            );
+            insert into ash.sample (
+              sample_ts, datid, active_count, data, slot
+            ) values
+              (v_start_ts, 0::oid, 1, array[-v_wait, 1, 0]::int4[], 0),
+              (v_recent_ts, 0::oid, 1, array[-v_wait, 1, 0]::int4[], 0);
+
+            select * into a from ash.aas(v_start, v_end);
+            assert a.source = 'raw',
+              'stale rollup must fall back to raw, got ' || a.source;
+            assert a.buckets_with_data = 2,
+              'raw fallback should retain both covered minutes, got '
+              || a.buckets_with_data;
+            assert a.backend_seconds = 2.00,
+              'raw fallback should retain both backend-seconds, got '
+              || a.backend_seconds;
+
+            truncate ash.rollup_1m;
+            insert into ash.rollup_1m (
+              ts, datid, samples, peak_backends, wait_counts, query_counts
+            )
+            select
+              ash.ts_from_timestamptz(minute_ts),
+              0::oid,
+              1,
+              1,
+              array[v_wait, 1]::int4[],
+              '{}'::int8[]
+            from generate_series(
+              v_start,
+              v_end - interval '1 minute',
+              interval '1 minute'
+            ) as minute_ts;
+            update ash.config
+            set last_rollup_1m_ts = ash.ts_from_timestamptz(v_historical_end)
+            where singleton;
+
+            select * into a from ash.aas(v_start, v_historical_end);
+            assert a.source = 'rollup_1m',
+              'healthy historical rollup must use its covered watermark, got '
+              || a.source;
+            assert a.buckets_with_data = 90,
+              'historical rollup should cover 90 complete minutes, got '
+              || a.buckets_with_data;
+            assert a.backend_seconds = 90.00,
+              'historical rollup should retain 90 backend-seconds, got '
+              || a.backend_seconds;
+
+            update ash.config
+            set last_rollup_1m_ts = ash.ts_from_timestamptz(v_end)
+            where singleton;
+            select * into a
+            from ash.aas(v_start, v_end + interval '1 minute');
+            assert a.source = 'rollup_1m',
+              'healthy rollup must keep the fast path for a rounded-up until, '
+              'got ' || a.source;
+            assert a.buckets_with_data = 120,
+              'healthy rollup should cover 120 complete minutes, got '
+              || a.buckets_with_data;
+            assert a.backend_seconds = 120.00,
+              'healthy rollup should retain 120 backend-seconds, got '
+              || a.backend_seconds;
+
+            raise notice 'stale/rounded-up rollup source selection PASSED';
+          end $$;
+
+          rollback;
+          EOF
+
       - name: "Test 2.0: minute-grain peak/p99 survive the rollup_1h seam"
         env:
           PGPASSWORD: postgres

RELEASE_NOTES.md

@@ -70,6 +70,13 @@ AAS-oriented 2.0 API:
   `ash.rollup_hour()` return processed minutes and hours rather than
   per-database rows upserted, so activity from multiple databases no longer
   inflates scheduler-visible results. (issue #191)
+- **Wide aggregate readers now reject stale rollup coverage without losing the
+  fast path.** When raw is the canonical source for a wide window, readers fall
+  back to it if `rollup_1m` has not reached the requested end, clamped at the
+  latest complete-minute boundary; a dashboard `until` rounded up to the next
+  minute no longer makes a healthy rollup fall back to raw. Completeness when a
+  window starts before physical raw coverage remains tracked by issue #122.
+  ([PR #199](https://github.com/NikolayS/pg_ash/pull/199))
 
 ## Retired tooling
 

sql/ash-install.sql

@@ -4041,12 +4041,15 @@ $$;
  * grain, so for anything wider than ~1 hour whose requested start is within
  * rollup_1m retention we prefer rollup_1m (a raw decode of a wide window spills
  * hundreds of MB — the last-24h read cost ~4.5s and ~500MB before this).
- * #122 tracks the separate end-watermark/completeness gap. Narrow windows
- * still fall through to _pick_source (raw preferred) so the freshest partial
- * minute is captured, and windows rollup can't cover (or where rollup is
- * disabled/lagging) still fall to raw / rollup_1h. Exact-query drills and
- * samples bypass this and force raw. The source column stays honest — it names
- * whatever was actually read.
+ * The rollup watermark must also reach the latest complete minute whenever raw
+ * is the preferred source at the window start; otherwise a stalled rollup
+ * worker would hide newer raw load. Narrow windows still fall through to
+ * _pick_source (raw preferred) so the freshest partial minute is captured.
+ * When raw covers the window start, a rollup that is disabled, cannot cover it,
+ * or lags the requested end falls back through _pick_source. Completeness for a
+ * stalled rollup when the window starts before physical raw coverage remains
+ * tracked by #122. Exact-query drills and samples bypass this and force raw.
+ * The source column stays honest — it names whatever was actually read.
  */
 create or replace function ash._pick_source_agg(
   since timestamptz,
@@ -4057,12 +4060,31 @@ language sql
 stable
 set search_path = pg_catalog, ash
 as $$
+  with fallback as (
+    select ash._pick_source(since) as source
+  )
   select case
     when extract(epoch from (until - since)) > 3600
          and ash._rollup_1m_retention_start() is not null
-         and since >= ash._rollup_1m_retention_start() then 'rollup_1m'
-    else ash._pick_source(since)
+         and since >= ash._rollup_1m_retention_start()
+         and (
+           fallback.source <> 'raw'
+           or coalesce(
+                (
+                  select ash.ts_to_timestamptz(last_rollup_1m_ts)
+                           >= least(
+                                until,
+                                date_trunc('minute', now())
+                              )
+                  from ash.config
+                  where singleton
+                ),
+                false
+              )
+         ) then 'rollup_1m'
+    else fallback.source
   end
+  from fallback
 $$;
 
 /*