Start GEFS 35-day virtual updates before extension completion

docs/virtual_datasets.md

@@ -10,7 +10,7 @@ A virtual dataset complements the materialized (rechunked) time-series dataset o
 
 - **Spatial / map-optimized chunking.** Chunks follow the native source-file layout — usually one time step over the full spatial grid — so a virtual `-spatial` dataset serves map and spatial queries efficiently, while the materialized dataset provides the complementary time-series chunking.
 - **Every curated source variable, cheaply.** A variable costs only a handful of byte-range refs, not a rechunked copy, so a virtual dataset can include every variable we have metadata for rather than just the materialized subset.
-- **Very low latency updates.** Writing refs records byte offsets without moving data, so a new source file's data is visible within seconds of publication (target ≤ 5 s end to end). See [Operational updates](#operational-updates-single-writer).
+- **Very low latency updates.** Writing refs records byte offsets without moving data, so a new source file's data can be visible within seconds of publication while an update is polling (target ≤ 5 s end to end). See [Operational updates](#operational-updates-single-writer).
 
 ## Scale
 
@@ -76,7 +76,7 @@ Virtual stores carry no consolidated metadata in the root zarr.json — nothing
 
 Crash recovery is automatic: committed refs are durable and the filter skips them on the next cron fire; uncommitted refs and expansions vanish and are re-discovered.
 
-**Files the source hasn't published yet roll to the next fire rather than failing the run.** An update stops polling at its `poll_deadline` — the scheduled fire plus the cron job's `pod_active_deadline`, less a small grace — and returns with those files still pending; final validation determines whether the Job succeeds. The next fire re-sweeps the same window and ingests them a tick after they publish, so waiting in-run and handing off cost the same latency. `pod_active_deadline` therefore sizes how long a fire chases its own files: cover the source's typical publication time with headroom, and let a late cycle roll over rather than stretching past the fire interval (`concurrencyPolicy: Replace` kills a run that outlives its successor's fire anyway). Both ends of a fire's work are anchored to its scheduled fire rather than to process start: a pod replacing an evicted one stops polling when the pod it replaced would have, and its window ends at that fire, so it finishes its predecessor's steps instead of adopting the next fire's and waiting out its deadline on files that publish after it. A source that stops publishing altogether is caught by validation in the final update pod (data currency and manifest completeness).
+**Files the source hasn't published yet roll to the next fire rather than failing the run.** An update stops polling at its `poll_deadline` — the scheduled fire plus the cron job's `pod_active_deadline`, less a small grace — and returns with those files still pending; final validation determines whether the Job succeeds. The next fire re-sweeps an overlapping window. Files published during an inactive gap wait until that fire; low latency requires a polling pod when the source files arrive. `pod_active_deadline` therefore sizes how long a fire chases its own files: cover the source's typical publication time with headroom, and let a late cycle roll over rather than stretching past the fire interval (`concurrencyPolicy: Replace` kills a run that outlives its successor's fire anyway). Both ends of a fire's work are anchored to its scheduled fire rather than to process start: a pod replacing an evicted one stops polling when the pod it replaced would have, and its window ends at that fire, so it finishes its predecessor's steps instead of adopting the next fire's and waiting out its deadline on files that publish after it. A source that stops publishing altogether is caught by validation in the final update pod (data currency and manifest completeness).
 
 Validation shares the single `pod_active_deadline` with polling. The polling cutoff is derived from that deadline using `virtual_poll_deadline_grace`.
 

src/reformatters/noaa/gefs/forecast_35_day_0_5_degree_virtual/dynamical_dataset.py

