Writing readings

Routed writes, the bulk path for steady-state ingest, idempotent redelivery, and knowing what a write displaced.

Two entry points, and when each is right

append routes each interval to the tier that owns it. That is the only safe way to record a correction for an already-archived interval: such a row cannot go to PostgreSQL, because it would sit below the watermark where no query looks, so the correction would be silently ignored while appearing to have been accepted.

let outcome = store.append(&[stored_series]).await?;
outcome.hot_rows;               // landed in PostgreSQL
outcome.cold_rows;              // late corrections, appended to Iceberg
outcome.had_late_corrections(); // whether any interval was below the boundary

For a service landing MSCONS or iMSys batches continuously, that boundary read is per batch — one catalogue load per 96 values if a batch is one meter-day. hot_writer reads it once:

let writer = store.hot_writer().await?;
for batch in stream {
    writer.append(&batch).await?;
}

The boundary can move under a write

Routing is decided against a boundary read before the write, and archival advances that boundary in Iceberg — a different system, sharing no transaction with the insert. Detaching a partition before archiving it closes most of the gap, since an insert into one being archived fails outright. What is left is the moment after the drop, when the partition is recreated and the insert lands in a range the watermark has since claimed for the cold tier. The row is not lost; it is invisible, which is worse.

So append reads the boundary again after writing and routes a second time if it moved. Both writes are idempotent — the hot tier’s primary key, the cold tier’s reconciliation — so the second pass restores rather than duplicates, and the outcome carries both rounds.

Why reusing a boundary is safe in hot_writer

The writer refuses any interval below the boundary it was opened at, rather than routing it. That is the whole safety argument: routing on a stale boundary could place a row below the true watermark; refusing cannot. A refusal names the interval and points at append.

A snapshot can only be stale in one direction, because the watermark is monotonic. The margin against that is the settlement lag — archival never closes a window newer than now - settlement_lag, a week by default, while current data has from near now. A writer held for the length of an ingest run is nowhere near that margin; one held for days is, so reopen per run rather than caching one for the process lifetime.

That margin is thinnest in two states: a backfill, whose from is old rather than current, and a catch-up, where archival advances several windows in one run. Use append for both — the second boundary read is what it is for.

This appends; it never overwrites

A correction is a new row at a higher version, not an update. That is MSCONS’s own rule, and it is why the resolved readings relation applies latest-version-wins over the raw readings_versions one. A caller expecting update semantics gets an append, and the prior value stays readable — which is what makes a past settlement reproducible.

Redelivery is ordinary traffic

Every transport worth deploying delivers at least once: Kafka redelivers after a failed offset commit, a webhook retries on a timeout, an operator replays a file. A store that errored on redelivery could not be driven by one.

A row already stored at the same (merge key, version) is skipped, in both tiers — reported as Duplicate, never appended twice.

A cold duplicate is not merely untidy. Resolution ranks one row per version, so two rows at one version are two winners — and resolution is elided entirely for a historical scan whose files provably hold a single version, which is exactly the shape a duplicated late correction produces. The sum is overstated with nothing reporting it.

And a version identifies one assertion. The same (malo_id, obis_code, from, version) must always carry the same value. Redelivering an identical row is fine; restating a different value under an existing version means a producer is wrong, and silently keeping either copy would bury that. Both tiers refuse it. A corrected value needs a higher version — which is the whole mechanism.

Rows go as arrays expanded with unnest, one statement per batch rather than one per reading. At 96 values per meter per day, per-row round trips would be the write path’s entire cost.

A delivery that states no version

Not every reading arrives with an MSCONS label — an SMGW push, a manual entry, a CSV backfill. Resolution still has to order them, so Version::arrival derives one from when the delivery was recorded:

ScopedVersion::new(scope, Version::arrival(recorded_at)?)

Unix milliseconds, and the unit is the whole of the decision. MSCONS labels are ≥ 14 digits; a millisecond timestamp is 13 and stays 13 until November 2286, so an arrival-derived version always sorts below a stated one and a late delivery that finally carries its own version wins. Microseconds would be 16 digits — inside the band, silently outranking the network operator. Seconds would be below it but not sub-second, so two writes in one second would collide on (merge key, version).

is_well_formed() stays false for these, which is what distinguishes “the operator said so” from “we assigned one”.

Values the operator authors

Two kinds of write, and only one of them is a delivery:

Through append, an authored value that a higher version already beats is stored and quietly shadowed — audited, confirmed, never current. That is the ordinary case for Ersatzwertbildung: the FAULTY interval being replaced arrived under a real MSCONS version, so the version you can put on your own value is lower than the one you have to beat.

let outcome = store.append_authoritative(&[ersatzwert]).await?;
// Every displacement is Inserted or Superseded. Anything else was retried.
for d in outcome.displacements {
    audit.record(d.malo_id, d.from, d.written.version);   // the version it landed at
}

