Sentinel: Ingest Path and Log Schema

Last week's sensor is still reporting 71.6 in Fahrenheit, and its driver has published that number untouched, as the contract requires. The conversion to Celsius happens here instead, at the eighth of thirteen stages between the bus and the database, and the row it writes keeps both numbers: 22.0 in the canonical column, and 71.6 in the native one beside the unit the device actually used.

Keeping both is what makes a conversion reversible. Suppose that weeks later the descriptor's scale factor turns out to have been wrong. Nothing collected in the meantime is lost, because the number the device produced was never overwritten; a reprocess recomputes every canonical value from native under a corrected registry version, rebuilds the current state of the affected points, and refreshes the aggregates built over them. It can only reach rows that still exist, and it refuses outright rather than correcting the part of a request it can reach. What it will not do is re-run the rules that fired on the wrong numbers, because those decisions were made on the data as it stood.

The other load-bearing decision is smaller and much easier to get wrong: where ingested_at comes from. Timescale requires the partitioning column in every unique index, so the dedup key cannot be just the three columns that identify a message; it has to carry the timestamp as well. Stamp that from the core's clock at processing time, and a message redelivered after a crash arrives with a new timestamp, slips past the index and inserts a duplicate, while the index goes on looking as though it enforces dedup. Take it from the bus instead, where it is fixed once and identical on every redelivery, and the same index turns at-least-once delivery into exactly-once effects without a watermark table of its own.

Those two decisions are the spine of this document. The two logs, the per-class hypertables, the thirteen stages and their ordering rules, the restart sequence and the reprocess command are what the rest of the schema has to look like once both are held fixed.

What this document owns: the split into an observation log and a control-event log, the observation schema and its physical split into one hypertable per retention class, where ingested_at comes from and why, the versioned and content-addressed registry, the thirteen-stage ingest pipeline and its three ordering rules, the in-memory projection and its checkpoint, the restart sequence, reprocess and its refusal to apply partially, and the properties of the continuous aggregates that belong with the schema rather than the policy. What it deliberately does not own is how long anything is kept, how each class rolls up and how state-class compaction works, which belong to Retention & Compaction; keeping cold-start rehydration bounded, which belongs to Watchdog & Derived Points; the core epoch, restarting against a backlog and the utilisation arithmetic the ingest ceiling feeds into, which belong to Core Runtime; and who a principal is and why a wrong_instance count is an alert rather than a dashboard figure, which belong to Security Model. Everything downstream that asks what a point's value is, or was, is reading a projection of the two logs defined here.

Two things drive the whole design here: the bus message's own identity doubles as the dedup key and the ingest watermark, and value_native in the log is what makes conversion reversible. Get those right and restart and reprocess both fall out.

Two logs, not one

The pure event-sourcing instinct is one ordered log. It is the wrong shape here. Observations and control events have different volumes, retention, and query shapes, and Timescale hypertables want homogeneous rows.

• observation — high volume, append-only, compressed, aged out. Ordering matters only within a point. Physically several hypertables, one per retention class; see below.
• control_event — commands, availability, quality transitions, assignment changes, reprocess operations. Low volume, globally sequenced, kept forever, not classed.

The projection is built from both. Quality transitions from the freshness watchdog go in control_event because they carry no new value; in a healthy system they are rare, and when they are not rare that is the thing most worth a permanent record.

Schema

CREATE TYPE quality AS ENUM (
  'unknown','restored','live','stale','unavailable','bad','assumed');

-- identical definition for each retention class:
--   observation_measurement, observation_state, observation_diagnostic
CREATE TABLE observation_<class> (
  ingested_at       timestamptz  NOT NULL,   -- bus-accept time; see below
  point_id          text         NOT NULL,
  observed_at       timestamptz  NOT NULL,
  device_time       timestamptz,

  value_num         double precision,   -- canonical
  value_bool        boolean,
  value_text        text,               -- canonical enum member / string
  native_num        double precision,
  native_text       text,               -- raw device representation

  unit_native       text,
  quality           quality      NOT NULL,
  reason            text,
  detail            jsonb,

  changed           boolean      NOT NULL,
  registry_version  integer      NOT NULL,
  driver_instance   text         NOT NULL,
  source_epoch      integer      NOT NULL,
  source_seq        bigint       NOT NULL,
  skew_flagged      boolean      NOT NULL DEFAULT false
);

