--- title: "Designing IIoT Ingest That Survives a Cut Uplink" published: 2026-10-21T08:08:45.000-04:00 updated: 2026-10-21T08:08:45.000-04:00 excerpt: "IIoT ingest that survives a cut uplink: enforce each reading’s identity, buffer durably at the edge, and cap the replay so catch-up can’t starve live sites." tags: IIoT, PostgreSQL authors: John Shi --- > **TimescaleDB is now Tiger Data.** A remote Industrial Internet of Things (IIoT) site can lose its uplink for hours while its sensors keep reading. When the link returns, every buffered row has to land in PostgreSQL exactly once, while the sites that stayed online keep writing at close to their normal rate. In this guide, you'll learn how to build that reconnect path. The resulting database will enforce an identity on every reading, and the gateway keeps each row until the server commits it, so a retry never creates a duplicate. Catch-up traffic will also get its own capped path. ## What you will build - A `readings` [hypertable](https://www.tigerdata.com/docs/reference/timescaledb/hypertables) that skips readings it already holds. - A gateway buffer whose high-water mark moves after each commit. - A rate-capped Python replay loop with a staging table. - A scoped [continuous aggregate](https://www.tigerdata.com/docs/reference/timescaledb/continuous-aggregates) refresh and a per-site lag query. ## Why it matters When a connection drops after the server commits a batch and before the acknowledgement reaches the gateway, the gateway sends the batch again. Without an enforced identity, that copy doubles every count and sum over the batch's time range. A replay also takes capacity from the sites that stayed online. A site sending 500 readings per second that loses its uplink for six hours returns with 10.8 million rows. At 20 times its normal rate it clears them in about 19 minutes, writing as much as 20 sites its size out of the ingest headroom in the [IIoT PostgreSQL performance envelope](https://www.tigerdata.com/blog/the-iiot-postgresql-performance-envelope). ## Prerequisites - TimescaleDB 2.18 or later, self-hosted or on [Tiger Cloud](https://www.tigerdata.com/cloud). - A sites table with one row per `site_id`. - A role with `CREATEROLE`. - Python 3 with psycopg 3 on the gateway. - A staging site whose uplink you can disconnect. ## Build the reconnect path ### Step 1: Assign each reading an identity at the edge A reading's identity is its site, device, channel, and device timestamp. The gateway stamps `device_ts` once per poll and resends that exact value on every retry. The server's `received_at` changes with each attempt, so it stays out of the identity. The Python below is the gateway's collector, which writes each poll to the buffer from Step 3 in one transaction. ```Python from datetime import datetime, timezone def record(edge, site_id, readings): # one poll: (device_id, channel, value) per reading ts = datetime.now(timezone.utc).isoformat( timespec="microseconds") edge.executemany( "INSERT INTO buffer (site_id, device_id, channel," " device_ts, value) VALUES (?, ?, ?, ?, ?)", [(site_id, d, c, ts, v) for d, c, v in readings]) edge.commit() ``` A sensor with its own clock should supply `device_ts` itself. The SQL below creates the server table as a [hypertable](https://www.tigerdata.com/docs/reference/timescaledb/hypertables/create_hypertable) partitioned on `device_ts`. ```SQL CREATE TABLE readings ( site_id text NOT NULL, device_id integer NOT NULL, channel smallint NOT NULL, device_ts timestamptz NOT NULL, value double precision, received_at timestamptz NOT NULL DEFAULT now() ); SELECT create_hypertable('readings', by_range('device_ts')); ``` ### Step 2: Enforce the identity with an idempotent insert A [unique constraint](https://www.postgresql.org/docs/current/ddl-constraints.html#DDL-CONSTRAINTS-UNIQUE-CONSTRAINTS) makes PostgreSQL refuse a second row with the same identity, and [`ON CONFLICT DO NOTHING`](https://www.postgresql.org/docs/current/sql-insert.html#SQL-ON-CONFLICT) turns each refusal into a skipped row. On a hypertable, the constraint must include `device_ts`. The SQL below adds the constraint. ```SQL ALTER TABLE readings ADD CONSTRAINT readings_identity UNIQUE (site_id, device_ts, device_id, channel); ``` The insert that follows writes one minute of readings for one site. ```SQL INSERT INTO readings (site_id, device_id, channel, device_ts, value) SELECT 's-001', d, c, ts, 20 + d * 0.1 + c FROM generate_series(1, 50) d, generate_series(1, 4) c, generate_series(timestamptz '2026-09-01 06:00:00+00', timestamptz '2026-09-01 06:00:59+00', interval '1 second') ts ON CONFLICT DO NOTHING; ``` Every output in this guide comes from a synthetic dataset. The PostgreSQL outputs came from a plain table on version 16.13 on a two-core x86\_64 container, with `shared_buffers` at 256 megabytes and fsync, `synchronous_commit`, and parallel workers off, which affects throughput but not the idempotency results. The Step 8 outputs ran on Tiger Cloud with PostgreSQL 18.6 on aarch64 and TimescaleDB 2.29.2. Your numbers will differ, and the method applies unchanged. Three runs of the insert produced the output below. ```markdown INSERT 0 12000 INSERT 0 0 INSERT 0 0 ``` ### Step 3: Buffer durably on the edge In SQLite's [write-ahead log (WAL) mode](https://www.sqlite.org/wal.html), the sender reads while the collector writes. Setting [synchronous = FULL](https://www.sqlite.org/pragma.html#pragma_synchronous) flushes each commit to disk before it returns, so the buffer survives a power cut. `AUTOINCREMENT` keeps `seq` from reusing values after pruning. The SQL below creates the buffer and a one-row high-water mark table. ```SQL PRAGMA journal_mode = WAL; PRAGMA synchronous = FULL; CREATE TABLE buffer ( seq INTEGER PRIMARY KEY AUTOINCREMENT, site_id TEXT NOT NULL, device_id INTEGER NOT NULL, channel INTEGER NOT NULL, device_ts TEXT NOT NULL, value REAL ); CREATE TABLE hwm ( id INTEGER PRIMARY KEY CHECK (id = 1), acked_seq INTEGER NOT NULL ); INSERT INTO hwm VALUES (1, 0); ``` The query that follows reports bytes per row for a filled buffer. ```SQL SELECT count(*) AS buffered_rows, (SELECT page_count * page_size FROM pragma_page_count(), pragma_page_size()) / count(*) AS bytes_per_row FROM buffer; ``` Filling the buffer with three and a half hours of generated readings from one site, s-017, at 200 readings per second gave the output below. ```markdown buffered_rows bytes_per_row ------------- ------------- 2520000 59 ``` At that width, 48 hours of unacknowledged readings at 200 per second take about 2 gigabytes, plus about 1 gigabyte per day of acknowledged readings kept for the validation count. The SQLite statement below prunes acknowledged rows older than a day. ```SQL DELETE FROM buffer WHERE seq <= (SELECT acked_seq FROM hwm) AND device_ts < strftime('%Y-%m-%dT%H:%M:%f', 'now', '-1 day'); ``` ### Step 4: Advance the high-water mark only after the server commits The high-water mark is the highest seq the server has committed. If the connection drops after the commit and before the gateway records the new mark, the gateway resends the batch and the constraint from Step 2 skips it. Bind `:last_seq` to the highest `seq` in the batch just sent. The SQL below is the gateway's half of the handshake. ```SQL SELECT seq, site_id, device_id, channel, device_ts, value FROM buffer WHERE seq > (SELECT acked_seq FROM hwm) ORDER BY seq LIMIT 5000; UPDATE hwm SET acked_seq = :last_seq; ``` A returned `COMMIT` means the rows are on disk only while [`synchronous_commit`](https://www.postgresql.org/docs/current/runtime-config-wal.html#GUC-SYNCHRONOUS-COMMIT) is on, so Step 6 sets it on the replay role. ### Step 5: Replay in ordered batches under a per-site rate limit Each gateway runs its own loop, so `MAX_ROWS_PER_SEC` caps one site. The Python below, the first half of replay.py, sends batches in seq order through `send_batch()` from Step 6 and sleeps between them to stay under that cap. ```Python import sqlite3, sys, time import psycopg BATCH = 5000 MAX_ROWS_PER_SEC = 10000 COLS = "site_id, device_id, channel, device_ts, value" def replay(buffer_path, dsn): edge = sqlite3.connect(buffer_path) pg = psycopg.connect(dsn, autocommit=True) pg.execute(STAGE_DDL) while True: started = time.monotonic() rows = edge.execute( f"SELECT seq, {COLS} FROM buffer" " WHERE seq > (SELECT acked_seq FROM hwm)" " ORDER BY seq LIMIT ?", (BATCH,)).fetchall() if not rows: return send_batch(pg, rows) # COMMIT returned, so the server has the rows edge.execute("UPDATE hwm SET acked_seq = ?", (rows[-1][0],)) edge.commit() budget = len(rows) / MAX_ROWS_PER_SEC time.sleep(max(0, budget - (time.monotonic() - started))) ``` The loop returns at the end of the buffer. The gateway's live sender then takes over, reading the same buffer and moving the same mark under your live ingest role, so run only one sender at a time. ### Step 6: Give catch-up traffic its own role and staging table A [replay role](https://www.postgresql.org/docs/current/sql-createrole.html) limited to two connections lets two sites catch up at once. Gateways should retry a refused connection. Replace `change-me` with a real password in the SQL below, which creates the role and a batch log. ```SQL CREATE TABLE backfill_log ( site_id text NOT NULL, ts_from timestamptz NOT NULL, ts_to timestamptz NOT NULL, rows_sent bigint NOT NULL, logged_at timestamptz NOT NULL DEFAULT now() ); CREATE ROLE replay LOGIN PASSWORD 'change-me' CONNECTION LIMIT 2; ALTER ROLE replay SET synchronous_commit = on; GRANT INSERT ON readings, backfill_log TO replay; ``` [`COPY`](https://www.postgresql.org/docs/current/sql-copy.html) has no `ON CONFLICT` clause, so each batch goes into a [temporary table](https://www.postgresql.org/docs/current/sql-createtable.html) through `COPY` first. One `INSERT` with `ON CONFLICT DO NOTHING` then moves it into `readings` in the constraint's column order, and a second logs its time range. The Python below completes `replay.py`. ```Python STAGE_DDL = f""" CREATE TEMP TABLE stage ( site_id text, device_id integer, channel smallint, device_ts timestamptz, value double precision ) ON COMMIT DELETE ROWS""" MOVE_SQL = f""" INSERT INTO readings ({COLS}) SELECT {COLS} FROM stage ORDER BY site_id, device_ts, device_id, channel ON CONFLICT DO NOTHING""" LOG_SQL = """ INSERT INTO backfill_log (site_id, ts_from, ts_to, rows_sent) SELECT site_id, min(device_ts), max(device_ts), count(*) FROM stage GROUP BY site_id""" def send_batch(pg, rows): with pg.transaction(): with pg.cursor().copy( f"COPY stage ({COLS}) FROM STDIN") as copy: for row in rows: copy.write_row(row[1:]) pg.execute(MOVE_SQL) pg.execute(LOG_SQL) if __name__ == "__main__": replay(sys.argv[1], sys.argv[2]) ``` Replace `s-017.db`, `HOST`, and `PORT` in the shell command below, which starts a replay. ```Bash python3 replay.py s-017.db \ "postgresql://replay:change-me@HOST:PORT/tsdb?sslmode=require" ``` ### Step 7: Decide how late rows enter the columnstore Before the first outage, set the [columnstore policy](https://www.tigerdata.com/docs/reference/timescaledb/hypercore/add_columnstore_policy)'s `after` interval longer than the longest outage the buffer holds plus the replay time. Every chunk a replay can reach then stays in the rowstore. The SQL below does that for a 48-hour buffer. ```SQL ALTER TABLE readings SET ( timescaledb.enable_columnstore = true, timescaledb.segmentby = 'site_id, device_id, channel', timescaledb.orderby = 'device_ts' ); CALL add_columnstore_policy('readings', after => INTERVAL '3 days'); ``` Before replaying, put the oldest unacknowledged `device_ts` in the query below, which lists [chunks](https://www.tigerdata.com/docs/reference/timescaledb/informational-views/chunks) already in the columnstore that the replay would reach. ```SQL SELECT chunk_name, range_start, range_end FROM timescaledb_information.chunks WHERE hypertable_name = 'readings' AND is_compressed AND range_end > '2026-09-28 21:30:00+00'; ``` An empty result means the replay lands in the rowstore. A replay into the columnstore also works, and TimescaleDB [converts the batches that could conflict](https://www.tigerdata.com/docs/build/tips-and-tricks/troubleshoot-hypercore) back to check the constraint, at more cost per row. ### Step 8: Refresh the continuous aggregate over the backfilled range If you have your own aggregate, put its name in the calls that follow. The SQL below creates one with one-hour buckets and a [refresh policy](https://www.tigerdata.com/docs/reference/timescaledb/continuous-aggregates/add_continuous_aggregate_policy). ```SQL CREATE MATERIALIZED VIEW readings_hourly WITH (timescaledb.continuous) AS SELECT site_id, time_bucket(INTERVAL '1 hour', device_ts) AS bucket, count(*) AS n, avg(value) AS avg_value FROM readings GROUP BY site_id, bucket WITH NO DATA; SELECT add_continuous_aggregate_policy('readings_hourly', start_offset => INTERVAL '3 hours', end_offset => INTERVAL '1 hour', schedule_interval => INTERVAL '10 minutes'); ``` This policy never returns to buckets older than three hours, so a replay that reaches further back leaves those buckets stale. The query below reads the bucket holding the start of the generated outage. ```SQL SELECT bucket, n FROM readings_hourly WHERE site_id = 's-017' AND bucket = '2026-09-28 21:00:00+00'; ``` After the replay and a policy run, that bucket kept its pre-outage count, shown in the output below. ```markdown bucket | n ------------------------+-------- 2026-09-28 21:00:00+00 | 360000 ``` Replace `s-017` with the site that reconnected. In psql, the SQL below reads the replayed range from `backfill_log`, widens it to whole hours, and passes it to [refresh\_continuous\_aggregate](https://www.tigerdata.com/docs/reference/timescaledb/continuous-aggregates/refresh_continuous_aggregate). ```SQL SELECT date_trunc('hour', min(ts_from)) AS lo, date_trunc('hour', max(ts_to)) + INTERVAL '1 hour' AS hi FROM backfill_log WHERE site_id = 's-017' AND logged_at > now() - INTERVAL '1 day' \gset CALL refresh_continuous_aggregate( 'readings_hourly', :'lo', :'hi'); ``` Rerunning the bucket query after that refresh returned the full hour, in the output below. ```markdown bucket | n ------------------------+-------- 2026-09-28 21:00:00+00 | 720000 ``` ### Step 9: Track per-site lag The `replay_lag` column is the gap between a site's newest device timestamp and the moment that row arrived, and `silence` is the time since then. The query below reads one row per site from the constraint's index, joined to the `sites` table. ```SQL SELECT s.site_id, date_trunc('second', r.received_at - r.device_ts) AS replay_lag, date_trunc('second', now() - r.received_at) AS silence FROM sites s LEFT JOIN LATERAL ( SELECT device_ts, received_at FROM readings WHERE readings.site_id = s.site_id ORDER BY device_ts DESC LIMIT 1 ) r ON true ORDER BY replay_lag DESC NULLS FIRST; ``` With s-017 offline, the query returned the output below. ```markdown site_id | replay_lag | silence ---------+------------+---------- s-017 | 00:00:01 | 03:35:53 s-002 | 00:00:00 | 00:00:00 s-003 | 00:00:00 | 00:00:01 ``` Sixty seconds into a replay capped at 10,000 rows per second, it returned the output below. ```markdown site_id | replay_lag | silence ---------+------------+---------- s-017 | 02:46:54 | 00:00:00 s-002 | 00:00:00 | 00:00:01 s-003 | 00:00:00 | 00:00:00 ``` ## Validate the reconnect path Replay the buffer three times, resetting `acked_seq` to `0` between runs. Then put your site and outage window into the query below, which compares rows sent with rows stored. ```SQL SELECT (SELECT sum(rows_sent) FROM backfill_log WHERE site_id = 's-017') AS rows_sent, (SELECT count(*) FROM readings WHERE site_id = 's-017' AND device_ts >= '2026-09-28 21:30:00+00' AND device_ts < '2026-09-29 01:00:00+00') AS rows_stored; ``` After three full replays of the generated buffer, the query returned the output below. ```markdown rows_sent | rows_stored -----------+------------- 7560000 | 2520000 ``` The gateway query below counts the same window, which should match `rows_stored`. ```SQL SELECT count(*) AS edge_rows FROM buffer WHERE device_ts >= '2026-09-28T21:30:00.000000+00:00' AND device_ts < '2026-09-29T01:00:00.000000+00:00'; ``` The table below shows live ingest from the 19 other generated sites during a replay, from two pgbench clients writing 200 rows per transaction, as the median of six 20-second runs. | Replay condition | Live rows/sec | Change | | --- | --- | --- | | None | 215,311 | Baseline | | Uncapped | 147,670 | -31% | | Capped at 10,000 rows/sec | 205,824 | -4% | Uncapped, the replay took almost a third of live ingest on this machine. The cap cut that to within the spread between runs. To check that no buffered reading aged past retention, the SQL below reads the gateway's oldest unacknowledged reading. ```SQL SELECT min(device_ts) AS oldest_unacked FROM buffer WHERE seq > (SELECT acked_seq FROM hwm); ``` Rows older than `now()` minus `drop_after` land in chunks the next retention run drops. The query below reads the [retention policy](https://www.tigerdata.com/docs/reference/timescaledb/data-retention/add_retention_policy)'s `drop_after`. ```SQL SELECT config ->> 'drop_after' AS drop_after FROM timescaledb_information.jobs WHERE proc_name = 'policy_retention' AND hypertable_name = 'readings'; ``` ## Conclusion A reconnect is safe to retry once the server enforces each reading's identity and the gateway forgets a row only after `COMMIT` returns. With that in place, a replay can run at whatever rate its cap allows, and one scoped refresh puts its rows into the aggregate. This week, cut the uplink to a staging site for an hour, reconnect it, and use the validation queries above to compare the gateway's buffered row count with what landed. If you need a TimescaleDB service for the test, [start a Tiger Cloud free trial](https://console.cloud.timescale.com/signup) today.