From 9d3e28a529974c3a673b027f6c3c72c757a1d514 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Wed, 24 Jun 2026 12:44:05 +0200 Subject: [PATCH] Billing fix: don't bill periods from stale/out-of-range DSMR readings Root cause (found in the live DB): register_at() returned the latest reading for any boundary beyond the DSMR data range (a future instant or a long gap), so future periods got start==end -> delta 0 and were written as degraded=False 'ok' rows with cost 0. recompute_range() over a future end then filled the whole month with such fake rows, which then occupied the slots so the per-minute tick (immutability guard skips non-degraded rows) never recomputed the real data. - register_at: ignore a reading older than one period (15min) before the boundary -> returns None -> period is marked degraded instead of a fake 0. - recompute_range: only process closed periods (t1 <= now); never future ones. - Tests cover staleness window, out-of-range -> degraded, recompute skips future, and a regression that normal closed periods are unaffected. --- app/services/energy_cost.py | 42 +++++-- tests/test_energy_cost.py | 220 ++++++++++++++++++++++++++++++++++++ 2 files changed, 253 insertions(+), 9 deletions(-) diff --git a/app/services/energy_cost.py b/app/services/energy_cost.py index 8e4a2ce..7a54437 100644 --- a/app/services/energy_cost.py +++ b/app/services/energy_cost.py @@ -71,6 +71,15 @@ logger = logging.getLogger(__name__) _PERIOD_MINUTES = 15 _LOOKBACK_DAYS = 7 # maximum lookback window for compute_closed_periods +# Maximum age a DSMR reading may have relative to the boundary being queried. +# Under normal operation DSMR readings arrive every ~10 seconds, so a reading +# more than one period (15 minutes) old at the boundary indicates either a data +# gap or — critically — a *future* boundary being resolved against the last +# historical reading. In both cases the reading is considered stale and +# ``register_at`` returns None, letting the period be marked degraded instead of +# producing a spurious zero-delta "successful" row. +_READING_MAX_STALENESS = timedelta(minutes=_PERIOD_MINUTES) + # DSMR payload register keys (cumulative kWh, JSON string values). _KEY_D1 = "electricity_delivered_1" # delivered low-tariff (dal / _1) _KEY_D2 = "electricity_delivered_2" # delivered high-tariff (normal / _2) @@ -148,6 +157,13 @@ def register_at(session: Session, boundary: datetime) -> dict[str, Decimal] | No if row is None: return None + # Freshness guard: reject readings that are too old relative to *boundary*. + # ``recorded_at`` is stored as a naive UTC datetime in SQLite; attach UTC + # tzinfo before comparing with *boundary* (which is always tz-aware UTC) to + # avoid an "offset-naive vs offset-aware" TypeError. + if _as_utc(row.recorded_at) < _as_utc(boundary) - _READING_MAX_STALENESS: + return None + payload = row.payload or {} try: d1_raw = payload[_KEY_D1] @@ -437,18 +453,26 @@ def recompute_range(session: Session, start: datetime, end: datetime) -> int: """ t0 = floor_to_quarter(_as_utc(start)) end_utc = _as_utc(end) + now = datetime.now(UTC) written = 0 while t0 < end_utc: - try: - did_write = compute_period(session, t0, overwrite=True) - if did_write: - written += 1 - except Exception: - logger.exception( - "recompute_range: unexpected error for t0=%s — continuing.", - t0.isoformat(), - ) + t1 = t0 + timedelta(minutes=_PERIOD_MINUTES) + # Only recompute periods that have already closed (t1 ≤ now). Even + # when the caller passes a future *end*, we must not write speculative + # "future" rows: register_at would resolve to the latest historical + # reading for both boundaries, producing a zero-delta fake-success row + # that blocks the real computation once the period actually closes. + if t1 <= now: + try: + did_write = compute_period(session, t0, overwrite=True) + if did_write: + written += 1 + except Exception: + logger.exception( + "recompute_range: unexpected error for t0=%s — continuing.", + t0.isoformat(), + ) t0 += timedelta(minutes=_PERIOD_MINUTES) session.commit() diff --git a/tests/test_energy_cost.py b/tests/test_energy_cost.py index a078c3c..e2eb38e 100644 --- a/tests/test_energy_cost.py +++ b/tests/test_energy_cost.py @@ -1195,3 +1195,223 @@ class TestRecomputeRange: assert row_after.r2_kwh == 0.0 assert row_after.contract_version_id is None assert row_after.pricing == {} + + +# --------------------------------------------------------------------------- +# Bug-fix regression tests: register_at freshness + recompute_range future guard +# --------------------------------------------------------------------------- + + +class TestRegisterAtFreshness: + """Tests for the staleness guard added to register_at (Fix A). + + Normal DSMR readings arrive every ~10 seconds. A reading more than + ``_READING_MAX_STALENESS`` (15 minutes) old at a given boundary is treated + as absent so the period is marked degraded rather than producing a + zero-delta fake-success row. + """ + + def test_stale_reading_returns_none(self, energy_db: Session) -> None: + """A reading older than 15 min before the boundary must be rejected.""" + reading_time = _ts(10, 0) # 10:00 + boundary = _ts(10, 30) # 10:30 — 30 min later (> 15 min staleness) + _make_reading(energy_db, recorded_at=reading_time, source_id=1) + energy_db.commit() + + result = register_at(energy_db, boundary) + assert result is None, ( + "register_at must return None when the closest reading is more than " + "15 minutes before the boundary" + ) + + def test_fresh_reading_within_window_returned(self, energy_db: Session) -> None: + """A reading within 15 min of the boundary must be returned normally.""" + reading_time = _ts(10, 5) # 10:05 + boundary = _ts(10, 15) # 10:15 — only 10 min gap (within window) + _make_reading(energy_db, recorded_at=reading_time, + d1="30000.0", d2="15000.0", r1="1000.0", r2="500.0", + source_id=1) + energy_db.commit() + + result = register_at(energy_db, boundary) + assert result is not None, ( + "register_at must return the reading when it is within the 15-min staleness window" + ) + assert result["d1"] == Decimal("30000.0") + + def test_exact_15min_boundary_is_accepted(self, energy_db: Session) -> None: + """A reading exactly 15 min before the boundary sits at the edge — accepted.""" + reading_time = _ts(10, 0) # 10:00 + boundary = _ts(10, 15) # 10:15 — exactly 15 min gap + _make_reading(energy_db, recorded_at=reading_time, + d1="40000.0", d2="20000.0", r1="2000.0", r2="1000.0", + source_id=1) + energy_db.commit() + + result = register_at(energy_db, boundary) + # boundary - reading == 15 min == staleness limit → NOT stale (strict <) + assert result is not None + + def test_future_boundary_returns_none(self, energy_db: Session) -> None: + """A far-future boundary must return None (not the latest historical reading). + + This is the root cause of the production bug: without the freshness + guard, register_at(far_future) returned the last available reading, + causing both period boundaries to resolve to the same row and producing + a zero-delta fake-success period. + """ + # Insert a reading at a realistic "now" time. + reading_time = _ts(10, 0) # 10:00 on 2026-06-23 + _make_reading(energy_db, recorded_at=reading_time, source_id=1) + energy_db.commit() + + # Query with a far-future boundary (e.g. end of month, same day +8 hours) + far_future_boundary = _ts(18, 0) # 18:00 — 8 hours later + result = register_at(energy_db, far_future_boundary) + assert result is None, ( + "register_at must return None for a far-future boundary rather than " + "the latest historical reading" + ) + + +class TestFuturePeriodDegraded: + """Ensure that a period whose boundaries fall entirely outside DSMR data + is written as degraded=True (not as a fake zero-cost success). + + This covers the production scenario where recompute_range was called with + a future end date, causing compute_period to be called on not-yet-closed + periods. With Fix A (register_at freshness) the boundary lookups return + None → degraded; with Fix B (recompute_range t1<=now guard) such periods + are skipped entirely. This test exercises the Fix A path directly via + compute_period(overwrite=True) on a future-like period with no nearby data. + """ + + def test_out_of_range_period_is_degraded(self, energy_db: Session) -> None: + """A period whose boundaries have no fresh DSMR readings → degraded=True. + + We insert a reading at 10:00 and call compute_period for a period + starting at 14:00 (4 hours later). The boundary readings (14:00 and + 14:15) are both more than 15 min from any available data, so + register_at returns None for both → degraded row written. + """ + contract = _make_contract(energy_db, kind="manual", active=True) + _make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0)) + # Only a reading at 10:00 — far from the 14:00/14:15 boundaries. + _make_reading(energy_db, recorded_at=_ts(10, 0), source_id=1) + energy_db.commit() + + future_t0 = _ts(14, 0) + result = compute_period(energy_db, future_t0, overwrite=True) + assert result is True # a record was written + + row = energy_db.execute( + select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == future_t0) + ).scalar_one() + assert row.degraded is True, ( + "compute_period must write degraded=True when both boundaries lack fresh readings" + ) + assert row.import_cost == 0.0 + assert row.net_cost == 0.0 + + +class TestRecomputeRangeNoFuturePeriods: + """Verify that recompute_range never writes periods whose t1 > now (Fix B).""" + + def test_recompute_range_skips_future_periods(self, energy_db: Session) -> None: + """When end is in the future, recompute_range must not write any future rows. + + This is the direct test for Fix B. We call recompute_range with a + future end datetime; the only periods that may be written are those + where t1 <= now. No row with period_start > now should appear in the DB. + """ + contract = _make_contract(energy_db, kind="manual", active=True) + _make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0)) + # No readings needed — we only care that future rows are NOT written. + energy_db.commit() + + now = datetime.now(UTC) + far_future_end = now + timedelta(days=7) + # Use a past start so there are some candidate periods. + past_start = floor_to_quarter(now - timedelta(minutes=30)) + + recompute_range(energy_db, past_start, far_future_end) + + # Assert no row has period_start beyond now. + rows = energy_db.execute(select(EnergyCostPeriod)).scalars().all() + future_rows = [ + r for r in rows + if (r.period_start if r.period_start.tzinfo else r.period_start.replace(tzinfo=UTC)) > now + ] + assert future_rows == [], ( + f"recompute_range must not write future periods; " + f"found {len(future_rows)} row(s) with period_start > now" + ) + + def test_recompute_range_chain_no_slot_squatting(self, energy_db: Session) -> None: + """Simulate the full production failure scenario and verify it is fixed. + + Scenario: + 1. recompute_range is called with end=far_future (reproduces the bug trigger). + 2. compute_closed_periods is then called (reproduces the normal tick). + 3. After the fix, step 1 must not have written any future rows, so the + tick in step 2 is free to compute periods normally. + + We only verify that no future rows exist after step 1 (Fix B), because + that is the core slot-squatting prevention. + """ + contract = _make_contract(energy_db, kind="manual", active=True) + # Contract effective from far past so all periods in scope have a version. + _make_version( + energy_db, contract, _MANUAL_VALUES, + effective_from=datetime(2020, 1, 1, tzinfo=UTC), + ) + energy_db.commit() + + now = datetime.now(UTC) + far_future = now + timedelta(days=30) + past_start = floor_to_quarter(now - timedelta(hours=1)) + + # Step 1: buggy call with future end. + recompute_range(energy_db, past_start, far_future) + + # Verify no future rows were squatted. + all_rows = energy_db.execute(select(EnergyCostPeriod)).scalars().all() + future_rows = [ + r for r in all_rows + if (r.period_start if r.period_start.tzinfo else r.period_start.replace(tzinfo=UTC)) > now + ] + assert future_rows == [], ( + f"After recompute_range with future end, found {len(future_rows)} future row(s). " + "These would block the real tick from computing those periods." + ) + + +class TestNormalPeriodsUnaffected: + """Regression: the freshness guard must not break computation of normal + closed periods where readings are available close to the boundaries. + """ + + def test_normal_closed_period_still_computes_ok(self, energy_db: Session) -> None: + """A normal period with readings seconds before each boundary → non-degraded.""" + contract = _make_contract(energy_db, kind="manual", active=True) + _make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0)) + + # Readings placed very close to boundaries (as DSMR normally delivers them). + # t0=10:00, reading at 09:59:50 (10 s before); t1=10:15, reading at 10:14:55 (5 s before) + _make_reading(energy_db, recorded_at=_ts(9, 59, 50), + d1=_START_D1, d2=_START_D2, r1=_START_R1, r2=_START_R2, source_id=1) + _make_reading(energy_db, recorded_at=_ts(10, 14, 55), + d1=_END_D1, d2=_END_D2, r1=_END_R1, r2=_END_R2, source_id=2) + energy_db.commit() + + result = compute_period(energy_db, _T0) + assert result is True + + row = energy_db.execute( + select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0) + ).scalar_one() + assert row.degraded is False, ( + "A normal period with readings just before each boundary must not be degraded" + ) + # import_cost matches hand-calc: same deltas as standard scenario + assert abs(row.import_cost - 0.4101) < 1e-6