SELECT create_hypertable('observation_<class>','ingested_at',
                         chunk_time_interval => INTERVAL '1 day');

CREATE UNIQUE INDEX ON observation_<class>
  (driver_instance, source_epoch, source_seq, ingested_at);
CREATE INDEX ON observation_<class> (point_id, observed_at DESC);

Typed columns rather than JSONB for values. JSONB costs space on the highest-volume table and blocks index-only scans; five nullable narrow columns cost less, with only one or two populated per row.

One hypertable per retention class. Timescale attaches retention, compression and continuous-aggregate policies to a table rather than to a row, so per-class policy means a physical split. The class comes from the resolved descriptor, which ingest already has by the time it writes. The policies themselves are in Retention & Compaction; what belongs here is that the split exists, that the schema is identical across classes, and that it cannot be deferred — adding the class field later is easy, splitting a populated table is a rewrite of everything ever collected.

An observation view over the three, as UNION ALL, exists for ad-hoc querying. The hot paths do not use it: rehydration and reprocess resolve the class from the descriptor and query the specific table.

Chunk on ingested_at, not observed_at. A driver host replaying a buffered partition writes rows with old observed_at values, which would land in chunks that are already compressed. ingested_at is monotonic by construction, so chunks close cleanly and never reopen. The cost is that analytical queries filtering on observed_at lose chunk exclusion, which the secondary index mostly covers at these volumes.

Every observation writes a row, including one that changed nothing. Suppressing unchanged rows at write time is a tempting optimisation and it breaks two mechanisms: derived-point windows rehydrate by querying these tables, so mean, count and min_samples need the samples that were suppressed; and reprocess can only reconvert rows that exist. Volume is managed afterwards, by retention and compaction, where the trade is explicit and bounded.

Where ingested_at comes from

It is the JetStream message timestamp, not now() at processing time. This is load-bearing and easy to get wrong.

Timescale requires the partitioning column in every unique index, so the dedup index is necessarily (driver_instance, source_epoch, source_seq, ingested_at) rather than the three-column key that would otherwise be correct. If ingested_at were assigned when the core happened to process the message, a redelivery after a crash would carry a different value, land outside the index, and insert a duplicate row. The index would look like it was enforcing dedup and would not be.

The bus timestamp is fixed when the message enters the stream and is identical on every redelivery, which makes the four-column index behave exactly like the three-column one. It also keeps the monotonicity that chunking depends on, because a single obs stream assigns timestamps in stream order. If observations ever arrive from more than one stream, that property weakens and this decision needs revisiting.

The cost is a change of meaning: ingested_at is when the bus accepted the observation, not when the core recorded it. That is the more useful of the two anyway — it is the moment the observation became durable, it comes from one clock rather than from whichever core process handled it, and it is stable across a core restart. Core processing lag belongs on a self point, not on a column of the highest-volume table in the system.

CREATE TABLE control_event (
  seq          bigserial PRIMARY KEY,
  occurred_at  timestamptz NOT NULL,
  kind         text        NOT NULL,   -- quality_changed, command_*, availability, assignment, reprocess
  point_id     text,
  subject_key  text,                   -- driver_instance / device / component
  principal    text        NOT NULL,   -- who caused it; see Security Model
  core_epoch   integer     NOT NULL,   -- which core instance; see Core Runtime
  payload      jsonb       NOT NULL,
  registry_version integer  NOT NULL
);

principal and core_epoch are both NOT NULL deliberately. A code path that cannot name who asked, or which core acted, cannot write a control event — which means it cannot change state.

The registry is versioned data

Vocabularies and point descriptors load from git into Postgres. Every registry load creates a version row, and descriptors are stored per version by content hash:

CREATE TABLE registry_version (
  version    integer PRIMARY KEY,
  loaded_at  timestamptz NOT NULL,
  git_sha    text        NOT NULL,
  content_hash text      NOT NULL
);

CREATE TABLE descriptor_blob (
  hash       text PRIMARY KEY,          -- of the canonicalised descriptor
  descriptor jsonb NOT NULL
);

