Building an Ingest Safety Monitor for Your IIoT Pipeline in SQL
Ingest fails suddenly, not gradually. Sample WAL volume, replica lag, and dead tuples every minute, and get one headroom percentage you can alert on.
By John Shi
October 16th, 2026
J
By John Shi
October 21st, 2026
9 min
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.
readings
hypertable that skips readings it already holds.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.
site_id
.CREATEROLE
.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.
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 partitioned on device_ts
.
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'));
A unique constraint makes PostgreSQL refuse a second row with the same identity, and ON CONFLICT DO NOTHING
turns each refusal into a skipped row. On a hypertable, the constraint must include device_ts
. The SQL below adds the constraint.
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.
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.
INSERT 0 12000
INSERT 0 0
INSERT 0 0
In SQLite's write-ahead log (WAL) mode, the sender reads while the collector writes. Setting synchronous = FULL 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.
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.
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.
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.
DELETE FROM buffer
WHERE seq <= (SELECT acked_seq FROM hwm)
AND device_ts < strftime('%Y-%m-%dT%H:%M:%f',
'now', '-1 day');
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.
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
is on, so Step 6 sets it on the replay role.
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.
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.
A replay role 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.
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
has no ON CONFLICT
clause, so each batch goes into a temporary table 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
.
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.
python3 replay.py s-017.db \
"postgresql://replay:change-me@HOST:PORT/tsdb?sslmode=require"
Before the first outage, set the 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.
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 already in the columnstore that the replay would reach.
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 back to check the constraint, at more cost per row.
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.
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.
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.
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.
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.
bucket | n
------------------------+--------
2026-09-28 21:00:00+00 | 720000
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.
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.
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.
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
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.
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.
rows_sent | rows_stored
-----------+-------------
7560000 | 2520000
The gateway query below counts the same window, which should match rows_stored
.
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.
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's drop_after
.
SELECT config ->> 'drop_after' AS drop_after
FROM timescaledb_information.jobs
WHERE proc_name = 'policy_retention'
AND hypertable_name = 'readings';
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 today.
About the author
J
By John Shi
// Related posts
Ingest fails suddenly, not gradually. Sample WAL volume, replica lag, and dead tuples every minute, and get one headroom percentage you can alert on.
By John Shi
October 16th, 2026
Stop waiting for the dashboard to get slow. Record latency against row count, fit the curve, and get a date when your query breaches its target.
By John Shi
October 9th, 2026
Five refresh policy shapes for solar and wind SCADA: frequent, daily, hierarchical, retention-aware, and manual. Pick the one that fits each job.
By Damaso Sanoja
October 7th, 2026
Receive the latest technical articles and release notes in your inbox.