The version each row carries is a floor, not an assertion. Where a higher one already holds the reading, the store re-appends at ScopedVersion::next of the one actually in force — continuing the stored sequence under the stored scope, because a version is comparable only within its own and the hot tier refuses a second network operator for one reading.

That decision comes from Displacement::superseded, which the write observed inside its own transaction — a read-then-write cannot say that, which is why the loop belongs here rather than in every caller.

Re-asserting a value already in force is a no-op, not a version bump. After AUTHORITATIVE_ATTEMPTS rounds it gives up: another writer is authoring the same reading continuously, which is a conflict for an operator rather than something to retry through.

Knowing what a write displaced

A count cannot distinguish a new reading from a correction from a backfill that changed nothing. A caller building a correction audit trail needs that distinction, and the obvious way to get it — read the prior state, then write — is a race, wrong exactly when two corrections arrive together, which is when an audit trail is worth having.

So the store reports it:

for d in outcome.displacements {
    if d.effect.changed_current_value() {
        audit.record(d.malo_id, d.from, d.superseded, d.written);
    }
}

Shadowed is the one a naive design loses. Backfilling an older delivery after a newer one has arrived is legitimate, joins the audit trail, and changes nothing any query returns — a caller treating every accepted write as a change would report a correction that never took effect.

Quality and unit travel with the value. A substitute replaced by a measured reading is a change even when the number is identical, and a value without a unit is dimensionless.

displacements covers both tiers. The hot tier takes the prior state and the insert in one transaction; the cold tier has no transaction a reader can join, so it holds an exclusive cold-append lease across the read and the write instead — two processes appending the same late correction serialise rather than both finding no stored row.

The loser waits, with a backoff budget of about 14 seconds, and then gives up with a retryable Error::LockTimeout having changed nothing. The budget is sized for the operation rather than for the common case: the Iceberg commit is a compare-and-swap that retries under contention, and the Parquet write before it is sized by the delivery. A bulk correction run over an archived month is both the largest write and the one time several writers queue, so it is the case the budget has to cover.

Zählerstandsgänge

A Lastgang is energy over [from, to). A Zählerstandsgang is a cumulative register value at an instant — and since BK6-24-174 (in force 06.06.2025) a German MSB holds one per measuring point at the same cadence as the Lastgang it is differenced into. The primary record is therefore exactly as voluminous as the derived one, and § 146 Abs. 4 AO means it cannot be discarded after differencing: a stored difference cannot reproduce the register values it came from.

Left in plain PostgreSQL it is the one table nobody may delete from and nothing tiers out of. So a table declares which shape it holds:

let config = TableConfig::new("meter_reads_versions")
    .time_model(TimeModel::Point)
    .build()?;

store.append_readings(&[zaehlerstandsgang]).await?;      // metering::MeterReading
let back = store.readings(malo)?.melo(melo)?             // ...and back
    .range(from, to).collect().await?;

Everything else is identical — the same attribute and subject columns, the same version scoping, the same watermark, the same partitioning, the same routed writes and late corrections. All of that reads the start timestamp, and a reading has one.

Four things differ:

A register belongs to a meter, not to a market location

A Marktlokation may be measured by more than one Messlokation, and both meters carry 1-0:1.8.0 at the same instants. Keyed on the market location alone the second reads as a restatement of the first — and where the two agree, which two freshly installed meters do, one is dropped.

So a point table puts melo_id in the merge key by default. The column is NOT NULL there and a delivery naming no Messlokation is refused:

StoredReadings::new(malo, obis, Sparte::Strom, readings, source, version, recorded_at)
    .with_melo_id(melo)          // required on a point table

identify_by_melo pins it either way. See the storage model.

They are never the same table. value would mean two things in one column, and summing Zählerstände gives a number with no meaning that looks exactly like a consumption total. append_readings is refused on an interval table and append on a point one, each naming the other. to IS NULL is the row-level signal, so an external engine holding only the Parquet can tell them apart too.

A zero-width interval would make MeterInterval a lie — metering computes demand_kw as energy over duration — and leaves value meaning two things anyway.

Partitions appear when you need them

A partitioned table rejects a row with no partition to hold it. Both write paths ensure the partitions for the range they are about to write, so a fresh deployment or a backfill reaching past the pre-created headroom does not get a bare no partition of relation found for row.

Creation is serialised by an advisory lock, because the first batch of a new day has every ingest worker creating the same partition at once — the ordinary topology. The fast path is unaffected: an existing partition costs one catalogue lookup, and hot_writer caches the ones it has already confirmed.

Writing through your own driver

It means reproducing the hot schema, its primary key, the canonical-OBIS CHECK, the Sparte/unit/quality code lists and the intra-version overlap exclusion — a contract carried in prose, where drift shows up as readings that silently fail to supersede.

StoredSeries makes that contract compiler-checked instead. The one rule the type cannot carry is the tier routing, which is what append and hot_writer enforce.