WIP: 🐛 Harden streaming energy path — 4 review findings (fix PR) #61
Loading…
Add table
Add a link
Reference in a new issue
No description provided.
Delete branch "fix/stream-robustness"
Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?
The deferred 'fix PR' for the Minor findings raised across the 5 streaming slices (#56-#60). Polish/robustness, no happy-path behavior change. Stacked on #60.
Fixes
_fetch_device_metrics(executor) snapshottedlive_total_wh, did the QUANTITY HTTP fetch (GIL released), then OVERWROTElive_total_whwith the stale snapshot — clobbering stream deltas thatintegrate_live_energy(event loop) added during the fetch. Now applies only the reconciliation drift aslive_total_wh += (live_total - initial_live), narrowing the race from seconds to a nanosecond RMW. Test injects a concurrent +30 during the fetch and asserts the delta survives (120.0, not the old 90.0).live_by_hourunbounded growth. After reconciliation, prune hours strictly older than the high-water mark (reconciled hours are never re-read).list()-snapshots keys to stay safe vs the event loop mutating the current-hour key._fold_capacity_mapKeyError.cap["nature"]→cap.get("nature")+ skip if absent, so one malformed capacity can't fail the whole refresh.dt_h <= 0guard inintegrate_live_energy(clock jump / duplicate burst: no delta, but last_power still advances). Code was already correct.Tests
4 new tests (Fix 1 is a real old-vs-new gate) + 1 assertion updated (a pruned hour becomes
not in, a correct consequence of Fix 2). 73/73 passing, ruff + mypy clean.stream.py/integrate_live_energy/UPDATE_INTERVAL/throttle/reconciliation-formula all unchanged.Stacking
Base:
feat/stream-resync(PR #60, tip of the streaming stack). Rebase down tomainas the stack merges.Fix 1 (thread-safety): _fetch_device_metrics snapshots initial_live before the QUANTITY HTTP fetch and applies only the reconciliation drift as state.live_total_wh += (live_total - initial_live), so concurrent stream deltas added by integrate_live_energy during the fetch are not clobbered. Fix 2 (live_by_hour pruning): after reconciliation, prune entries strictly older than hwm_hour (state.last_bucket_ts truncated to the hour), using a list() snapshot of keys to avoid mutation-during-iteration. Reconciled hours below the high-water mark are never read again. Fix 3 (_fold_capacity_map KeyError guard): read nature via cap.get('nature') and skip capacities where nature is None, avoiding KeyError propagation as UpdateFailed when a capacity has no nature key. Fix 4 (test-only): add test_integrate_live_energy_skips_delta_on_non_positive_dt to cover the dt_h <= 0 guard in integrate_live_energy. Also update test_reconcile_across_multiple_new_buckets: the 10:00 bucket is now pruned after reconciliation reaches hwm 11:00, so assert it is absent.Landed in
mainvia the linear stack fast-forward (commit9d016f7). This slice is part of the streaming feature merged as #56…#62 — closing (its commits are already inmain).Pull request closed