Ingestion Pipeline

The Ingestion service is a standalone Node.js process that continuously reads device telemetry from Redis Streams and persists it to TimescaleDB. It has no user-facing REST API — only internal health and metrics endpoints.

Why a Separate Service?

Separating ingestion from the API keeps two concerns independent:

The Cloud API stays fast — it receives MQTT messages and queues telemetry batches into Redis immediately, without waiting for the database write to complete.
The Ingestion service owns the write rate to TimescaleDB, applying batching, deduplication, and backpressure so the database is never overwhelmed during traffic spikes.
Multiple Ingestion pods can consume the same Redis consumer group for horizontal scale-out.

Data Flow

Agent
  │
  │  MQTT publish
  │  i/{tenant}/a/{agent}/endpoints/{protocol}
  ▼
MQTT Broker (Mosquitto)
  │
  │  MQTT subscribe
  ▼
Cloud API (MQTT subscriber)
  │  deserialize: DEFLATE → msgpack → JSON
  │  expand key-compaction dictionary (if enabled)
  │  deduplicate via Redis SETNX
  │
  ├──▶ Redis Pub/Sub    ← real-time state / metrics (dashboard)
  │
  └──▶ Redis Stream  tenant:{id}:agent:devices:ingestion
              │
              │  XREADGROUP (consumer group)
              ▼
       Ingestion Worker
              │
    ┌─────────┴──────────────┐
    │  1. Decode payload     │
    │  2. Normalize readings  │
    │  3. Deduplicate batch   │
    │  4. Bulk INSERT         │
    └─────────┬──────────────┘
              │
         TimescaleDB
       readings hypertable

Redis Streams

Stream keys

Key	Direction	Content
`tenant:{id}:agent:devices:ingestion`	Consumer reads	Device telemetry batches from MQTT
`tenant:{id}:agent:devices:dlq`	Worker writes on failure	Failed message batches
`tenant:{id}:agent:logs`	Consumer reads	Agent log batches
`tenant:{id}:device:{uuid}:metrics`	Pub/Sub (publish only)	Real-time metrics for WebSocket clients

Consumer group

The Ingestion service creates a consumer group {tenantId}:device-writers on startup if it does not already exist. Each worker in the group uses XREADGROUP to claim exclusive ownership of messages. Processed messages are acknowledged with XACK and trimmed from the stream. Unacknowledged messages (worker crash) are reclaimed via XAUTOCLAIM after a configurable timeout.

Processing Pipeline

1. Decode payload

The Cloud API MQTT handler has already deserialized and decoded the payload before writing it to Redis. The Ingestion worker receives a normalized JS object. Payloads from the MQTT path may have gone through:

DEFLATE decompression (outer wrapper)
MessagePack decoding (binary format)
Key-compaction dictionary expansion (integer indices → field names)

Unrecognized formats are moved to the DLQ rather than crashing the worker.

2. Normalization

The normalizer expands the device representation into individual readings rows. A single MQTT endpoints message typically carries readings from one sensor device with multiple metrics:

Input (from Redis — decoded MQTT endpoints payload):
{
  agentUuid: "abc-123",
  deviceName: "modbus-plc",
  protocol: "modbus",
  readings: [
    { name: "temperature", value: 72.4, unit: "°C", quality: "good" },
    { name: "pressure",    value: 1013,  unit: "hPa", quality: "good" }
  ]
}

Output (rows inserted into TimescaleDB):
{ time, agent_uuid, metric_name: "modbus-plc.temperature", value: 72.4, protocol: "modbus", ... }
{ time, agent_uuid, metric_name: "modbus-plc.pressure",    value: 1013,  protocol: "modbus", ... }

3. Deduplication

Within each batch, if the same (agent_uuid, metric_name, time) tuple appears more than once, the last value wins. Rows are then sorted by primary key before the database insert to prevent deadlocks when multiple workers operate concurrently.

At the MQTT layer, the Cloud API already deduplicates by msgId using Redis SETNX (24-hour TTL) to drop duplicate MQTT deliveries. The in-batch deduplication in the Ingestion worker is a second-pass safeguard for overlapping windows during reconnects.

4. Database Insert