CREATE TABLE point_descriptor (
  version   integer NOT NULL REFERENCES registry_version,
  point_id  text    NOT NULL,
  hash      text    NOT NULL REFERENCES descriptor_blob,
  PRIMARY KEY (version, point_id)
);

The indirection exists because of scale. At a few hundred points a full descriptor copy per version is a rounding error; at the four thousand points targeted in Core Runtime it is around 1.6 MB per commit, growing without bound, when a typical commit changes a handful of points. Content-addressing collapses that to one small row per point per version plus one blob per genuinely new descriptor.

It preserves the property that matters: any historical observation can be reinterpreted under the exact descriptor that produced it, or under any other version, because every version still names a complete descriptor set.

Ingest pipeline

Stages, in order, all core-side:

1. Consume from the JetStream durable consumer on obs.>, taking ingested_at from the message
2. Validate envelope — required fields, parseable timestamps, observed_at not absurdly in the future
3. Dedup on (driver_instance, source_epoch, source_seq) against an in-memory set of recently seen keys; this is a fast path, and the unique index is the authority
4. Resolve point — registry lookup; miss goes to quarantine and the message is acked
5. Check publisher — the host and instance assigned to this point must match the ones the message was published from; a mismatch goes to quarantine with reason wrong_instance and the message is acked
6. Unit check — driver-supplied unit_native disagreeing with the descriptor gives quality bad, reason unit_mismatch, no conversion attempted
7. Validate value — range check, enum codebook lookup; failures give bad with the offending value preserved in native_text
8. Convert — value_native × factor + offset, affine-aware
9. Skew check — device_time against observed_at, sets skew_flagged
10. Evaluate change — deadband against the in-memory projection, sets changed
11. Append to the hypertable for the descriptor's retention class, batched, as INSERT ... ON CONFLICT DO NOTHING ... RETURNING
12. Commit, then ack the JetStream messages
13. Update projection and publish transitions to state.{point_id}, for returned rows only

Stage 5 is what makes bus permissions mean anything. Subject scoping confines a host to publishing under its own prefix, but says nothing about which point IDs it puts inside those subjects. Without this check, one compromised driver host could publish readings for every point in the system, including the inputs to thermal interlocks, entirely within its permissions. It is one comparison against the assignment mapping the registry already holds.

The result is a security signal rather than a data-quality one. A unit_mismatch is a configuration error; a wrong_instance is a host claiming points that were never assigned to it, and the count belongs in the alerting path rather than on a dashboard. See Security Model.

Stage 11 batches per class, since a batch has to target one table. That is a grouping in the flush, not three pipelines — a batch is partitioned by class immediately before the insert and committed in one transaction.

Stages 2 to 10 are the throughput ceiling, not the insert. Per-observation validation, registry lookup, codebook resolution and conversion all happen once per row in application code, and that is where sustainable ingest rate is actually determined. It is the unmeasured term in the utilisation arithmetic in Core Runtime, and it is worth benchmarking against the design rate before sizing anything that depends on it.

Three ordering rules that are easy to get wrong and expensive to debug:

Ack after commit. The consumer's ack position is the ingest watermark. Crash between commit and ack, and the redelivered messages carry the same (instance, epoch, seq, ingested_at) and are swallowed by the unique index. Crash before commit, and they are redelivered and processed properly. This produces exactly-once effects from at-least-once delivery without a separate watermark table — but only because ingested_at is deterministic. If that ever changes, the whole property goes with it.

Publish after commit. Rules must never fire on data that could roll back. A relay that clicks because of an aborted transaction is a bad afternoon.

Project only what actually inserted. ON CONFLICT DO NOTHING makes a redelivered batch a no-op in the table, but the projection update and the transition publish are not idempotent in the way that matters: re-publishing a transition re-fires the rules engine on an edge that already happened. Stage 13 is driven from the RETURNING set rather than from the batch submitted. This is the redelivery case doing its job silently, and it is invisible in testing unless a core is crashed mid-batch on purpose.

Inserts are batched — multi-row INSERT flushed on 500 rows or 100 ms, whichever comes first. Per-row inserts will not keep up at this scale, and the batch window is invisible next to the polling intervals.

Failure at stages 6–8 still writes a row. A bad observation is data — it refreshes freshness, it feeds bad-data rule triggers, and it is the evidence when working out why a sensor went strange at 3am. Silence is the one outcome that helps nobody.

Failures at stages 4 and 5 do not write a row, because there is no descriptor to write one against, and therefore no class to route to. Both go to quarantine with the full native payload.

Projection

The authoritative copy is in memory in the core. Postgres holds a checkpoint:

CREATE TABLE point_state (
  point_id        text PRIMARY KEY,
  value_num       double precision,
  value_bool      boolean,
  value_text      text,
  quality         quality     NOT NULL,
  reason          text,
  observed_at     timestamptz,
  ingested_at     timestamptz,
  changed_at      timestamptz,          -- last value change past deadband
  quality_since   timestamptz NOT NULL,
  watermark_seq   bigint                -- last control_event applied
);

Write policy: immediately on quality change, at most every 5 s per point on value change, plus a full flush every 30 s. Per-observation UPDATE on a hot table generates MVCC bloat for no benefit — the value churn does not need durability because the log already has it. Quality changes are rare and operationally important, so those go through synchronously.

The value in point_state is never nulled when quality degrades. Consumers get (value, quality, observed_at) and decide.

Restart

1. Load vocabularies and current registry version
2. Load point_state, force every quality to restored (except rows already unknown)
3. Resume the JetStream consumer from its ack position; replay anything the checkpoint missed
4. Replay control_event above each point's watermark_seq
5. Rehydrate derived-point windows by querying the appropriate class table for the required lookback
6. Rehydrate durable timers from control_event
7. Start drivers; instances go connecting → connected → ready
8. Rules leave warmup as their inputs reach live, or when their warmup deadline expires

Step 3 can be blunt precisely because of the dedup key: anything already committed is re-consumed and discarded, so the consumer resumes from a conservative position without the core having to reason about how far it got.

Step 5 is the quiet payoff of log-first. A 15-minute moving average needs no separately persisted ring buffer — it is a query. Derived points become stateless to restart, which removes an entire category of subtle post-restart wrongness. It is also the step that dominates cold start, and Watchdog & Derived Points covers how to keep it bounded.

Step 2's blanket restored is deliberately pessimistic. Something that was bad before the restart might be fine now, and the driver will say so within one poll interval.

Where the restart follows an outage long enough to have built a backlog, steps 3 onward run against a draining queue rather than a current one. What that changes is in Core Runtime.

Reprocess

value_native is never rewritten. That is what makes this possible:

reprocess --point-glob 'boiler.*' --from 2026-06-01 --registry-version current

Recompute canonical values from native under a chosen registry version, update the affected rows, rebuild the projection for those points, refresh every dependent continuous aggregate over the affected range, and write a reprocess entry to control_event recording the scope, the old version, the new one, and the aggregate ranges refreshed.

ingested_at is never rewritten either. Reprocess changes what a row means, not when it arrived, and the dedup key has to keep pointing at the same message.

Refreshing the aggregates is part of the operation, not a side effect of it. Timescale records an invalidation when the underlying rows change, but the background refresh policy only revisits its own recent window — typically the last day or two. A correction to June's data leaves an invalidation nobody will ever process, and the aggregates keep serving the old numbers indefinitely while the raw table is right. That divergence is worse than the original error, because the raw table and the dashboard now disagree and nothing says so.

So the reprocess command calls refresh_continuous_aggregate explicitly for each aggregate over the affected range, after the row updates commit. Two details:

• Round the range outward to bucket boundaries. A range that starts mid-bucket produces a partially recomputed bucket, which is a worse artefact than either the old or the new value.
• The refresh is point-blind. Aggregates bucket by time, not by point glob, so refreshing recomputes every point in those buckets, not just the ones reprocessed. That is more work than the reprocess itself and it is the right trade: filtering by point would mean maintaining a parallel invalidation map, and the whole operation is rare.

The horizon is per class, and partial reprocess is refused

Reprocess can only reach rows that still exist, and raw retention differs by retention class. A point glob can therefore span classes with different horizons, and a range that is valid for a measurement point may be long gone for a diagnostic one.

The command validates the whole scope first and refuses outright if any point in it cannot be covered, reporting which points and which classes made it impossible. It does not correct the part it can reach.

