Hold a chart's last bin back until all of its data arrived

The newest metric bucket is upserted every flush while its minute runs, and
both rollup endpoints summed it in — so every curve on Home ended on a fall
that was only the clock. The timeseries now stops at the last closed bin, and
the flow rollups' window ends on the last closed minute, so all sixty slices
are whole ones.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/fluksio/api/routes/observability.py

@@ -124,6 +124,17 @@ def _since(hours: int) -> datetime:
     return datetime.now(UTC) - timedelta(hours=hours)
 
 
+def _closed_before(stride: int) -> datetime:
+    """The start of the bin `now` falls in — the newest one still filling.
+
+    A bucket is upserted every flush while its minute runs, so the newest bin
+    always holds part of a minute. Drawn, it reads as a fall that never
+    happened; excluded, the curve ends on the last bin that is all there.
+    """
+    now = int(datetime.now(UTC).timestamp())
+    return datetime.fromtimestamp(now // stride * stride, UTC)
+
+
 def _window_hours(hours: int) -> int:
     """A window the rollups can answer for: an hour at least, retention at most.
 
@@ -235,7 +246,10 @@ def read_timeseries(
         func.max(col(MetricBucket.duration_max_ms)).label("max_ms"),
         func.sum(col(MetricBucket.lag_sum_ms)).label("lag_sum_ms"),
         func.sum(col(MetricBucket.items)).label("items"),
-    ).where(col(MetricBucket.bucket) >= _since(hours))
+    ).where(
+        col(MetricBucket.bucket) >= _since(hours),
+        col(MetricBucket.bucket) < _closed_before(stride),
+    )
     if flow:
         statement = statement.where(col(MetricBucket.flow) == flow)
     if node:
@@ -259,9 +273,14 @@ def read_timeseries(
 def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any:
     """One row per flow, with a coarse trend of how much it ran."""
     hours = _window_hours(hours)
-    since = _since(hours)
     window = hours * 3600
-    start = since.timestamp()
+    # The window ends on the last closed minute rather than on `now`, so every
+    # slice it hands back is a whole one. Ending at `now` cut the last slice
+    # wherever the request happened to land, and the trend fell off a cliff
+    # that was only the clock.
+    end = int(datetime.now(UTC).timestamp()) // 60 * 60
+    start = float(end - window)
+    since = datetime.fromtimestamp(start, UTC)
 
     # Binned to the sparkline slice rather than the minute, so a flow costs at
     # most SPARK_SLICES rows however long the window is. The slice is the
@@ -282,7 +301,10 @@ def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any:
             func.sum(col(MetricBucket.lag_sum_ms)).label("lag_sum_ms"),
             func.sum(col(MetricBucket.items)).label("items"),
         )
-        .where(col(MetricBucket.bucket) >= since)
+        .where(
+            col(MetricBucket.bucket) >= since,
+            col(MetricBucket.bucket) < datetime.fromtimestamp(float(end), UTC),
+        )
         .group_by(col(MetricBucket.flow), slot)
     )
 

backend/tests/api/routes/test_observability.py

@@ -195,6 +195,47 @@ def test_the_timeseries_folds_into_the_requested_bucket(
     assert points[0]["ts"] % 900 == 0
 
 
+def test_the_bucket_still_filling_is_held_back(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    """The minute in progress is not drawn.
+
+    A bucket is upserted every flush while its minute runs, so the newest one
+    only ever holds part of a minute. Charted, it reads as a fall that never
+    happened.
+    """
+    flow = "open-bucket-test"
+    minute = int(datetime.now(UTC).timestamp()) // 60 * 60
+    for offset, executions in ((-60, 3), (0, 1)):
+        db.add(
+            MetricBucket(
+                flow=flow,
+                node=f"{flow}.calc",
+                bucket=datetime.fromtimestamp(minute + offset, UTC),
+                executions=executions,
+            )
+        )
+    db.commit()
+
+    points = client.get(
+        f"{PREFIX}/timeseries",
+        headers=superuser_token_headers,
+        params={"flow": flow, "hours": 1},
+    ).json()
+    flows = client.get(
+        f"{PREFIX}/flows", headers=superuser_token_headers, params={"hours": 1}
+    ).json()
+    row = next(entry for entry in flows if entry["flow"] == flow)
+
+    # Unless that minute closed while the requests were in flight — then both
+    # buckets are complete and counting both is right.
+    if int(datetime.now(UTC).timestamp()) // 60 * 60 == minute:
+        assert [int(point["ts"]) for point in points] == [minute - 60]
+        # The window ends on the last closed minute, so the final slice is a
+        # whole one rather than however much of this minute has arrived.
+        assert (row["executions"], row["spark"][-1]) == (3, 3)
+
+
 def test_a_zero_hour_window_is_still_an_hour(
     client: TestClient, superuser_token_headers: dict[str, str]
 ) -> None: