Building an At-Least-Once Faust Agent for MQTT Sensor Streams

An at-least-once faust agent guarantees that no bridged MQTT reading is lost, at the price of occasionally processing one twice — a trade almost always correct for environmental telemetry, where a missed exceedance is a compliance failure but a duplicated reading is a non-event once the sink deduplicates. The recipe is precise: consume the Kafka topic fed by your MQTT broker integration for environmental sensors, do the durable write before the offset commits, and upsert on the natural (device_id, sensor_ts) key so a replay overwrites in place. This page builds that agent end to end and explains exactly where crash recovery does and does not protect you.

Why At-Least-Once Plus Idempotency Beats Chasing Exactly-Once

The word “exactly-once” promises what everyone wants, but in a system that terminates in a database it is a category error. faust-streaming’s exactly-once mode wraps its Kafka produces and offset commit in a single transaction, so a downstream Kafka topic never sees a duplicate. Your PostGIS sink is not a Kafka topic; the INSERT is a side effect that executes outside the transaction boundary. If the process dies after the insert commits to the database but before the Kafka transaction commits, the transaction rolls back, the offset is not advanced, and on restart the same reading is inserted again. Exactly-once did not save you — the database still saw the row twice.

The durable fix lives at the sink, not in the delivery semantics. If the write is idempotent — an upsert keyed on something intrinsic to the measurement — then it does not matter how many times an event is delivered; the final table state is identical. This is often called effectively-once, and it is achieved with cheap at-least-once delivery plus a well-chosen conflict key. You get lower latency, no transactional overhead, and a system that is correct under crash, redelivery, and rebalance alike. The processing-guarantee trade-offs are laid out in full in the parent guide, Faust and Kafka consumer patterns for sensor streams.

The ordering that makes it work is the whole point: process, write durably, then commit the offset. faust’s default at-least-once flow already does exactly this — the offset for an event is only eligible to commit after the agent’s body returns. Your job is to make sure the sink write inside that body is both durable and idempotent.

It helps to name the three delivery regimes precisely, because the failure they each admit is different. At-most-once commits the offset before processing, so a crash in the gap silently drops the reading — unacceptable when the dropped packet might be a threshold exceedance. At-least-once commits after processing, so a crash replays the reading — harmless once the sink deduplicates. Exactly-once attempts to make delivery itself singular, but as shown above it cannot reach across the boundary into an external database. Only one of these three is both loss-free and cheap for a pipeline that ends in PostGIS, and it is at-least-once paired with an idempotent write.

Production-Ready Implementation

The agent below consumes bridged MQTT telemetry from Kafka, validates each record, and upserts it into a PostGIS readings table using an async connection pool. The offset commit is left to faust’s post-processing default, so it lands only after upsert_reading awaits successfully. Because the write is an ON CONFLICT upsert on (device_id, sensor_ts), a replayed event after a crash overwrites its own row rather than creating a duplicate.

# python 3.11 · faust-streaming==0.11.1 · confluent-kafka==2.3.0 · pydantic==2.6.4
from __future__ import annotations
import faust
import asyncpg
from pydantic import BaseModel, ValidationError, field_validator


class MqttReading(faust.Record, serializer="json"):
    """Wire shape emitted by the MQTT-to-Kafka bridge."""
    device_id: str
    sensor_ts: int          # Unix ms from the device clock
    lat: float
    lon: float
    pm25: float | None = None


class ValidReading(BaseModel):
    device_id: str
    sensor_ts: int
    lat: float
    lon: float
    pm25: float | None = None

    @field_validator("lat")
    @classmethod
    def _lat(cls, v: float) -> float:
        if not -90.0 <= v <= 90.0:
            raise ValueError(f"latitude {v} out of range")
        return v

    @field_validator("lon")
    @classmethod
    def _lon(cls, v: float) -> float:
        if not -180.0 <= v <= 180.0:
            raise ValueError(f"longitude {v} out of range")
        return v


app = faust.App(
    "mqtt-readings-consumer",           # stable Kafka consumer group id
    broker="kafka://broker:9092",
    processing_guarantee="at_least_once",
    broker_commit_every=500,            # commit cadence — see tuning table
    stream_buffer_maxsize=4096,
)

# Topic populated by the MQTT-to-Kafka bridge, keyed by device_id.
mqtt_topic = app.topic("env.mqtt.telemetry", value_type=MqttReading)
quarantine = app.topic("env.mqtt.quarantine", value_type=MqttReading)

_pool: asyncpg.Pool | None = None


@app.on_start
async def _open_pool() -> None:
    """Open the async DB pool once, on the worker's event loop."""
    global _pool
    _pool = await asyncpg.create_pool(
        dsn="postgresql://iot:secret@postgis:5432/telemetry",
        min_size=2,
        max_size=8,
    )


async def upsert_reading(reading: ValidReading) -> None:
    """
    Idempotent write: ON CONFLICT on the natural (device_id, sensor_ts) key
    means a replayed event overwrites its own row instead of duplicating it.
    This is what makes at-least-once delivery safe at the sink.
    """
    assert _pool is not None
    await _pool.execute(
        """
        INSERT INTO readings (device_id, sensor_ts, geom, pm25)
        VALUES ($1, $2, ST_SetSRID(ST_MakePoint($3, $4), 4326), $5)
        ON CONFLICT (device_id, sensor_ts)
        DO UPDATE SET pm25 = EXCLUDED.pm25, geom = EXCLUDED.geom
        """,
        reading.device_id,
        reading.sensor_ts,
        reading.lon,
        reading.lat,
        reading.pm25,
    )


