Downlink control messages

{root_topic}/sync/ingest/control

​

How controls reach the device (edge v2)

1. The edge worker commits a downlink intent in the transactional outbox.
2. It stages EdgeIngestControlRequestedV2 on the MQTT downlink exchange (edge.ingest_control.requested.v1, domain edge).
3. The mqtt-broker consumes from mqtt_broker_downlink_queue and publishes the JSON body to the device topic at QoS 1.

​

Shared envelope

{
  "control_id": "server-assigned-uuid",
  "control_kind": "pause_lineage",
  "source_table": "device_state",
  "server_watermark": 103,
  "reason": "backpressure: staged row cap exceeded",
  "presigned_url": "https://...",
  "presigned_url_expires_at": "2026-04-28T09:15:00Z",
  "rows": []
}

​

Required device behavior (all kinds)

1. ACK immediately on sync/ingest/ack with same control_id.
2. Then execute kind-specific logic.

​

pause_lineage

• Volume backpressure (staged rows exceed cap)
• Materialization circuit breaker

• Schema review pause alone (server stages while device keeps sending)

1. Stop publishing batches for source_table.
2. Buffer new journal rows durably (Omega: state.db + buffer store).
3. Record server_watermark — last accepted commit_seq (rows) or batch_seq (telemetry).

​

resume_lineage

• Backlog drained below resume threshold (auto sweep)
• After reset-materialization when device had been pause-downlinked

1. Drain buffer in commit_seq / batch_seq order starting at server_watermark + 1.
2. Resume normal 5s flush loop.
3. If local buffer was lost during pause → request bootstrap snapshot (secondary flow, when available).

​

url_grant

1. ACK control.
2. HTTP PUT zstd-compressed JSONL (.jsonl.zst) to presigned_url (15 min TTL).
3. Publish finalize on sync/ingest/request with matching batch_id and segment_kind.

​

state_write

• Operator calls POST .../devices/{device_id}/edge/state-tables/{table_id}/rows
• Edge worker upserts rows in the cloud state table, then stages a downlink

{
  "control_id": "…",
  "control_kind": "state_write",
  "source_table": "device_config",
  "rows": [
    {
      "pk": { "id": 1 },
      "values": { "mode": "cooling", "setpoint": 21 },
      "cloud_version": 3
    }
  ]
}

1. ACK on sync/ingest/ack.
2. Apply rows to the local SQLite table for the matching source_table.
3. Persist the supplied cloud_version per primary key so the next sync/rows/batch upload for those rows carries lkcv.

​

Protocol diagram

Edge worker                          mqtt-broker                    Device
  |                                       |                            |
  |-- EdgeIngestControlRequestedV2 ------> |                            |
  |   (mqtt_downlink exchange)            |                            |
  |                                       |-- sync/ingest/control ---->|
  |                                       |                            | ACK ingest/ack
  |<-- edge.ingest_ack.received.v2 -------|<-- sync/ingest/ack --------|
  |   (edge_ingress_exchange)             |                            |

​

Omega implementation

pause_lineage source_table=device_state watermark=103
resume_lineage source_table=device_state watermark=150
state_write source_table=device_config rows=1

​

Related

• Backpressure
• Topics and connection
• Payload formats