Telemetry Pipeline

The telemetry pipeline is a background service that batches messages from the adapter and publishes them to NATS JetStream. It implements both ITelemetryClient and IPropertiesClient — a single channel feeds two branches (telemetry vs reported properties). The canonical contract lives in the telemetry-pipeline spec.

Flow

Adapter.Telemetry.Send(deviceId, TelemetryMessage)
Adapter.Properties.Report(deviceId, JsonObject)
       │
       ▼
┌─────────────────────┐
│  Bounded Channel    │  capacity: 50,000 (default)
│  DropWrite on full  │  single reader, multi-writer
└────────┬────────────┘
         │
         ▼
┌─────────────────────┐
│  Batch Accumulator  │  256 messages OR 50 ms window
└────────┬────────────┘
         │
    ┌────┴────────┐
    │             │
    ▼             ▼
┌─────────┐ ┌──────────────────┐
│Telemetry│ │Reported Properties│
│ (batch) │ │ (semaphore-bound) │
└────┬───┘ └────────┬─────────┘
     │              │
     ▼              ▼
  Scatter-      DB twin update
  gather        (jsonb_set + ver++)
  JetStream     +
  publish       JetStream publish

Both branches publish to the same JetStream stream (TELEMETRY); only the messageType header differs.

Hot-path bytes (no JSON round-trip)

TelemetryMessage.Payload is ReadOnlyMemory<byte>. The pipeline writes those bytes directly as the NATS message body — there is no parse/re-serialise on the telemetry branch. Per-message overhead is the channel write and one NATS publish.

If you need per-message routing decisions, do them in the adapter before Telemetry.Send and surface the result as a header or as a separate publish — do not reach into the bytes from inside the pipeline.

Telemetry branch

Telemetry is published to JetStream using a scatter-gather pattern:

1. Scatter — fire all publishes in the batch without awaiting acks.
2. Gather — collect all ack responses (pipelined, ≈1 RTT total).

Subject: telemetry.{providerId}::{deviceId}

Headers added per message:

Header	Value
messageType	ProviderClientTelemetry
providerId	the provider's id
providerClientId	the device id
enqueuedTime	ISO 8601 UTC timestamp

A fresh NatsHeaders instance is allocated per publish call because the NATS .NET client mutates it during publish — do not cache or reuse a NatsHeaders across messages.

Reported-properties branch

Each reported message requires a database update before publishing:

1. Update device_twin_entity atomically — jsonb_set merges the reported document, increments ReportedVersion, sets LastActivity.
2. If the twin row doesn't exist yet, it is created in the same call.
3. Publish to JetStream with messageType: ProviderClientPropertiesChanged.

Concurrency is bounded by MaxReportedConcurrency (default 16) so a fleet-wide reported burst can't exhaust the PostgreSQL connection pool. The DB write and the NATS publish are paired — if the DB write fails, the publish is skipped to avoid storing inconsistent state.

The twin endpoints never accept reported writes; reported state flows exclusively through this branch.

Backpressure

The channel uses BoundedChannelFullMode.DropWrite. When the channel is full:

• The new message is dropped (not queued, not blocked).
• pipeline.messages.dropped increments.
• A warning is logged, rate-limited so a sustained outage doesn't flood logs.

This guarantees Telemetry.Send and Properties.Report never block the adapter, even under extreme load. The trade-off is silent loss under sustained over-capacity; monitor the dropped counter and tune ChannelCapacity accordingly.

Configuration

{
  "TelemetryPipeline": {
    "ChannelCapacity": 50000,
    "BatchSize": 256,
    "FlushIntervalMs": 50,
    "MaxReportedConcurrency": 16
  }
}

Option	Default	Description
ChannelCapacity	50,000	Bounded channel size. Messages dropped when full. Sized to absorb ≈5 s of replication lag at 10 k msg/s.
BatchSize	256	Maximum messages per batch before flush.
FlushIntervalMs	50	Maximum wait time before flushing a partial batch.
MaxReportedConcurrency	16	Parallel reported-property DB updates. Bounds connection-pool use under burst.

Metrics

Metric	Type	Tags
pipeline.messages.received	Counter	type=telemetry|reported
pipeline.messages.published	Counter	—
pipeline.messages.dropped	Counter	—
pipeline.errors	Counter	stage=parse|db|nats-ack|nats-reported
pipeline.batch.duration_ms	Histogram	—
pipeline.batch.size	Histogram	—
pipeline.nats.publish_ms	Histogram	scatter-gather duration
pipeline.db.update_ms	Histogram	reported DB update + NATS publish
pipeline.reported.semaphore_wait_ms	Histogram	wait for reported concurrency slot
pipeline.channel.depth	Gauge	current channel reader count

JetStream stream provisioning

The pipeline creates the TELEMETRY JetStream stream at startup if it doesn't exist (subjects: telemetry.>). Replication is decided by the NATS monitoring port (Nats:MonitoringPort, default 8222):

• Cluster size ≥ Nats:ClusterReplicaThreshold (default 3) → NumReplicas = 3.
• Otherwise → NumReplicas = 1.

The BEACONTOWER_EVENTS stream used by the CloudEvents publisher is created the same way.