> ## Documentation Index
> Fetch the complete documentation index at: https://docs.ilyama.golain.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Downlink control messages

> pause_lineage, resume_lineage, url_grant, and state_write on sync/ingest/control.

The cloud sends **control messages** to devices on:

```
{root_topic}/sync/ingest/control
```

Devices **subscribe** at QoS 1; **never publish** to this topic.

## 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.

This replaces the v1 path where the integration worker owned downlink staging. Wire envelope shape is unchanged — see [ilyama edge-client-guide §6](https://github.com/golain-io/ilyama/blob/main/docs/knowledge/edge-client-guide.md#6-downlink-control-messages--syncingestcontrol).

<Note>
  `sync/ingest/ack` and `sync/ingest/request` are **device → cloud** topics. They ingress on `edge_ingress_exchange` and are consumed by the edge worker — not delivered as MQTT downlinks.
</Note>

## Shared envelope

```json theme={null}
{
  "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": []
}
```

| Field                      | pause | resume | url\_grant | state\_write |
| -------------------------- | ----- | ------ | ---------- | ------------ |
| `control_id`               | ✓     | ✓      | ✓          | ✓            |
| `control_kind`             | ✓     | ✓      | ✓          | ✓            |
| `source_table`             | ✓     | ✓      | —          | ✓            |
| `server_watermark`         | ✓     | ✓      | —          | —            |
| `reason`                   | ✓     | —      | —          | —            |
| `presigned_url`            | —     | —      | ✓          | —            |
| `presigned_url_expires_at` | —     | —      | ✓          | —            |
| `rows`                     | —     | —      | —          | ✓            |

## Required device behavior (all kinds)

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

Unacked controls remain in `edge_pending_controls` until ACK received.

## pause\_lineage

**When sent:**

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

**Not sent for:**

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

**Device actions:**

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

**When sent:**

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

**Device actions:**

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

Secondary flow for large backfill or spill objects.

**Device actions:**

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`.

Object max size: **1 GiB**.

## state\_write

Cloud-originated desired state for **cloud\_authoritative** tables (Pipeline B).

**When sent:**

* 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

**Payload:**

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

**Device actions:**

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`.

→ [HTTP API — state write-back](/edge/data-sync/api-reference#cloud-state-write-back)

## 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)             |                            |
```

For `url_grant`, the device also publishes `sync/ingest/request` (finalize) after the HTTP PUT.

## Omega implementation

Module `sqlite-replication` subscribes to the control topic, handles pause/resume in `state/store.go`, ACKs via `sync/ingest/ack`, applies `state_write` rows into the local SQLite database, and records `cloud_version` so the next row-batch upload includes `lkcv` for the touched primary keys.

Verify in logs:

```
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](/edge/data-sync/backpressure)
* [Topics and connection](/edge/data-sync/topics-and-connection)
* [Payload formats](/edge/data-sync/payload-formats)