From 0298a0b944ab10486c4bd80b5098ddcb09976d41 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 16 Aug 2026 04:48:55 -0400 Subject: [PATCH] chore(historian): batch-era defaults + appsettings/doc alignment --- CLAUDE.md | 5 ++- docs/Configuration.md | 18 ++++++++ docs/Historian.md | 41 +++++++++++++++++++ .../ZB.MOM.WW.OtOpcUa.Host/appsettings.json | 11 ++++- .../ContinuousHistorizationOptions.cs | 29 +++++++++++-- .../ContinuousHistorizationOptionsTests.cs | 20 +++++++++ 6 files changed, 117 insertions(+), 7 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 75610494..69af6317 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -934,10 +934,11 @@ only the recorder + outbox knobs: |---|---|---| | `Enabled` | `false` | `true` (with `ServerHistorian` configured) wires + spawns the recorder | | `OutboxPath` | `""` (required when enabled) | **Directory** holding the FasterLog segment + commit files. In production set an **absolute** path on durable storage | -| `CommitMode` | `PerEntry` | `PerEntry` = fsync before each append returns (no loss window); `Periodic` = batched commits every `CommitIntervalMs` | +| `CommitMode` | `PerEntry` | `PerEntry` = fsync before the append returns (no loss window); `Periodic` = batched commits every `CommitIntervalMs`. ⚠️ `PerEntry` now means **one fsync per ingest burst, not per value** — the recorder appends a burst through `AppendBatchAsync`, which enqueues every entry then issues a single group commit; the call still does not return until that commit lands | | `CommitIntervalMs` | `100` | Periodic-mode commit cadence; required `> 0` only under `Periodic` | -| `DrainBatchSize` | `64` | Entries peeked + written per drain pass | +| `DrainBatchSize` | `1024` | Entries peeked + written per drain pass. **Raised from 64**: the peek is an in-memory scan and the pass is bounded-parallel (not sequential), so a wide batch no longer costs `tags × latency` — while 64 drew down ≤ 64 entries per `DrainIntervalSeconds`, letting the outbox fall permanently behind during any gateway hiccup | | `DrainIntervalSeconds` | `2` | Steady drain cadence (and post-success reschedule) | +| `DrainConcurrency` | `8` | Historian tags one drain pass writes **concurrently** (a batch is grouped per tag; each group is its own gateway round-trip), so a sequential `tags × latency` pass becomes `⌈tags / concurrency⌉ × latency` with outstanding round-trips bounded. `1` = fully sequential. The ack is unchanged — all-or-nothing after every group reports, and the #538 quarantine still keys on "another group wrote this pass" | | `Capacity` | `500000` | Max un-acked outbox entries before drop-oldest; `0` = unbounded. Non-zero **by default since #538** — the disk backstop for a historian refusing *every* write, the one case the per-tag quarantine deliberately cannot resolve | | `MaxWriteAttempts` | `5` | #538: consecutive failed drain passes one tag may accumulate — counted **only across passes where other tags wrote successfully** — before its queued entries are abandoned so the FIFO head can advance | | `MinBackoffSeconds` | `1` | Initial retry backoff after a failed drain pass | diff --git a/docs/Configuration.md b/docs/Configuration.md index b83a8c30..c28def93 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -383,6 +383,24 @@ key must carry the scopes `historian:read`, `historian:write`, `historian:tags:w [`docs/Historian.md`](Historian.md) for the full key reference, the migration note (old Wonderware keys → gateway keys), and the deployment prerequisites. +#### `ContinuousHistorization` (recorder + outbox knobs) + +Only the knobs; the gateway connection comes from `ServerHistorian`. Full descriptions in +[`docs/Historian.md` §ContinuousHistorization configuration](Historian.md#continuoushistorization-configuration). + +| Key | Default | Meaning | +|---|---|---| +| `ContinuousHistorization:Enabled` | `false` | `true` **and** a configured `ServerHistorian` wires the FasterLog outbox and spawns the recorder. | +| `ContinuousHistorization:OutboxPath` | `""` | **Directory** for the FasterLog segment + commit files. Required when enabled; use an **absolute** path on durable storage. | +| `ContinuousHistorization:CommitMode` | `PerEntry` | `PerEntry` fsyncs before the append returns; `Periodic` commits on a timer every `CommitIntervalMs`. **`PerEntry` is one fsync per ingest *burst*, not per value** — the recorder appends a burst through `AppendBatchAsync`, which enqueues every entry and issues a single group commit. Durability is unchanged: the call does not return until that commit lands. | +| `ContinuousHistorization:CommitIntervalMs` | `100` | Periodic-mode cadence (ms); must be `> 0` under `Periodic`. | +| `ContinuousHistorization:DrainBatchSize` | `1024` | Entries peeked + written per drain pass. **Raised from 64**: the peek is an in-memory scan (widening it costs allocation, not I/O) and the pass is now bounded-parallel rather than sequential, so a wide batch no longer costs `tags × latency`. 64 drew down ≤ 64 entries per `DrainIntervalSeconds`, so any gateway hiccup left the outbox permanently behind, growing toward `Capacity` and shedding oldest samples. | +| `ContinuousHistorization:DrainIntervalSeconds` | `2` | Steady drain cadence (and post-success reschedule). | +| `ContinuousHistorization:DrainConcurrency` | `8` | Historian tags one drain pass writes **concurrently** — a batch is grouped per tag and each group is its own gateway round-trip, so this turns a sequential `tags × latency` pass into `⌈tags / concurrency⌉ × latency` while bounding outstanding round-trips. `1` = fully sequential; `< 1` is clamped to 1 with a `Validate()` warning. The ack is unchanged (all-or-nothing after every group reports), as is the #538 quarantine rule. | +| `ContinuousHistorization:Capacity` | `500000` | Max un-acked entries before drop-oldest; `0` = unbounded. The #538 disk backstop for a historian refusing **every** write — the one case the per-tag quarantine cannot resolve. | +| `ContinuousHistorization:MaxWriteAttempts` | `5` | #538 poison-tag budget: consecutive failed passes one tag may accumulate — **counted only across passes where *other* tags wrote successfully** — before its entries are abandoned so the FIFO head can advance. | +| `ContinuousHistorization:MinBackoffSeconds` / `MaxBackoffSeconds` | `1` / `30` | Initial and capped exponential retry backoff after a failed drain pass. | + ### `LocalDb` (pair-local config cache — driver role only) Driver-role nodes keep a consolidated **`ZB.MOM.WW.LocalDb`** SQLite database that caches the deployed diff --git a/docs/Historian.md b/docs/Historian.md index 1d360b87..02b0919a 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -142,6 +142,47 @@ source it from there. --- +## ContinuousHistorization configuration + +`ContinuousHistorizationOptions` binds the `ContinuousHistorization` section. It carries **only** the +recorder + outbox knobs — the gateway connection (endpoint / key / TLS) is sourced from +`ServerHistorian` (above), never restated here. + +| Key | Type | Default | Description | +|---|---|---|---| +| `Enabled` | bool | `false` | `true` **and** a configured `ServerHistorian` wires the FasterLog outbox + gateway value-writer and spawns the `ContinuousHistorizationRecorder`. `false` → no recorder; driver tag values are not historized. | +| `OutboxPath` | string | `""` | **Directory** holding the FasterLog segment + commit files. Required when `Enabled`. In production set an **absolute** path on durable storage — a relative path resolves against the host's working directory. | +| `CommitMode` | string | `PerEntry` | `PerEntry` fsyncs before the append returns (no loss window); `Periodic` batches commits onto a background timer every `CommitIntervalMs`. Parsed case-insensitively; an unrecognized value falls back to `PerEntry`. **See the batch-commit note below — `PerEntry` is one fsync per ingest *burst*, not per value.** | +| `CommitIntervalMs` | int | `100` | Periodic-mode commit cadence (ms); must be `> 0` under `Periodic`, ignored under `PerEntry`. | +| `DrainBatchSize` | int | `1024` | Outbox entries peeked + written per drain pass. **Raised from 64**: the peek is an in-memory scan of the outbox's live index (widening it costs allocation, not I/O), and a pass is now bounded-parallel over `DrainConcurrency` rather than sequential, so a wide batch no longer costs `tags × latency`. At 64 a drain drew down at most 64 entries every `DrainIntervalSeconds` — below the arrival rate of even a modest historized tag set — so any gateway hiccup left the outbox permanently behind, growing toward `Capacity` and shedding oldest samples instead of catching up. | +| `DrainIntervalSeconds` | double | `2` | Steady drain cadence (and the post-success reschedule). | +| `DrainConcurrency` | int | `8` | Historian tags a single drain pass writes **concurrently**. One peeked batch is grouped per historian tag and each group is an independent gateway round-trip, so a sequential pass cost `tags × latency`; this window makes it `⌈tags / concurrency⌉ × latency` while bounding how many round-trips can be outstanding against the gateway at once. `1` restores the fully sequential drain; a value `< 1` is clamped to 1 (with a `Validate()` warning). **The ack is unchanged** — still all-or-nothing after every group has reported, and the #538 per-tag quarantine still keys on "another group wrote this pass". | +| `Capacity` | int | `500000` | Max un-acked outbox entries before the drop-oldest policy sheds the front of the queue; `0` = unbounded. Roughly 50 MB of FasterLog. The #538 **disk backstop** for the one case the per-tag quarantine deliberately cannot resolve — a historian unreachable or refusing *every* write, where nothing proves any particular tag is at fault so the recorder correctly retains. | +| `MaxWriteAttempts` | int | `5` | #538 poison-tag budget: consecutive failed drain passes one historian tag may accumulate before its queued entries are abandoned and the FIFO head is allowed to advance. **Counted only across passes where *other* tags wrote successfully**, so an ordinary gateway outage never advances it. Clamped to at least 1; each abandonment logs a Warning naming the tag. | +| `MinBackoffSeconds` | double | `1` | Initial retry backoff after a failed drain pass. | +| `MaxBackoffSeconds` | double | `30` | Cap on the exponential retry backoff. | + +> **`PerEntry` now means one fsync per ingest *burst*, not per value.** The recorder appends a whole +> value burst through `IHistorizationOutbox.AppendBatchAsync`, which enqueues every entry and then +> issues a **single** group commit — so a burst of N values costs one fsync rather than N. The +> durability contract is unchanged: the call does not return until that commit completes, so nothing +> the caller has been told is buffered can be lost to an unclean restart. (The documented residual is +> unchanged too: a throw *before* the commit leaves the enqueued prefix tracked-but-uncommitted — +> served correctly for the life of the process, absent after an unclean restart — exactly the +> enqueue-then-commit window the single-entry path always had, widened from one entry to the batch.) + +> **Cross-repo follow-up — a multi-tag batch `WriteLiveValues` on the gateway contract.** The drain's +> unit of work is one gateway round-trip **per historian tag**, because +> `ZB.MOM.WW.HistorianGateway.Client` **0.3.0** exposes no RPC that writes values for more than one tag +> in a call. Bounded parallelism (`DrainConcurrency`) is therefore the ceiling: it overlaps the +> round-trips but cannot remove them, so a wide tag set still pays `⌈tags / concurrency⌉` network +> hops per drain pass. Collapsing that to a single hop needs a **batch/multi-tag `WriteLiveValues`** on +> the `historian_gateway.v1` contract in the `~/Desktop/HistorianGateway` sister repo. Until that +> lands, raising `DrainBatchSize` widens each hop's payload but not the hop count — which is why the +> concurrency window, not the batch width, is what governs drain throughput today. + +--- + ## A poison tag no longer pins the outbox (#538) `DrainOnceAsync` peeks a batch, writes **every** per-tag group in it, and then acks **all-or-nothing** — diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/appsettings.json b/src/Server/ZB.MOM.WW.OtOpcUa.Host/appsettings.json index a1392094..d40730a8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/appsettings.json +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/appsettings.json @@ -43,10 +43,17 @@ "OutboxPath": "", "_OutboxPathComment": "Directory holding the FasterLog segment + commit files. Required when Enabled=true. In production set an ABSOLUTE path on durable storage.", "CommitMode": "PerEntry", + "_CommitModeComment": "PerEntry means one fsync per ingest BURST, not per value: the recorder appends a whole burst through AppendBatchAsync, which enqueues every entry and issues a single group commit. Periodic batches commits onto a timer every CommitIntervalMs (bounded loss window).", "CommitIntervalMs": 100, - "DrainBatchSize": 64, + "DrainBatchSize": 1024, + "_DrainBatchSizeComment": "Entries peeked + written per drain pass. Was 64 when a pass wrote its per-tag groups sequentially; the peek is an in-memory scan and the pass is now bounded-parallel over DrainConcurrency, so a wide batch is cheap — and 64 let the outbox fall permanently behind during any gateway hiccup.", "DrainIntervalSeconds": 2, - "Capacity": 0, + "DrainConcurrency": 8, + "_DrainConcurrencyComment": "Historian tags a single drain pass writes concurrently (one gateway round-trip per tag group). 1 restores the fully sequential drain. The ack is unchanged: still all-or-nothing after every group reports.", + "Capacity": 500000, + "_CapacityComment": "Max un-acked outbox entries before drop-oldest; 0 = unbounded. The Gitea #538 disk backstop for a historian refusing EVERY write — the one case the per-tag quarantine cannot resolve.", + "MaxWriteAttempts": 5, + "_MaxWriteAttemptsComment": "Gitea #538: consecutive failed drain passes one tag may accumulate — counted only across passes where OTHER tags wrote successfully — before its queued entries are abandoned so the FIFO head can advance.", "MinBackoffSeconds": 1, "MaxBackoffSeconds": 30 }, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs index 25fc806f..a690835b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs @@ -43,6 +43,15 @@ public sealed class ContinuousHistorizationOptions /// every ms (higher throughput, a bounded worst-case loss window). /// Parsed case-insensitively against HistorizationCommitMode; an unrecognized value falls /// back to PerEntry. + /// + /// PerEntry now means one fsync per ingest burst, not per value. The + /// recorder appends a whole value burst through + /// IHistorizationOutbox.AppendBatchAsync, which enqueues every entry and then issues a + /// single group commit — so a burst of N values costs one fsync instead of N. The + /// durability contract is unchanged: AppendBatchAsync does not return until that + /// commit completes, so nothing the caller has been told is buffered can be lost to an + /// unclean restart. + /// /// public string CommitMode { get; init; } = "PerEntry"; @@ -50,9 +59,23 @@ public sealed class ContinuousHistorizationOptions /// is Periodic. Ignored under PerEntry. public int CommitIntervalMs { get; init; } = 100; - /// Maximum outbox entries peeked + written per drain pass; clamped to a positive value by - /// the recorder (a non-positive value falls back to 64). - public int DrainBatchSize { get; init; } = 64; + /// + /// Maximum outbox entries peeked + written per drain pass; clamped to a positive value by the + /// recorder (a non-positive value falls back to the default). + /// + /// Defaults to 1024. It was 64 when a pass wrote its per-tag groups sequentially, so + /// a wide batch cost tags × latency on the drain thread. Both halves of that reasoning + /// are gone: the peek is an in-memory scan of the outbox's live index (widening it costs + /// allocation, not I/O), and the pass is now bounded-parallel over + /// , so its cost is ⌈tags / concurrency⌉ × latency + /// regardless of batch width. At 64, a drain draws down at most 64 entries every + /// — below the arrival rate of even a modest historized + /// tag set — so any gateway hiccup left the outbox permanently behind, growing toward + /// and shedding oldest samples, rather than catching up once the + /// gateway recovered. + /// + /// + public int DrainBatchSize { get; init; } = 1024; /// Steady drain cadence in seconds (also the post-success reschedule). Defaults to 2. public double DrainIntervalSeconds { get; init; } = 2; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationOptionsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationOptionsTests.cs index 58c64674..b3d3e7a3 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationOptionsTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationOptionsTests.cs @@ -36,6 +36,26 @@ public sealed class ContinuousHistorizationOptionsTests }.Validate(), m => m.Contains("CommitIntervalMs")); + /// + /// Pins the shipped defaults, which are a tuned set rather than arbitrary numbers: the drain is + /// bounded-parallel over an in-memory peek (so a wide DrainBatchSize is cheap and a narrow + /// one lets the outbox fill during a gateway hiccup), Capacity is the #538 disk backstop, + /// and MaxWriteAttempts is the #538 poison-tag budget. The shipped + /// appsettings.json restates these values, so a silent drift between the two is exactly + /// what this test exists to catch. + /// + [Fact] + public void Defaults_are_the_documented_batch_era_values() + { + var options = new ContinuousHistorizationOptions(); + + Assert.Equal("PerEntry", options.CommitMode); + Assert.Equal(1024, options.DrainBatchSize); + Assert.Equal(8, options.DrainConcurrency); + Assert.Equal(500_000, options.Capacity); + Assert.Equal(5, options.MaxWriteAttempts); + } + /// A fully-configured enabled recorder produces no warnings. [Fact] public void Valid_config_is_clean()