@@ -52,17 +52,18 @@ class NoaaGefsForecast35Day05DegreeVirtualDataset(
 
     def operational_kubernetes_resources(self, image_tag: str) -> Sequence[CronJob]:
         cron_job_name_prefix = self.dataset_id.replace("-0-5-degree", "-0-5")
-        # f000-f384 publishes ~init+3h46m through ~init+6h43m; f390-f840 publishes in
-        # bursts until ~init+28h05m. Fire just before the first burst; the 6h deadline
-        # covers it and early extension members; later members roll to the next fire.
+        # f000-f384 publishes ~init+3h46m to +6h43m. Extension member waves start
+        # near init+6h30m, +12h30m and +18h30m; overnight dumps finish around
+        # init+27h06m to +28h11m. Poll before the previous cycle finishes and through
+        # the newest morning extension wave, which can end around init+10h06m.
         operational_update_cron_job = ReformatCronJob(
             name=f"{cron_job_name_prefix}-update",
-            schedule="45 3 * * *",
-            pod_active_deadline=timedelta(hours=6),
+            schedule="15 2 * * *",
+            pod_active_deadline=timedelta(hours=8, minutes=30),
             image=image_tag,
             dataset_id=self.dataset_id,
             cpu="3.5",
-            # A fire opens by ingesting the whole extension of the previous cycle,
+            # A fire opens by catching up most of the previous cycle's extension,
             # the largest single batch of refs it holds.
             memory="15G",
             secret_names=self.store_factory.k8s_secret_names(),

src/reformatters/noaa/gefs/forecast_35_day_0_5_degree_virtual/region_job.py

@@ -9,5 +9,5 @@
 class NoaaGefsForecast35Day05DegreeVirtualRegionJob(NoaaGefsForecastVirtualRegionJob):
     """RegionJob for the GEFS 35 day 0.5 degree virtual forecast dataset."""
 
-    # Three update cron fires' span, so two consecutive missed runs still self-heal.
+    # Three daily inits retain the previous cycle's extension after a missed fire.
     operational_update_window: ClassVar[Timedelta] = pd.Timedelta("72h")

tests/noaa/gefs/forecast_35_day_0_5_degree_virtual/dynamical_dataset_test.py

@@ -45,9 +45,9 @@ def test_polling_leaves_validation_time_inside_single_deadline(
         )
     )
     (cron_job,) = dataset.operational_kubernetes_resources("test")
-    fire = pd.Timestamp("2026-09-27T03:45")
-    assert cron_job.pod_active_deadline == timedelta(hours=6)
-    assert dataset._virtual_poll_deadline(fire) == pd.Timestamp("2026-09-27T09:38")
+    fire = pd.Timestamp("2026-09-27T02:15")
+    assert cron_job.pod_active_deadline == timedelta(hours=8, minutes=30)
+    assert dataset._virtual_poll_deadline(fire) == pd.Timestamp("2026-09-27T10:38")
     assert dataset._virtual_poll_deadline(fire + pd.Timedelta("2h")) == (
         fire + cron_job.pod_active_deadline - timedelta(minutes=7)
     )
@@ -58,7 +58,14 @@ def test_polling_leaves_validation_time_inside_single_deadline(
     monkeypatch.setattr(sentry_sdk, "flush", Mock())
     with monitor_cron(cron_job, "test"):
         pass
-    assert capture.call_args_list[0].kwargs["monitor_config"]["max_runtime"] == 360
+    assert capture.call_args_list[0].kwargs["monitor_config"]["max_runtime"] == 510
+
+    monkeypatch.setattr("reformatters.common.operational._RUN_MONITORS", [monitor_cron])
+    monkeypatch.setattr(type(dataset), "_validate_dataset", Mock())
+    dataset.validate_dataset("test")
+    validation_monitor = capture.call_args_list[-2].kwargs["monitor_config"]
+    assert validation_monitor["schedule"]["value"] == "15 2 * * *"
+    assert validation_monitor["checkin_margin"] == 520
 
 
 # 40N 100W, a land cell so the soil and snow bitmaps carry values there.
@@ -309,19 +316,27 @@ def test_operational_kubernetes_resources(
 ) -> None:
     (update_cron_job,) = dataset.operational_kubernetes_resources("test-image-tag")
 
-    # The first file lands ~init+3h46m and the lead times through 384 hours finish
-    # ~init+6h43m, so the fire leads that stage and the deadline covers its end with
-    # three hours to spare. It also reaches past ~init+28h05m of the cycle a day older,
-    # where that cycle's 840 hour extension finishes.
-    assert update_cron_job.schedule == "45 3 * * *"
-    assert update_cron_job.pod_active_deadline == timedelta(hours=6)
+    # Poll before the previous cycle completes and through the newest morning
+    # extension, including a late member wave ending around 10:06 UTC.
+    assert update_cron_job.schedule == "15 2 * * *"
+    assert update_cron_job.pod_active_deadline == timedelta(hours=8, minutes=30)
     fire_to_next_fire = timedelta(hours=24)
     assert update_cron_job.pod_active_deadline < fire_to_next_fire
     (fire,) = _fire_minutes(update_cron_job.schedule)
-    previous_cycle_complete = timedelta(hours=28, minutes=5) - timedelta(days=1)
-    assert (
-        timedelta(minutes=fire) + update_cron_job.pod_active_deadline
-        > previous_cycle_complete
+    previous_cycle_complete = timedelta(hours=28, minutes=11) - timedelta(days=1)
+    assert timedelta(minutes=fire) < timedelta(hours=3, minutes=6)
+    poll_cutoff = (
+        timedelta(minutes=fire)
+        + update_cron_job.pod_active_deadline
+        - dataset.virtual_poll_deadline_grace
+    )
+    assert poll_cutoff > previous_cycle_complete
+    assert poll_cutoff > timedelta(hours=10, minutes=6)
+    validation_end = (
+        pd.Timestamp("2026-09-27T02:15") + update_cron_job.pod_active_deadline
+    )
+    assert update_cron_job.next_fire_time(validation_end) > (
+        validation_end + update_cron_job.pod_active_deadline
     )
     # Virtual updates are single writer.
     assert update_cron_job.workers_total == 1
@@ -334,14 +349,13 @@ def test_operational_kubernetes_resources(
 def test_operational_update_window_spans_three_update_fires(
     dataset: NoaaGefsForecast35Day05DegreeVirtualDataset,
 ) -> None:
-    """Two consecutive failed or lost updates still self-heal: the span the next fire
-    re-sweeps reaches back past both. A cycle's own extension also publishes past the
-    next fire, so even with none missed the window has to reach a second init."""
+    """A missed daily fire must not strand the previous cycle's extension, which
+    publishes past its own next fire."""
     (update_cron_job,) = dataset.operational_kubernetes_resources("test-image-tag")
     (fire,) = _fire_minutes(update_cron_job.schedule)
     window = NoaaGefsForecast35Day05DegreeVirtualRegionJob.operational_update_window
     assert window == 3 * _INIT_TIME_FREQUENCY
-    assert window > pd.Timedelta(minutes=fire) + pd.Timedelta("28h5m")
+    assert window >= 2 * _INIT_TIME_FREQUENCY + pd.Timedelta(minutes=fire)
 
 
 def test_cron_job_names_fit_the_kubernetes_limit(

tests/noaa/gefs/forecast_35_day_0_5_degree_virtual/region_job_test.py

@@ -175,7 +175,7 @@ def test_operational_update_jobs_reach_back_past_the_extension(
 ) -> None:
     """A cycle's 840 hour lead times publish until ~init+28h, past the next fire, so the
     window has to cover more than the newest init even when no run is missed."""
-    now = pd.Timestamp("2024-06-03T03:45")
+    now = pd.Timestamp("2024-06-03T02:15")
     monkeypatch.setattr(pd.Timestamp, "now", classmethod(lambda *a, **kw: now))
 
     jobs, template_ds = (