Rows are inserted using a parameterized INSERT … ON CONFLICT DO NOTHING or PostgreSQL COPY (configurable via DB_INSERT_METHOD). The insert target is the readings hypertable — TimescaleDB routes each row to the correct 1-day chunk automatically.

After a successful batch, a background task updates last_telemetry_at on the relevant endpoint rows (fire-and-forget, non-blocking).

Resilience

Circuit Breaker

The circuit breaker monitors database insert failures. If errors exceed the threshold within a rolling window, it opens — workers stop attempting inserts and immediately route messages to the disk spool instead. The breaker polls the database connection in the background and closes automatically when the database recovers.

CLOSED (normal) ──errors exceed threshold──▶ OPEN (fault)
     ▲                                             │
     └────── DB healthy, spool drained ────────────┘

Disk Spool

When the circuit is open (or Redis itself is unreachable), the spool writes raw message payloads to local disk (/tmp/iotistic-spool by default, configurable). The spool is capped at 500 MB. When the circuit closes, a replayer drains the spool back into the normal insert pipeline in order.

Dead-Letter Queue

Messages that fail parsing or produce unrecoverable insert errors after MAX_RETRIES attempts are written to the DLQ stream (tenant:{id}:agent:devices:dlq). The DLQ can be inspected and replayed manually using iotctl or directly via redis-cli XRANGE.

Autoscaling

The Ingestion service auto-adjusts its worker count (1–20) based on four signals sampled every 5 seconds:

Signal	Scale up when	Scale down when
Consumer lag	> 5 000 ms behind real-time	< 500 ms
DB pool saturation	> 80 % of connections in use	< 40 %
Redis stream depth	> 60 % of `MAXLEN`	< 20 %
Redis memory pressure	> 75 % of `maxmemory`	< 50 %

Worker count changes are gradual (±1 per cycle) to avoid thundering-herd on the database connection pool.

Ingestion Profiles

The INGESTION_PROFILE environment variable selects a pre-tuned configuration bundle. Every value in a profile can be overridden by setting the individual environment variable directly — the profile is just the starting point.

Profile overview

Profile	Use case	Insert mode	Batch size	Flush interval	Workers (min / max)	Lag target
`batch`	Bulk / delay-tolerant, high throughput	COPY	500	2 000 ms	2 / 12	10 000 ms
`balanced`	General production fleet (default)	COPY	100	500 ms	4 / 16	5 000 ms
`streaming`	Near-real-time dashboards, low latency	realtime	20	100 ms	6 / 20	2 000 ms
`benchmark`	Load / stress testing only	COPY	1 000	200 ms	8 / 24	8 000 ms
`hp`	Large fleets, stable long-running	COPY	300	100 ms	8 / 20	5 000 ms

Insert modes:

copy — uses PostgreSQL COPY FROM via a binary stream. Fastest bulk insert, minimum DB overhead. Requires READINGS_COPY_MIN_ROWS to be met before switching from INSERT to COPY.
realtime — uses regular parameterized INSERT in small bursts. Lower per-row throughput but starts writing immediately without waiting for a large batch to accumulate.

`batch`

Optimized for maximum throughput at the cost of write latency. Accumulates large batches before flushing, keeps the connection pool small, and backs off aggressively on DB pressure. Best for fleets with bursty or asynchronous ingestion where a 2–15 second data delay is acceptable.

Parameter	Value
`WORKER_COUNT`	4
`BATCH_SIZE`	500
`FLUSH_INTERVAL_MS`	2 000
`READINGS_BULK_INSERT_MODE`	`copy`
`READINGS_COPY_MIN_ROWS`	1 000
`AUTOSCALE_MIN_WORKERS`	2
`AUTOSCALE_MAX_WORKERS`	12
`AUTOSCALE_LAG_TARGET_MS`	10 000
`AUTOSCALE_LAG_SCALE_UP_MS`	5 000
`AUTOSCALE_LAG_CRITICAL_MS`	15 000
`AUTOSCALE_SCALE_DOWN_STABLE_CHECKS`	12
`AUTOSCALE_COOLDOWN_MS`	5 000
`AUTOSCALE_DB_BLOCK_PCT`	80
`DB_POOL_SIZE`	20
`DB_SATURATION_HIGH_WATERMARK_PCT`	70
`DB_BACKPRESSURE_SLEEP_MS`	500
`REDIS_INGESTION_STREAM_MAXLEN`	10 000
`REDIS_PIPELINE_FLUSH_INTERVAL_MS`	50

