Harden mqtt_logger against partial payloads, stale cache values, and silent reconnects #4

agent-pi · 2026-09-09T01:35:40Z

agent-pi commented

2026-09-09 01:35:40 +00:00

Three related fixes to the MQTT stats logger, developed TDD-style with a new standalone unittest suite (test_mqtt_logger.py, matching the test_analyze.py conventions).

Atomic payload extraction — a JSON topic payload missing one of its expected fields used to leave the fields before it already merged into the cache, producing records that mixed half-delivered values with stale cache. Extraction now happens into a temporary dict merged only when the payload is complete. Closes #2.

Per-field freshness ages — every record now carries an _age_s map (seconds since each field was last published), so analysis can reject records whose cached sources (forecasts, Zappi state, weather) went quiet instead of mistaking stale values for fresh ones. Closes #3.

Connection markers + resubscribe — on_connect/on_disconnect emit JSON event lines into the same stream, making broker restarts and network gaps attributable in the stats file (the 2026-07-03..13 gap had no forensics trail). on_connect also resubscribes to every topic: paho uses a clean session and does not restore subscriptions, so a broker restart previously risked a connected-but-deaf logger.

Verified against a local mosquitto lifecycle: connect marker, record, broker kill, disconnect marker, broker restart, connect marker, records resume; 11 logger tests + 6 analyze tests green.

Also documents the new metadata in mqtt-topic-logger.md / AGENTS.md, and lands the Working Conventions change adopting the standard forge workflow.

Closes #2
Closes #3

Three related fixes to the MQTT stats logger, developed TDD-style with a new standalone unittest suite (`test_mqtt_logger.py`, matching the `test_analyze.py` conventions). **Atomic payload extraction** — a JSON topic payload missing one of its expected fields used to leave the fields before it already merged into the cache, producing records that mixed half-delivered values with stale cache. Extraction now happens into a temporary dict merged only when the payload is complete. Closes #2. **Per-field freshness ages** — every record now carries an `_age_s` map (seconds since each field was last published), so analysis can reject records whose cached sources (forecasts, Zappi state, weather) went quiet instead of mistaking stale values for fresh ones. Closes #3. **Connection markers + resubscribe** — `on_connect`/`on_disconnect` emit JSON event lines into the same stream, making broker restarts and network gaps attributable in the stats file (the 2026-07-03..13 gap had no forensics trail). `on_connect` also resubscribes to every topic: paho uses a clean session and does not restore subscriptions, so a broker restart previously risked a connected-but-deaf logger. Verified against a local mosquitto lifecycle: connect marker, record, broker kill, disconnect marker, broker restart, connect marker, records resume; 11 logger tests + 6 analyze tests green. Also documents the new metadata in `mqtt-topic-logger.md` / `AGENTS.md`, and lands the Working Conventions change adopting the standard forge workflow. Closes #2 Closes #3

agent-pi added 3 commits

2026-09-09 01:35:40 +00:00

🐛 reject incomplete MQTT payloads without touching the field cache 8275bf958f

A JSON topic payload missing one of its expected fields used to leave the
fields before it already merged into the cache, so the next emitted record
could mix half-delivered values with stale cache. Extraction now happens
into a temporary dict that is merged only when the payload is complete.

Closes #2

✨ add per-field freshness ages and connect/disconnect markers dbfffcd9a3

Each record now carries an _age_s map (seconds since each field's source
topic last published) so analysis can reject records whose cached sources
went quiet. on_connect/on_disconnect emit JSON event lines into the same
stream, making broker restarts and network gaps attributable in the stats
file. on_connect resubscribes to all topics: the client uses a clean
session, and paho does not restore subscriptions, so a broker restart
previously risked a connected-but-deaf logger.

Closes #3

📝 document logger metadata; adopt standard forge workflow in project guide 3821d897aa

mqtt-topic-logger.md: all-or-nothing extraction, _age_s freshness map,
connection markers, resubscribe-on-connect, corrected the record example
(local timestamp, not UTC), extended verification steps.

AGENTS.md: new Logger metadata note (2026-09-09) and Working Conventions
now follow the standard global forge workflow (issues, branches, PRs).

agent-pi referenced this pull request

2026-09-09 01:56:45 +00:00

analyze.py counts logger connection-marker lines as data records #5

agent-pi referenced this pull request

2026-09-09 01:58:05 +00:00

Skip logger event-marker lines when loading stats records #6

marvin8 requested changes

2026-09-09 03:19:05 +00:00

Dismissed

mqtt_logger.py Outdated

					
				@ -112,1 +113,4 @@

				fresh: set = set()

				# field -> monotonic() reading of its last successful publish; drives _age_s

				# so downstream analysis can reject records carrying stale cached values.

				published_at: dict = {}

Do we at this point already know what types store in this dict? If so we should be more specific with our typing!

agent-pi added 1 commit

2026-09-09 03:27:28 +00:00

🎨 type the published_at cache as dict[str, float] 169eecf1b6

Every value written is a time.monotonic() reading, so the dict can carry
a precise annotation instead of bare dict.

agent-pi commented

2026-09-09 03:27:41 +00:00

mqtt_logger.py line 116

Do we at this point already know what types store in this dict? If so we should be more specific with our typing!

✅ Yes — every value written is a time.monotonic() reading, so the keys are field names and the values are floats. Typed as dict[str, float] in commit 169eecf.

The two sibling declarations are vaguer than they could be — want me to tighten those too? latest: dict[str, str | int | float | bool] (raw topics store decoded strings; JSON topics yield scalars) and fresh: set[str].

[`mqtt_logger.py` line 116](https://forge.marvin8.zone/marvin8/IrradiationToPowerProduction/pulls/4#issuecomment-2334) > Do we at this point already know what types store in this dict? If so we should be more specific with our typing! ✅ Yes — every value written is a `time.monotonic()` reading, so the keys are field names and the values are floats. Typed as `dict[str, float]` in commit `169eecf`. The two sibling declarations are vaguer than they could be — want me to tighten those too? `latest: dict[str, str | int | float | bool]` (raw topics store decoded strings; JSON topics yield scalars) and `fresh: set[str]`.

agent-pi added 1 commit

2026-09-09 03:30:58 +00:00

🎨 type the latest cache and fresh set precisely 25ddb19d8e

latest holds JSON scalars only: raw-key topics store decoded strings,
JSON-extracted fields are str/int/float/bool. fresh collects topic
names.

agent-pi commented

2026-09-09 03:31:07 +00:00

Following up on the published_at typing thread — both sibling declarations tightened in commit 25ddb19:

latest: dict[str, str | int | float | bool] — raw-key topics store decoded strings, JSON-extracted fields are scalars
fresh: set[str] — topic names

All 11 logger tests green.

Following up on the [`published_at` typing thread](https://forge.marvin8.zone/marvin8/IrradiationToPowerProduction/pulls/4#issuecomment-2334) — both sibling declarations tightened in commit `25ddb19`: - `latest: dict[str, str | int | float | bool]` — raw-key topics store decoded strings, JSON-extracted fields are scalars - `fresh: set[str]` — topic names All 11 logger tests green.

marvin8 approved these changes

2026-09-09 03:34:28 +00:00

Dismissed

agent-pi force-pushed fix/issue-2-mqtt-logger-quality from 25ddb19d8e to 78ebbb453b

2026-09-09 03:37:55 +00:00

Compare

agent-pi dismissed marvin8's review

2026-09-09 03:37:55 +00:00

Reason:

New commits pushed, approval review dismissed automatically according to repository settings

marvin8 approved these changes

2026-09-09 03:42:16 +00:00