Data contract

This page is the contract for integrators receiving data from connector.push_endpoint.

Record schema (CanonicalMachineData, telemetry.v1)

{
  "source": "modbus",
  "timestamp": "2026-08-01T09:30:00Z",
  "machine": { "id": "PLC-01", "name": "Assembly Line PLC" },
  "status": { "state": "running", "alarmCode": "E102", "alarmMessage": "spindle overload" },
  "axes": [ { "name": "X", "position": 12.5, "unit": "mm" } ],
  "spindle": { "speed": 8000, "load": 41.5 },
  "quality": "good",
  "valueQuality": { "spindle.load": "uncertain" },
  "metadata": { "temp": 61.2 }
}

The schema sets additionalProperties: false everywhere except metadata: a script that adds a new top-level field produces an invalid record (handled per transform.on_error). Put custom values under metadata. The canonical schema is published in the opex-contracts repo (schemas/telemetry.v1.json, since contracts v0.4.0).

quality and valueQuality are additive and optional, but because of additionalProperties: false a receiver still validating against contracts v0.3.0 rejects records that carry them (or status.state: "unknown"): upgrade receivers to v0.4.0 before upgrading the connector.

Data quality

NATS canonical records pass through with whatever quality the producer set. The push to batch.format: opexmx maps quality/valueQuality onto the per-reading quality string.

Request formats

Single record (default): body is the record JSON, headers Content-Type: application/json and the signature headers above.

Batch, batch.format: hmac: body is {"connector":"<name>","sequence":N,"records":[ ...records ]} with the signature headers and X-Batch-Id: <connector>:<N>. Retries are byte-identical.

Batch, batch.format: opexmx: Authorization: Bearer <batch.api_token>, body {"sequence":N,"backfill":false,"metadata":{"sentAt":"…"},"readings":[…]}, X-Batch-Id, no body signature. sequence is the dedupe key.

Batching requires push_queue.enabled: true.

Verifying the signature

Every push carries three signature headers (batches included, except format: opexmx):

Prefer the timestamped signature and reject requests outside a replay window (±5 minutes). Always sign the exact raw bytes received, never re-serialized JSON, compare in constant time and check lengths first.

// Node.js (Express: use express.raw({ type: "application/json" }))
const crypto = require("crypto");
function verify(rawBody /* Buffer */, headers, secret, windowSecs = 300) {
  const ts = Number(headers["x-opex-timestamp"]);
  const sig = headers["x-opex-signature"];
  if (!Number.isFinite(ts) || typeof sig !== "string") return false;
  if (Math.abs(Date.now() / 1000 - ts) > windowSecs) return false;
  const expected = Buffer.from("sha256=" + crypto.createHmac("sha256", secret)
    .update(Buffer.concat([Buffer.from(ts + "."), rawBody])).digest("hex"));
  const got = Buffer.from(sig);
  return got.length === expected.length && crypto.timingSafeEqual(got, expected);
}

# Python (Flask: request.get_data())
import hmac, hashlib, time

def verify(raw_body: bytes, headers, secret: str, window: int = 300) -> bool:
    ts, sig = headers.get("X-Opex-Timestamp"), headers.get("X-Opex-Signature")
    if not ts or not sig or not ts.isdigit() or abs(time.time() - int(ts)) > window:
        return False
    mac = hmac.new(secret.encode(), ts.encode() + b"." + raw_body, hashlib.sha256)
    return hmac.compare_digest(sig, "sha256=" + mac.hexdigest())

// Go
func verify(raw []byte, ts, sig, secret string, window time.Duration) bool {
	n, err := strconv.ParseInt(ts, 10, 64)
	if err != nil || time.Since(time.Unix(n, 0)).Abs() > window {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(ts + "."))
	mac.Write(raw)
	return hmac.Equal([]byte(sig), []byte("sha256="+hex.EncodeToString(mac.Sum(nil))))
}

Delivery semantics

• At-least-once, ordered. A timed-out request may have been processed; the retry produces a duplicate. Single-record pushes have no idempotency key: dedupe on (machine.id, timestamp, source). Batches carry X-Batch-Id / sequence.
• Duplicates also originate upstream of the push. NATS JetStream ingest is acknowledged only after the record is handled downstream (for the push sink: stored in the push queue), so a failed sink, a restart or a slow pipeline redelivers the source message and the same record can be pushed twice. A map fan-out whose children partly failed re-runs the children that had succeeded. Dedupe on (machine.id, timestamp, source); see “NATS delivery semantics” in the sources guide.
• Retry: 408, 429, 5xx and transport errors back off exponentially (honoring Retry-After, capped at 300 s). 401/403 are treated as credential problems: the connector backs off and keeps the row. 409 means “already have it” and is acknowledged. Any other 4xx is permanent: the row moves to the dead-letter table (GET /api/push/dead, replay with POST /api/push/dead/replay, purge with DELETE /api/push/dead; queue health at GET /api/push/status).
• Durability: without push_queue.enabled the retry buffer is in memory (1000 entries, oldest evicted) and lost on restart. Enable the durable queue for store-and-forward across restarts and outages; it is bounded by max_quota_mb (default 500) and evictions are counted.
• Preprocessing (deadband / aggregation) only affects the push path.

Replay protection

X-Opex-Signature binds a send-time timestamp to the body, so a captured request is only replayable inside your window. Enforce the window, TLS, and still dedupe on (machine.id, timestamp, source) / X-Batch-Id, since an unacknowledged request is legitimately re-sent (with a fresh timestamp). batch.format: opexmx is authenticated by the bearer API key instead of a body signature. To rotate hmac_secret, set connector.hmac_key_id, accept both keys on the receiver keyed by X-Opex-Key-Id, then retire the old one.