`balanced`

The default profile. Balances write latency and throughput for a typical production fleet. COPY inserts with moderate batch sizes, a 500 ms flush interval, and a wide autoscaling range give it headroom to handle traffic spikes without over-committing DB connections at idle.

Parameter	Value
`WORKER_COUNT`	6
`BATCH_SIZE`	100
`FLUSH_INTERVAL_MS`	500
`READINGS_BULK_INSERT_MODE`	`copy`
`READINGS_COPY_MIN_ROWS`	1 000
`AUTOSCALE_MIN_WORKERS`	4
`AUTOSCALE_MAX_WORKERS`	16
`AUTOSCALE_LAG_TARGET_MS`	5 000
`AUTOSCALE_LAG_SCALE_UP_MS`	2 000
`AUTOSCALE_LAG_CRITICAL_MS`	10 000
`AUTOSCALE_SCALE_DOWN_STABLE_CHECKS`	6
`AUTOSCALE_COOLDOWN_MS`	3 000
`AUTOSCALE_DB_BLOCK_PCT`	80
`DB_POOL_SIZE`	24
`DB_SATURATION_HIGH_WATERMARK_PCT`	80
`DB_BACKPRESSURE_SLEEP_MS`	250
`REDIS_INGESTION_STREAM_MAXLEN`	5 000
`REDIS_PIPELINE_FLUSH_INTERVAL_MS`	10

`streaming`

Prioritizes low write latency over throughput. Uses realtime insert mode — small batches of 10–25 rows written every 100 ms without waiting for COPY thresholds. Worker count starts higher and autoscaling reacts faster. Increases DB connection usage and write IOPS; suitable when near-real-time visibility matters more than raw throughput.

Parameter	Value
`WORKER_COUNT`	8
`BATCH_SIZE`	20
`FLUSH_INTERVAL_MS`	100
`READINGS_BULK_INSERT_MODE`	`realtime`
`READINGS_REALTIME_MAX_ROWS`	100
`READINGS_REALTIME_ROWS_PER_INSERT`	10
`AUTOSCALE_MIN_WORKERS`	6
`AUTOSCALE_MAX_WORKERS`	20
`AUTOSCALE_LAG_TARGET_MS`	2 000
`AUTOSCALE_LAG_SCALE_UP_MS`	1 000
`AUTOSCALE_LAG_CRITICAL_MS`	5 000
`AUTOSCALE_SCALE_DOWN_STABLE_CHECKS`	10
`AUTOSCALE_COOLDOWN_MS`	2 000
`AUTOSCALE_DB_BLOCK_PCT`	80
`DB_POOL_SIZE`	30
`DB_SATURATION_HIGH_WATERMARK_PCT`	85
`DB_BACKPRESSURE_SLEEP_MS`	100
`REDIS_INGESTION_STREAM_MAXLEN`	2 000
`REDIS_PIPELINE_FLUSH_INTERVAL_MS`	0

:::caution Higher DB load streaming keeps fewer messages buffered in Redis and writes much more frequently. Ensure your TimescaleDB instance has enough connection headroom and IOPS before switching to this profile in production. :::

`benchmark`

Stress-testing and capacity planning only. Maximizes raw throughput — very large batches (1 000 rows), high worker ceiling (24), a large Redis stream buffer (50 k messages), and aggressive DB saturation tolerance. Not intended for steady-state production use; scale-down is intentionally slow to maintain sustained pressure during tests.

Parameter	Value
`WORKER_COUNT`	12
`BATCH_SIZE`	1 000
`FLUSH_INTERVAL_MS`	200
`READINGS_BULK_INSERT_MODE`	`copy`
`READINGS_COPY_MIN_ROWS`	500
`AUTOSCALE_MIN_WORKERS`	8
`AUTOSCALE_MAX_WORKERS`	24
`AUTOSCALE_LAG_TARGET_MS`	8 000
`AUTOSCALE_LAG_SCALE_UP_MS`	2 000
`AUTOSCALE_LAG_CRITICAL_MS`	12 000
`AUTOSCALE_SCALE_DOWN_STABLE_CHECKS`	20
`AUTOSCALE_COOLDOWN_MS`	1 000
`AUTOSCALE_DB_BLOCK_PCT`	90
`DB_POOL_SIZE`	40
`DB_SATURATION_HIGH_WATERMARK_PCT`	92
`DB_BACKPRESSURE_SLEEP_MS`	50
`REDIS_INGESTION_STREAM_MAXLEN`	50 000
`REDIS_PIPELINE_FLUSH_INTERVAL_MS`	5