@app.agent(mqtt_topic)
async def consume_mqtt(stream: faust.Stream[MqttReading]) -> None:
    """
    At-least-once agent: validate, write durably, then let faust commit.

    Crash recovery: if the worker dies after upsert but before the offset
    commit, faust replays from the last committed offset on restart. The
    idempotent upsert absorbs the replay with no duplicate rows.
    """
    async for msg in stream:
        try:
            reading = ValidReading(
                device_id=msg.device_id,
                sensor_ts=msg.sensor_ts,
                lat=msg.lat,
                lon=msg.lon,
                pm25=msg.pm25,
            )
        except ValidationError:
            # Never raise out of the loop — a crash forces a group rebalance.
            await quarantine.send(value=msg)
            continue

        # Durable write happens BEFORE the offset is eligible to commit.
        await upsert_reading(reading)

The design decisions worth calling out: the asyncpg pool is opened in @app.on_start so it binds to the worker’s own event loop and never blocks the poll loop; validation failures are quarantined rather than raised, because an uncaught exception crashes the agent and forces a rebalance across the whole consumer group; and the upsert conflict target is the measurement’s intrinsic identity, not any Kafka-assigned value.

If you want to shrink the replay window further, switch the stream to manual acking and commit right after the write:

@app.agent(mqtt_topic)
async def consume_mqtt_manual(stream: faust.Stream[MqttReading]) -> None:
    # no_auto_ack lets us acknowledge only after the durable write lands.
    async for event in stream.events():
        msg: MqttReading = event.value
        reading = ValidReading(**msg.asdict())
        await upsert_reading(reading)
        await event.ack()          # commit-eligible only now

Manual acking tightens the interval between durable write and commit, so a crash replays fewer events. For an already-idempotent sink the extra control rarely changes correctness — it only reduces redundant reprocessing.

Parameter Tuning Guide

Commit cadence is the main lever: broker_commit_every sets how many messages faust processes per partition before flushing an offset commit. A larger value amortizes commit overhead but widens the replay window on a crash. Size it against how expensive reprocessing is for each sensor type.

Sensor type Typical rate broker_commit_every stream_buffer_maxsize Notes
PM2.5 (optical) 1 / min 500 4096 Cheap upserts; wider commit batch is fine
Air temperature 1 / min 500 4096 Idempotent write is trivial; favor throughput
Dissolved oxygen 1 / 15 min 100 2048 Low rate; small batch keeps replay window tiny
Conductivity (EC) 1 / 15 min 100 2048 As above; commit often, little cost
High-rate acoustic 10 / s 2000 8192 Commit overhead dominates; large batch essential

Rule of thumb: set broker_commit_every so the worst-case replay (a full uncommitted batch per partition) reprocesses no more than a few seconds of data. For a 1/min PM2.5 sensor, even a batch of 500 is trivial to replay; for a 10 Hz acoustic feed, keep the batch large to avoid commit thrash but ensure the sink can absorb a 2000-event replay quickly.

Keep stream_buffer_maxsize proportional to sink throughput. If the buffer is larger than the sink can drain during a burst, memory grows before backpressure engages; if it is too small, throughput is capped below the sink’s capacity even when the sink is idle.

Verification and Testing

The property that must hold is idempotency under replay: processing the same event twice leaves the sink identical to processing it once. Test it directly by upserting the same reading twice and asserting a single row.

import pytest
import asyncpg


@pytest.mark.asyncio
async def test_upsert_is_idempotent(pg_pool: asyncpg.Pool):
    reading = ValidReading(
        device_id="aq-node-07",
        sensor_ts=1_720_000_000_000,
        lat=52.37,
        lon=4.90,
        pm25=18.4,
    )

    # Simulate an at-least-once replay: same event delivered twice.
    await upsert_reading(reading)
    await upsert_reading(reading)

    rows = await pg_pool.fetch(
        "SELECT pm25 FROM readings WHERE device_id = $1 AND sensor_ts = $2",
        reading.device_id, reading.sensor_ts,
    )
    assert len(rows) == 1, "replayed event must not duplicate the row"
    assert rows[0]["pm25"] == pytest.approx(18.4)

For an end-to-end crash test, run the agent against a test broker, publish a batch, SIGKILL the worker mid-batch, restart it, and confirm the sink row count equals the number of distinct (device_id, sensor_ts) pairs published — not the number of messages delivered. Any excess means the conflict key is wrong.

Gotchas

A non-unique conflict key silently duplicates. If two physical measurements from the same device can share a sensor_ts (for example, a device that resets its clock to zero on reboot), the upsert collapses distinct readings into one. Add a monotonic sequence or reboot counter to the key when device clocks are unreliable.

Blocking DB drivers stall the heartbeat. Using synchronous psycopg2 directly inside the agent blocks the shared event loop; heartbeats miss, the broker evicts the instance, and a rebalance fires mid-batch. Use asyncpg, or wrap blocking calls in loop.run_in_executor. This is the same event-loop discipline that keeps rebalances rare, covered in managing consumer group rebalancing for sensor topics.

Committing before the write inverts the guarantee. If you ever restructure the agent so the offset commits before upsert_reading completes, you have built at-most-once: a crash in that gap drops the reading permanently. The ordering process-write-then-commit is load-bearing; never commit optimistically.

Quarantine backpressure hides schema drift. A firmware change that renames a payload field sends every record to the quarantine topic. If nothing alarms on quarantine volume, the sink simply goes quiet and the drift looks like a dead sensor. Alert on quarantine rate, not just on sink write errors.