# Historian Bus Worker ## Definition A historian worker is a consumer component that subscribes to one or more canonical semantic MQTT buses and writes normalized measurements into the PostgreSQL telemetry historian. Its role is the inverse of an adapter: - adapters translate vendor-specific inputs into canonical bus topics - historian workers translate canonical bus topics into historian API calls The worker is a consumer of the semantic contract, not an owner of it. --- ## Purpose The worker exists to decouple historian persistence from: - vendor topic structures - protocol-specific payload formats - source-specific device identities - source-side timing quirks This keeps the historian integration generic enough to ingest any bus that follows the shared MQTT contract. --- ## Architectural Position The historian worker operates at the egress side of the semantic bus. Pipeline: Device / External System ↓ Protocol Adapter ↓ Canonical MQTT Bus ↓ Historian Worker ↓ `telemetry.ingest_measurement(...)` ↓ Historian tables The worker consumes canonical MQTT only. It MUST NOT subscribe to raw vendor topics such as `zigbee2mqtt/...` for normal historian ingestion. --- ## Shared Inputs The worker consumes data defined by: - `mqtt_contract.md` - bus-specific contracts such as `home_bus.md` and `energy_bus.md` - `tdb_ingestion/mqtt_ingestion_api.md` - `tdb_ingestion/counter_ingestion_api.md` when cumulative counter metrics are present on the bus The worker should treat the semantic bus as the source of truth for topic grammar and stream meaning. The database API is the source of truth for persistence behavior. Boundary decisions are recorded in `consolidated_spec.md` §10. --- ## Responsibilities The historian worker is responsible for: 1. Topic subscription Subscribe to canonical topics for the target buses. 2. Topic parsing Extract `metric_name`, `device_id`, `stream`, and bus-specific dimensions from the topic. 3. Payload decoding Decode scalar payloads and Profile B envelopes according to `mqtt_contract.md`. 4. Meta caching Consume retained `meta` topics and keep a local cache keyed by topic stem. 5. Historian policy enforcement Respect `meta.historian.enabled` and `meta.historian.mode` when deciding what should be persisted. 6. Type selection Choose the numeric or boolean PostgreSQL overload correctly for measurement-style metrics. 7. Timestamp resolution Use `observed_at` from the envelope when present, otherwise fall back to ingestion time. 8. Per-stream ordering Preserve source order for the same `(metric_name, device_id)` path when sending measurements to PostgreSQL. 9. Error handling Separate permanent message errors from transient infrastructure failures. 10. Observability Emit worker health, counters, and persistence failures on operational topics. --- ## Explicit Non-Responsibilities The historian worker must NOT: - parse vendor-specific topics as part of the normal pipeline - redefine semantic topic contracts - invent new device identities outside the bus contract - perform automation logic - aggregate unrelated sensors into synthetic measurements - reimplement historian segment logic already handled in PostgreSQL - repair malformed semantic messages silently If semantic messages are invalid, the worker should surface them operationally and skip persistence. --- ## Subscription Model Recommended subscriptions per site: - `+/home/+/+/+/value` - `+/home/+/+/+/last` - `+/energy/+/+/+/value` - `+/energy/+/+/+/last` - `+/home/+/+/+/meta` - `+/energy/+/+/+/meta` Optional observability subscriptions: - `+/home/+/+/+/availability` - `+/energy/+/+/+/availability` - `+/sys/adapter/+/availability` - `+/sys/adapter/+/error` Rules: - `meta` SHOULD be subscribed and cached before relying on enrichment - `set` topics MUST be ignored by the historian worker - `last` topics MUST be ignored for normal time-series ingestion - `availability` SHOULD NOT be persisted as telemetry samples unless a separate policy explicitly requires it --- ## Topic-to-Historian Mapping ### Home Bus Topic: `/home////` Mapped fields: - `metric_name = ` - `device_id = .` ### Energy Bus Topic: `/energy////` Mapped fields: - `metric_name = ` - `device_id = .` The worker MUST NOT require database-side topic parsing. --- ## Payload Handling ### Scalar Payload If the stream uses Profile A: - parse the scalar payload as number, boolean, or enum string - resolve `unit` from cached retained `meta` - set `observed_at` to ingestion time unless source time is available out of band ### Envelope Payload If the stream uses Profile B: - read `value` - use `observed_at` when present - use `unit` from payload, falling back to cached `meta.unit` - use `quality` when present for logs or side metrics ### Type Compatibility Rule The current PostgreSQL historian API directly supports: - numeric values - boolean values String enum states are allowed on the semantic bus, but the worker SHOULD skip them unless there is an explicit encoding policy. Examples: - `open` / `closed` may be mapped to boolean if a metric policy expects that - `heat` / `cool` / `off` MUST NOT be guessed into arbitrary numeric values ### Counter Metric Boundary Counter-style cumulative metrics such as `energy_total`, `import_energy_total`, `export_energy_total`, `rx_bytes_total`, or `tx_packets_total` are valid semantic bus values, but they are not covered by the current measurement ingestion API. Rules: - the worker MUST NOT force cumulative counters through `telemetry.ingest_measurement(...)` just because the payload is numeric - the worker SHOULD route them to the separate counter ingestion path defined in `tdb_ingestion/counter_ingestion_api.md` - if the counter path is temporarily unavailable, the worker SHOULD skip them explicitly and expose that through operational stats or DLQ - measurement-style metrics such as `active_power`, `voltage`, `current`, `temperature`, and `soc` continue through the existing measurement API --- ## Stream Policy Default persistence policy: - ingest `value` by default - ignore `last` - use `meta.historian.mode` to interpret whether a `value` stream represents `sample`, `state`, or `event` semantics - ignore `set` - never treat `meta` itself as a measurement Additional rules: - if `meta.historian.enabled=false`, the worker MUST skip persistence - if `meta` is missing, the worker MAY continue with degraded defaults - missing `meta` MUST NOT block ingestion of otherwise valid numeric or boolean `value` samples --- ## Ordering and Delivery Ordering is critical because PostgreSQL enforces append-only writes per `(metric_name, device_id)`. Rules: - the worker MUST serialize writes for the same `(metric_name, device_id)` - retries MUST preserve order - the worker SHOULD partition concurrency by `(metric_name, device_id)` or an equivalent stable shard - out-of-order database errors SHOULD be treated as permanent message errors unless the worker can prove replay ordering was preserved Operational implication: - broad parallelism is acceptable across distinct devices or metrics - unordered fan-out for the same logical stream is not acceptable --- ## Meta Cache The worker should maintain a retained metadata cache. Recommended cache key: - topic stem without the final stream segment Example: - topic: `vad/home/bedroom/temperature/bedroom-sensor/meta` - cache key: `vad/home/bedroom/temperature/bedroom-sensor` Recommended cached fields: - `payload_profile` - `data_type` - `unit` - `historian.enabled` - `historian.mode` - `schema_ref` - `adapter_id` - `source` - `source_ref` Rules: - cached `meta` SHOULD be replaced atomically when newer retained data arrives - worker startup SHOULD tolerate receiving live `value` traffic before the corresponding retained `meta` - if `meta` is deleted, the cache entry SHOULD be removed as well --- ## Database Interaction The worker writes measurement-style samples through `telemetry.ingest_measurement(...)` as defined in `tdb_ingestion/mqtt_ingestion_api.md`. Rules: - use the numeric overload for numeric values - use the boolean overload for boolean values - type `NULL` explicitly if unknown values are ever supported by bus policy - log the returned `action` for observability - do not duplicate historian logic already implemented in PostgreSQL - do not send counter-style cumulative totals through this API; those follow the separate contract in `tdb_ingestion/counter_ingestion_api.md` The worker should assume the database is responsible for: - deduplication - gap detection - segment splitting - policy boundary handling --- ## Operational Topics Shared operational namespace rules are defined in `sys_bus.md`. The worker should expose its own operational topics under: - `/sys/historian//availability` - `/sys/historian//stats` - `/sys/historian//error` - `/sys/historian//dlq` Recommended uses: - `availability`: retained online/offline state - `stats`: low-rate counters such as ingested messages, skipped samples, and retry counts - `error`: structured infrastructure or persistence failures - `dlq`: semantic messages rejected by the worker This keeps historian worker observability separate from adapter observability. --- ## Failure Handling Permanent message failures include: - topic does not match a supported bus grammar - payload type cannot be mapped to the target historian type - unknown metric in PostgreSQL - out-of-order measurement - value violates metric policy Transient failures include: - PostgreSQL unavailable - network interruption between worker and database - temporary broker disconnect Rules: - permanent message failures SHOULD be logged and dead-lettered - transient failures SHOULD be retried without reordering per stream - ambiguous retries that later return out-of-order errors SHOULD be logged as likely duplicates --- ## Deployment Model A historian worker may be deployed in one of the following ways: - one worker for all buses - one worker per bus - one worker per site and bus Recommended starting model: - one worker per site consuming both `home` and `energy` This keeps the first implementation simple while preserving the option to split workers later for throughput or failure-domain reasons. --- ## Relationship with Adapters Adapters and historian workers are symmetrical integration roles. Adapter direction: - external protocol -> canonical bus Historian worker direction: - canonical bus -> historian API Both should be: - contract-driven - deterministic - observable - easy to replay in tests Neither should contain business automation logic. --- ## Design Principles 1. Consume canonical topics only The worker should depend on semantic contracts, not vendor payloads. 2. Keep persistence generic The same worker model should work for `home`, `energy`, and future buses. 3. Preserve ordering Correct historian writes matter more than maximum ingest parallelism. 4. Fail visibly Bad messages and persistence problems must be observable on operational topics. 5. Reuse shared contract rules The worker should not invent alternate payload or metadata semantics.