That refusal is the important behaviour. Silently correcting two thirds of a glob leaves the system in a state where some points are right, some are wrong, and nothing distinguishes them — which is precisely the divide-and-be-wrong failure the operation exists to prevent. Narrowing the glob or the range is then a deliberate second command, taken with the horizon in view.

The practical consequence: a unit error found within a class's raw retention is fully correctable, and one found after that is permanent in the aggregates. If a horizon is wrong, the retention policy is the thing to change, not this.

Compressed chunks need decompress-update-recompress, which is slow, but this is a maintenance operation run rarely. It is built as a real command with a dry-run mode from the start, while the surface is small. Dry run reports the row count, the classes in scope with their horizons, the aggregate ranges it would refresh, and a time estimate, because the refresh is usually the expensive half.

Reprocess does not re-run rules. Historical automation decisions were made on the data as it stood, and rewriting them would make the control event log a lie.

Retention and rollups

The continuous aggregates are defined here as schema; how long anything is kept, which functions each class rolls up with, and how state-class compaction works are all in Retention & Compaction.

Three properties of the aggregate definitions belong with the schema rather than the policy:

Store sum and count, not avg. An average is a view expression over the two. The moment anything rolls an hourly bucket up to a day, or a minutely bucket up to an hour, an average of averages is wrong wherever the bucket counts differ — and it is wrong quietly, by a few percent, in exactly the direction nobody checks. Sum and count compose; averages do not.

Aggregates are defined directly on the raw class tables, never hierarchically. Building the hourly on the minutely would be cheaper to maintain and would introduce a refresh ordering dependency that reprocess has to get right every time. Defining both on raw costs some materialisation work and removes the ordering question entirely.

agg_safe is enforced in the aggregate definition, not only in derived points. An hourly mean of RSSI in a continuous aggregate is exactly as wrong as one computed live, and much harder to notice once it is sitting in a dashboard.

Only rows where quality = 'live' are aggregated, and live_count is carried alongside count so a gappy hour is visibly gappy rather than silently averaged over three samples.

Deleting data by age happens in Postgres, under the policies in Retention & Compaction. The observation stream on the bus is bounded by size instead, for the reasons in Driver Contract.

Exit criteria

This is the step whose claims are most expensive to discover are false, and most of them are otherwise only prose. Prose cannot fail loudly. Each of the following is a test that runs in CI with a fault injector, not a procedure performed by hand once.

Exactly-once effects from at-least-once delivery. Kill the core mid-batch, restart, and assert two things: no duplicate rows, and no transition re-published for a row that was already committed. The second half is the one that gets missed, because the table looks correct either way — it is the rules engine that would have fired twice.

Both crash windows, separately. Crash before commit: the messages are redelivered and processed normally. Crash after commit but before ack: the redelivered messages are swallowed by the unique index and produce no projection update. If the second case ever starts inserting duplicates, ingested_at has stopped being deterministic and the whole property has gone with it.

Publisher check. An observation for a point assigned to a different host or instance is quarantined with reason wrong_instance, no row is written, and the message is acked.

Class routing. One point per retention class, each landing in its own hypertable; a count over the UNION ALL view equals the sum of the three. A point whose class changes is not silently split across two tables.

Reprocess refuses rather than partially applies. A dry run over a glob spanning classes, with one point out of horizon, refuses and names the offending points and classes. Assert that no rows were changed and no aggregate was refreshed.

Reinterpretation under an older registry version reproduces the original canonical value from native, which is the property that makes the whole log-first argument true.

Two measurements to record rather than assert, because their thresholds are open:

• Sustainable ingest throughput against the design rate, measured with synthetic load through stages 2 to 10 rather than against the insert alone.
• Cold-start duration, broken down by startup step.

On Monday - the Dev Diary. Not one finding this time but the whole apparatus: the five instruments that have grown up around the specification, and the different thing that keeps each of them honest.

The next part - The Watchdog and Derived Points. One timer wheel for every freshness deadline, why a sensor that only reports on change cannot be told apart from a dead one unless it is made to heartbeat, and how the restart step that dominates cold start is kept bounded.

Start of the series: An Introduction. The map, the two decisions every later document is downstream of, and why a specification at this scale is being published in public while the system it describes gets built.