`hp`

High-performance profile for large, stable fleets where message volume is high but consistent. Combines a large Redis stream buffer (100 k), medium batch size (300), a fast 100 ms flush, and a very conservative scale-down policy (30 stable checks before reducing workers). Avoids the oscillation of streaming while still writing faster than balanced. The long AUTOSCALE_SCALE_DOWN_STABLE_CHECKS keeps worker count stable during brief idle windows rather than spinning down and back up repeatedly.

Parameter	Value
`WORKER_COUNT`	10
`BATCH_SIZE`	300
`FLUSH_INTERVAL_MS`	100
`READINGS_BULK_INSERT_MODE`	`copy`
`READINGS_COPY_MIN_ROWS`	500
`AUTOSCALE_MIN_WORKERS`	8
`AUTOSCALE_MAX_WORKERS`	20
`AUTOSCALE_LAG_TARGET_MS`	5 000
`AUTOSCALE_LAG_SCALE_UP_MS`	2 000
`AUTOSCALE_LAG_CRITICAL_MS`	10 000
`AUTOSCALE_SCALE_DOWN_STABLE_CHECKS`	30
`AUTOSCALE_COOLDOWN_MS`	2 000
`AUTOSCALE_DB_BLOCK_PCT`	85
`DB_POOL_SIZE`	30
`DB_SATURATION_HIGH_WATERMARK_PCT`	85
`DB_BACKPRESSURE_SLEEP_MS`	150
`REDIS_INGESTION_STREAM_MAXLEN`	100 000
`REDIS_PIPELINE_FLUSH_INTERVAL_MS`	5

Metrics

The Ingestion service exposes Prometheus metrics at GET /metrics (port 3003):

Metric	Description
`ingestion_messages_processed_total`	Total device batches consumed from Redis
`ingestion_readings_inserted_total`	Individual metric rows written to TimescaleDB
`ingestion_dlq_length`	Current dead-letter queue depth
`ingestion_stream_length`	Pending messages in the ingest stream
`ingestion_worker_lag_ms`	Consumer group lag (ms behind real-time)
`ingestion_worker_count`	Current active worker count
`ingestion_batch_duration_p95_ms`	95th-percentile batch processing time
`ingestion_db_insert_duration_p95_ms`	95th-percentile database insert time

Configuration

# Database
DB_HOST=localhost
DB_PORT=5432
DB_NAME=iotistic
DB_USER=postgres
DB_PASSWORD=postgres
DB_SSL=true
DB_POOL_SIZE=50

# Redis
REDIS_HOST=localhost
REDIS_PORT=6379

# Ingestion tuning
WORKER_COUNT=2              # initial worker count (autoscaling adjusts this)
BATCH_SIZE=100              # max messages per worker cycle
MAX_RETRIES=3               # attempts before DLQ
DB_INSERT_METHOD=insert     # "insert" or "copy"

# Resilience
DISK_SPOOL_ENABLED=true
DISK_SPOOL_PATH=/tmp/iotistic-spool
DISK_SPOOL_MAX_SIZE_MB=500

# Port (health + metrics only)
PORT=3003

Health Check

GET /health

{
  "status": "ok",
  "uptime": 3847,
  "workers": 3,
  "streamLag": 120,
  "circuitBreaker": "closed"
}

A status of degraded indicates the circuit breaker is open or the disk spool is active. A status of error means the service cannot reach either Redis or PostgreSQL.

Why a Separate Service?​

Data Flow​

Redis Streams​

Stream keys​

Consumer group​

Processing Pipeline​

1. Decode payload​

2. Normalization​

3. Deduplication​

4. Database Insert​

Resilience​

Circuit Breaker​

Disk Spool​

Dead-Letter Queue​

Autoscaling​

Ingestion Profiles​

Profile overview​

batch​

balanced​

streaming​

benchmark​

hp​

Metrics​

Configuration​

Health Check​