From e8846f48319b78dc9037b3584d3ae7bc8e97c7b6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AD=8F=E9=A3=8E?= Date: Thu, 19 Mar 2026 22:08:16 +0800 Subject: [PATCH] refactor: simplify mqtt pipeline state and remove event table --- backend/README.md | 4 +- ...c5d6e7f809_simplify_mqtt_parse_tracking.py | 108 ++++++++++++++++++ ...8f90123_drop_terminal_observation_event.py | 60 ++++++++++ backend/app/models.py | 25 +--- backend/app/mqtt/ingestor.py | 29 +++-- backend/app/mqtt/parser_worker.py | 18 +-- backend/app/mqtt/parsers/__init__.py | 3 +- backend/app/mqtt/parsers/mixed.py | 22 +++- backend/app/mqtt/parsers/rtcm.py | 22 ---- backend/app/mqtt/repository.py | 29 +---- backend/app/mqtt/topic.py | 10 -- backend/tests/mqtt/test_models.py | 3 +- backend/tests/mqtt/test_parser_worker.py | 52 +++++++-- backend/tests/mqtt/test_topic.py | 2 +- 14 files changed, 269 insertions(+), 118 deletions(-) create mode 100644 backend/app/alembic/versions/b4c5d6e7f809_simplify_mqtt_parse_tracking.py create mode 100644 backend/app/alembic/versions/c6d7e8f90123_drop_terminal_observation_event.py delete mode 100644 backend/app/mqtt/parsers/rtcm.py delete mode 100644 backend/app/mqtt/topic.py diff --git a/backend/README.md b/backend/README.md index dcc7794..f7caf70 100644 --- a/backend/README.md +++ b/backend/README.md @@ -176,7 +176,8 @@ Once you have the MJML extension installed, you can create a new email template This project includes a two-stage MQTT pipeline: - `mqtt-ingestor` subscribes to `terminal/{device_id}/raw` and stores payload bytes losslessly in `terminal_message_raw`. -- `mqtt-parser-worker` consumes pending rows and writes parsed events/state. +- `mqtt-parser-worker` consumes unparsed rows (`parsed_at IS NULL` and `parse_error IS NULL`) and updates `terminal_device_state`. +- Parse is single-pass by design in the current implementation: failed rows keep `parse_error` and are not auto-retried with backoff. ### Run locally with Docker Compose @@ -199,7 +200,6 @@ $ PYTHONPATH=. ../.venv/bin/pytest tests/mqtt --confcutdir=tests/mqtt -q 2. Publish a sample message to `terminal/device-001/raw`. 3. Query PostgreSQL tables: - `terminal_message_raw` - - `terminal_observation_event` - `terminal_device_state` ### Archive run (manual trigger) diff --git a/backend/app/alembic/versions/b4c5d6e7f809_simplify_mqtt_parse_tracking.py b/backend/app/alembic/versions/b4c5d6e7f809_simplify_mqtt_parse_tracking.py new file mode 100644 index 0000000..7d60ea8 --- /dev/null +++ b/backend/app/alembic/versions/b4c5d6e7f809_simplify_mqtt_parse_tracking.py @@ -0,0 +1,108 @@ +"""simplify mqtt parse tracking + +Revision ID: b4c5d6e7f809 +Revises: a1b2c3d4e5f6 +Create Date: 2026-03-19 17:30:00.000000 + +""" + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = "b4c5d6e7f809" +down_revision = "a1b2c3d4e5f6" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "terminal_message_raw", + sa.Column("parsed_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index( + op.f("ix_terminal_message_raw_parsed_at"), + "terminal_message_raw", + ["parsed_at"], + unique=False, + ) + + # Preserve historical parse outcomes before removing state-machine columns. + op.execute( + sa.text( + """ + UPDATE terminal_message_raw + SET parsed_at = COALESCE(received_at, created_at) + WHERE parse_status <> 'pending' + """ + ) + ) + + op.drop_index( + op.f("ix_terminal_message_raw_next_retry_at"), + table_name="terminal_message_raw", + ) + op.drop_index( + op.f("ix_terminal_message_raw_parse_status"), + table_name="terminal_message_raw", + ) + op.drop_column("terminal_message_raw", "next_retry_at") + op.drop_column("terminal_message_raw", "parse_attempts") + op.drop_column("terminal_message_raw", "parse_status") + + +def downgrade() -> None: + op.add_column( + "terminal_message_raw", + sa.Column( + "parse_status", + sa.String(length=32), + nullable=False, + server_default="pending", + ), + ) + op.add_column( + "terminal_message_raw", + sa.Column("parse_attempts", sa.Integer(), nullable=False, server_default="0"), + ) + op.add_column( + "terminal_message_raw", + sa.Column("next_retry_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index( + op.f("ix_terminal_message_raw_parse_status"), + "terminal_message_raw", + ["parse_status"], + unique=False, + ) + op.create_index( + op.f("ix_terminal_message_raw_next_retry_at"), + "terminal_message_raw", + ["next_retry_at"], + unique=False, + ) + + op.execute( + sa.text( + """ + UPDATE terminal_message_raw + SET parse_status = CASE + WHEN parsed_at IS NULL THEN 'pending' + WHEN parse_error IS NULL THEN 'succeeded' + ELSE 'failed' + END, + parse_attempts = CASE + WHEN parsed_at IS NULL THEN 0 + ELSE 1 + END + """ + ) + ) + + op.drop_index( + op.f("ix_terminal_message_raw_parsed_at"), + table_name="terminal_message_raw", + ) + op.drop_column("terminal_message_raw", "parsed_at") diff --git a/backend/app/alembic/versions/c6d7e8f90123_drop_terminal_observation_event.py b/backend/app/alembic/versions/c6d7e8f90123_drop_terminal_observation_event.py new file mode 100644 index 0000000..15dd9f9 --- /dev/null +++ b/backend/app/alembic/versions/c6d7e8f90123_drop_terminal_observation_event.py @@ -0,0 +1,60 @@ +"""drop terminal observation event table + +Revision ID: c6d7e8f90123 +Revises: b4c5d6e7f809 +Create Date: 2026-03-19 18:20:00.000000 + +""" + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = "c6d7e8f90123" +down_revision = "b4c5d6e7f809" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.drop_table("terminal_observation_event") + + +def downgrade() -> None: + op.create_table( + "terminal_observation_event", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("raw_id", sa.Uuid(), nullable=False), + sa.Column("device_id", sa.String(length=255), nullable=False), + sa.Column("event_time", sa.DateTime(timezone=True), nullable=False), + sa.Column("event_type", sa.String(length=64), nullable=False), + sa.Column("payload_json", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["raw_id"], ["terminal_message_raw.id"]), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + op.f("ix_terminal_observation_event_device_id"), + "terminal_observation_event", + ["device_id"], + unique=False, + ) + op.create_index( + op.f("ix_terminal_observation_event_event_time"), + "terminal_observation_event", + ["event_time"], + unique=False, + ) + op.create_index( + op.f("ix_terminal_observation_event_event_type"), + "terminal_observation_event", + ["event_type"], + unique=False, + ) + op.create_index( + op.f("ix_terminal_observation_event_raw_id"), + "terminal_observation_event", + ["raw_id"], + unique=False, + ) diff --git a/backend/app/models.py b/backend/app/models.py index bedfd0e..1dc3847 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -126,9 +126,7 @@ class TerminalMessageRaw(SQLModel, table=True): payload_sha256: str = Field(default="", max_length=64, index=True) qos: int = 0 retain: bool = False - parse_status: str = Field(default="pending", max_length=32, index=True) - parse_attempts: int = 0 - next_retry_at: datetime | None = Field( + parsed_at: datetime | None = Field( default=None, sa_type=DateTime(timezone=True), # type: ignore index=True, @@ -140,27 +138,6 @@ class TerminalMessageRaw(SQLModel, table=True): ) -class TerminalObservationEvent(SQLModel, table=True): - __tablename__ = "terminal_observation_event" - - id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) - raw_id: uuid.UUID = Field( - foreign_key="terminal_message_raw.id", nullable=False, index=True - ) - device_id: str = Field(max_length=255, index=True) - event_time: datetime = Field( - default_factory=get_datetime_utc, - sa_type=DateTime(timezone=True), # type: ignore - index=True, - ) - event_type: str = Field(max_length=64, index=True) - payload_json: dict[str, Any] = Field(default_factory=dict, sa_type=JSON) - created_at: datetime = Field( - default_factory=get_datetime_utc, - sa_type=DateTime(timezone=True), # type: ignore - ) - - class TerminalDeviceState(SQLModel, table=True): __tablename__ = "terminal_device_state" diff --git a/backend/app/mqtt/ingestor.py b/backend/app/mqtt/ingestor.py index a50fa02..152eb85 100644 --- a/backend/app/mqtt/ingestor.py +++ b/backend/app/mqtt/ingestor.py @@ -1,24 +1,32 @@ from __future__ import annotations import logging +import re +from typing import Any -import paho.mqtt.client as mqtt # type: ignore[import-not-found,import-untyped] from sqlmodel import Session from app.core.config import settings from app.core.db import engine from app.mqtt.repository import insert_raw_message -from app.mqtt.topic import extract_device_id logger = logging.getLogger(__name__) +TOPIC_PATTERN = re.compile(r"^terminal/(?P[^/]+)/raw$") + + +def extract_device_id(topic: str) -> str: + match = TOPIC_PATTERN.match(topic) + if match is None: + raise ValueError("topic must match terminal/{device_id}/raw") + return match.group("device_id") def _on_connect( - client: mqtt.Client, + client: Any, _userdata: object, _flags: dict[str, int], reason_code: int, - _properties: mqtt.Properties | None = None, + _properties: object | None = None, ) -> None: if reason_code != 0: logger.error("MQTT connect failed: rc=%s", reason_code) @@ -31,9 +39,9 @@ def _on_connect( def _on_message( - _client: mqtt.Client, + _client: Any, _userdata: object, - message: mqtt.MQTTMessage, + message: Any, ) -> None: try: device_id = extract_device_id(message.topic) @@ -52,7 +60,14 @@ def _on_message( ) -def create_client() -> mqtt.Client: +def create_client() -> Any: + try: + import paho.mqtt.client as mqtt # type: ignore[import-not-found,import-untyped] + except ModuleNotFoundError as exc: # pragma: no cover - runtime guard + raise RuntimeError( + "paho-mqtt is required to run mqtt-ingestor. Install backend dependencies first." + ) from exc + client = mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id=settings.MQTT_CLIENT_ID, diff --git a/backend/app/mqtt/parser_worker.py b/backend/app/mqtt/parser_worker.py index 644352f..70de61e 100644 --- a/backend/app/mqtt/parser_worker.py +++ b/backend/app/mqtt/parser_worker.py @@ -2,45 +2,33 @@ from __future__ import annotations import logging import time -from datetime import timedelta from sqlmodel import Session from app.core.config import settings from app.core.db import engine -from app.models import TerminalMessageRaw, get_datetime_utc +from app.models import TerminalMessageRaw from app.mqtt.parsers.mixed import parse_mixed_payload from app.mqtt.repository import ( fetch_pending_rows_for_update, mark_failed, - mark_processing, mark_succeeded, - write_events_and_update_state, + update_device_state_from_events, ) logger = logging.getLogger(__name__) -def _next_retry_delay_seconds(attempts: int) -> int: - # Bounded exponential backoff: 30s, 60s, 120s, ... capped to 15m. - return int(min(30 * (2 ** max(attempts - 1, 0)), 900)) - - def _process_one_row(*, session: Session, row: TerminalMessageRaw) -> None: - mark_processing(session=session, row=row) - session.commit() - try: events = parse_mixed_payload(row.payload_bytes) - write_events_and_update_state(session=session, row=row, events=events) + update_device_state_from_events(session=session, row=row, events=events) mark_succeeded(session=session, row=row) except Exception as exc: # pragma: no cover - safety net path - delay = _next_retry_delay_seconds(row.parse_attempts) mark_failed( session=session, row=row, parse_error=str(exc), - next_retry_at=get_datetime_utc() + timedelta(seconds=delay), ) session.commit() diff --git a/backend/app/mqtt/parsers/__init__.py b/backend/app/mqtt/parsers/__init__.py index 04bc14e..15ffcb4 100644 --- a/backend/app/mqtt/parsers/__init__.py +++ b/backend/app/mqtt/parsers/__init__.py @@ -1,5 +1,4 @@ -from app.mqtt.parsers.mixed import parse_mixed_payload +from app.mqtt.parsers.mixed import parse_mixed_payload, summarize_rtcm_types from app.mqtt.parsers.nmea import parse_nmea_sentences -from app.mqtt.parsers.rtcm import summarize_rtcm_types __all__ = ["parse_mixed_payload", "parse_nmea_sentences", "summarize_rtcm_types"] diff --git a/backend/app/mqtt/parsers/mixed.py b/backend/app/mqtt/parsers/mixed.py index df0f864..75f9f12 100644 --- a/backend/app/mqtt/parsers/mixed.py +++ b/backend/app/mqtt/parsers/mixed.py @@ -3,7 +3,27 @@ from __future__ import annotations from typing import Any from app.mqtt.parsers.nmea import parse_nmea_sentences -from app.mqtt.parsers.rtcm import summarize_rtcm_types + + +def summarize_rtcm_types(payload: bytes) -> dict[int, int]: + counts: dict[int, int] = {} + i = 0 + + while i + 6 <= len(payload): + if payload[i] != 0xD3: + i += 1 + continue + + length = ((payload[i + 1] & 0x03) << 8) | payload[i + 2] + frame_end = i + 3 + length + 3 + if frame_end > len(payload): + break + + message_type = ((payload[i + 3] << 4) | (payload[i + 4] >> 4)) & 0x0FFF + counts[message_type] = counts.get(message_type, 0) + 1 + i = frame_end + + return counts def parse_mixed_payload(payload: bytes) -> list[dict[str, Any]]: diff --git a/backend/app/mqtt/parsers/rtcm.py b/backend/app/mqtt/parsers/rtcm.py deleted file mode 100644 index 839b259..0000000 --- a/backend/app/mqtt/parsers/rtcm.py +++ /dev/null @@ -1,22 +0,0 @@ -from __future__ import annotations - - -def summarize_rtcm_types(payload: bytes) -> dict[int, int]: - counts: dict[int, int] = {} - i = 0 - - while i + 6 <= len(payload): - if payload[i] != 0xD3: - i += 1 - continue - - length = ((payload[i + 1] & 0x03) << 8) | payload[i + 2] - frame_end = i + 3 + length + 3 - if frame_end > len(payload): - break - - message_type = ((payload[i + 3] << 4) | (payload[i + 4] >> 4)) & 0x0FFF - counts[message_type] = counts.get(message_type, 0) + 1 - i = frame_end - - return counts diff --git a/backend/app/mqtt/repository.py b/backend/app/mqtt/repository.py index 14e2d51..9bd1f31 100644 --- a/backend/app/mqtt/repository.py +++ b/backend/app/mqtt/repository.py @@ -9,7 +9,6 @@ from sqlmodel import Session, col, select from app.models import ( TerminalDeviceState, TerminalMessageRaw, - TerminalObservationEvent, get_datetime_utc, ) @@ -34,7 +33,6 @@ def insert_raw_message( payload_sha256=hashlib.sha256(payload).hexdigest(), qos=qos, retain=retain, - parse_status="pending", created_at=ts, ) session.add(row) @@ -48,7 +46,8 @@ def fetch_pending_rows_for_update( ) -> list[TerminalMessageRaw]: statement = ( select(TerminalMessageRaw) - .where(TerminalMessageRaw.parse_status == "pending") + .where(col(TerminalMessageRaw.parsed_at).is_(None)) + .where(col(TerminalMessageRaw.parse_error).is_(None)) .order_by(col(TerminalMessageRaw.received_at)) .limit(batch_size) ) @@ -61,17 +60,9 @@ def fetch_pending_rows_for_update( return list(rows) -def mark_processing(*, session: Session, row: TerminalMessageRaw) -> None: - row.parse_status = "processing" - row.parse_attempts += 1 - row.parse_error = None - session.add(row) - - def mark_succeeded(*, session: Session, row: TerminalMessageRaw) -> None: - row.parse_status = "succeeded" + row.parsed_at = get_datetime_utc() row.parse_error = None - row.next_retry_at = None session.add(row) @@ -80,15 +71,13 @@ def mark_failed( session: Session, row: TerminalMessageRaw, parse_error: str, - next_retry_at: datetime | None, ) -> None: - row.parse_status = "failed" + row.parsed_at = get_datetime_utc() row.parse_error = parse_error[:1024] - row.next_retry_at = next_retry_at session.add(row) -def write_events_and_update_state( +def update_device_state_from_events( *, session: Session, row: TerminalMessageRaw, @@ -110,14 +99,6 @@ def write_events_and_update_state( for event in events: payload = event.get("payload", {}) event_type = str(event.get("event_type", "unknown")) - db_event = TerminalObservationEvent( - raw_id=row.id, - device_id=row.device_id, - event_time=row.received_at, - event_type=event_type, - payload_json=payload if isinstance(payload, dict) else {"value": payload}, - ) - session.add(db_event) if event_type == "nmea_gga": state.lat = payload.get("lat") diff --git a/backend/app/mqtt/topic.py b/backend/app/mqtt/topic.py deleted file mode 100644 index e9127f8..0000000 --- a/backend/app/mqtt/topic.py +++ /dev/null @@ -1,10 +0,0 @@ -import re - -TOPIC_PATTERN = re.compile(r"^terminal/(?P[^/]+)/raw$") - - -def extract_device_id(topic: str) -> str: - match = TOPIC_PATTERN.match(topic) - if match is None: - raise ValueError("topic must match terminal/{device_id}/raw") - return match.group("device_id") diff --git a/backend/tests/mqtt/test_models.py b/backend/tests/mqtt/test_models.py index 5eadb7d..a90e2d3 100644 --- a/backend/tests/mqtt/test_models.py +++ b/backend/tests/mqtt/test_models.py @@ -18,4 +18,5 @@ def test_terminal_message_raw_model_can_persist() -> None: session.refresh(row) assert row.id is not None - assert row.parse_status == "pending" + assert row.parsed_at is None + assert row.parse_error is None diff --git a/backend/tests/mqtt/test_parser_worker.py b/backend/tests/mqtt/test_parser_worker.py index 836304a..9852ed0 100644 --- a/backend/tests/mqtt/test_parser_worker.py +++ b/backend/tests/mqtt/test_parser_worker.py @@ -1,6 +1,6 @@ -from sqlmodel import Session, SQLModel, create_engine, select +from sqlmodel import Session, SQLModel, create_engine -from app.models import TerminalDeviceState, TerminalMessageRaw, TerminalObservationEvent +from app.models import TerminalDeviceState, TerminalMessageRaw from app.mqtt.parser_worker import process_pending_batch from app.mqtt.repository import insert_raw_message @@ -11,7 +11,7 @@ def _session() -> Session: return Session(engine) -def test_insert_raw_message_sets_pending_status() -> None: +def test_insert_raw_message_sets_unparsed_defaults() -> None: with _session() as session: row = insert_raw_message( session=session, @@ -22,12 +22,13 @@ def test_insert_raw_message_sets_pending_status() -> None: retain=False, ) - assert row.parse_status == "pending" + assert row.parsed_at is None + assert row.parse_error is None assert row.payload_size > 0 assert len(row.payload_sha256) == 64 -def test_worker_marks_success_and_writes_events() -> None: +def test_worker_marks_success_and_updates_state() -> None: with _session() as session: raw = insert_raw_message( session=session, @@ -44,10 +45,43 @@ def test_worker_marks_success_and_writes_events() -> None: refreshed = session.get(TerminalMessageRaw, raw.id) assert refreshed is not None - assert refreshed.parse_status == "succeeded" - - events = session.exec(select(TerminalObservationEvent)).all() - assert len(events) >= 1 + assert refreshed.parsed_at is not None + assert refreshed.parse_error is None state = session.get(TerminalDeviceState, "device-001") assert state is not None + assert state.lat is not None + assert state.lon is not None + + +def test_worker_marks_failed_without_retry_schedule(monkeypatch) -> None: + def _raise_parse_error(_payload: bytes) -> list[dict[str, object]]: + raise ValueError("boom") + + monkeypatch.setattr( + "app.mqtt.parser_worker.parse_mixed_payload", _raise_parse_error + ) + + with _session() as session: + raw = insert_raw_message( + session=session, + device_id="device-err", + topic="terminal/device-err/raw", + payload=b"$GPGGA,broken*00\r\n", + qos=1, + retain=False, + ) + + processed = process_pending_batch(session=session, batch_size=10) + + assert processed == 1 + refreshed = session.get(TerminalMessageRaw, raw.id) + assert refreshed is not None + assert refreshed.parsed_at is not None + assert refreshed.parse_error == "boom" + processed_again = process_pending_batch(session=session, batch_size=10) + assert processed_again == 0 + + +def test_schema_excludes_terminal_observation_event_table() -> None: + assert "terminal_observation_event" not in SQLModel.metadata.tables diff --git a/backend/tests/mqtt/test_topic.py b/backend/tests/mqtt/test_topic.py index 61a938e..702734d 100644 --- a/backend/tests/mqtt/test_topic.py +++ b/backend/tests/mqtt/test_topic.py @@ -1,6 +1,6 @@ import pytest -from app.mqtt.topic import extract_device_id +from app.mqtt.ingestor import extract_device_id def test_extract_device_id_success() -> None: