Add processor-liveness health check: faulted and stale control loops reported as Unhealthy

[VDBBjorn]

Problem

ControlLoop.IsFaulted is set at ControlLoop.cs:275 (sequential) and :417 (pipelined) when an unrecoverable exception escapes the middleware chain and the loop stops processing permanently. The flag is never read in production code — the only readers are tests (ControlLoopTests.cs:323/330, DiscoveredIssuesTests.cs:291/356/414, ControlLoopMiddlewareTests.cs). The only IHealthCheck in the library is ShardHealthCheck (src/Alberto.Dcb/Tenancy/ShardHealthCheck.cs:19), which reports shard database connectivity. A host whose control loops are all faulted reports Healthy to its load balancer and keeps receiving traffic while projections are silently frozen.

There is also no liveness signal for a loop that becomes wedged without throwing — a blocking downstream call that never returns, for example. IsFaulted alone cannot detect that. #71 bounded the shutdown drain for exactly that failure mode; this issue covers detecting it while the host is still running.

In production DI, ControlLoopGroup and LeaseAwareControlLoopGroup are registered as plain IHostedService (ControlLoopRegistration.cs:80), so there is no typed key a health check can resolve to enumerate loops directly.

Evidence

File	Line	What
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	39	IsFaulted property
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	275	IsFaulted = true when the sequential path faults
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	417	IsFaulted = true when the pipelined path faults
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	225, 236, 266	AlbertoMetrics.RecordProcessorLag in sequential RunAsync — natural heartbeat seam
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	329, 340, 364	AlbertoMetrics.RecordProcessorLag in RunPipelinedAsync
src/Alberto.Dcb/Tenancy/ShardHealthCheck.cs	18-19	The only IHealthCheck in the library; covers connectivity, not processor liveness. Carries [Experimental("ALB9001")]
src/Alberto.Dcb/ControlLoopRegistration.cs	80	services.AddSingleton<IHostedService> for the loop group — no typed key
src/Alberto.Dcb/Subscriptions/LeaseAwareControlLoopGroup.cs	118	await Task.WhenAll(stopTasks) over _runningLoops, before lease release
src/Alberto.Dcb/Subscriptions/RebuildableProjection.cs	131	IsRebuilding always returns true — applies to shadow loops only, never live loops
src/Alberto.Dcb/Configuration/ShardingRegistration.cs	33	services.Configure<HealthCheckServiceOptions> — the registration pattern to copy

Design

Observation model

Introduce ProcessorHealthState in src/Alberto.Dcb/Subscriptions/ as a thread-safe observation board — the direct parallel of ShardHealth in Tenancy/. Each ControlLoop writes to it; the health check reads from it. This keeps ControlLoop's constructor internal, avoids a registry, and matches the observation-not-admission-control philosophy already in ShardHealth.

[Experimental("ALB9002")]
public sealed class ProcessorHealthState
{
    private readonly ConcurrentDictionary<string, ProcessorSnapshot> _snapshots = new();

    internal void Report(string processorId, bool isFaulted, long lag, bool isRebuilding, DateTimeOffset at) =>
        _snapshots[processorId] =
            new ProcessorSnapshot(processorId, isFaulted, lag, isRebuilding, at, HasReported: true);

    public ProcessorSnapshot Get(string processorId) =>
        _snapshots.TryGetValue(processorId, out var s) ? s : ProcessorSnapshot.NotYetReported(processorId);

    public IReadOnlyCollection<string> ProcessorIds => [.. _snapshots.Keys];
}

[Experimental("ALB9002")]
public sealed record ProcessorSnapshot(
    string ProcessorId,
    bool IsFaulted,
    long Lag,
    bool IsRebuilding,
    DateTimeOffset LastHeartbeatAt,
    bool HasReported)
{
    public static ProcessorSnapshot NotYetReported(string processorId) =>
        new(processorId, IsFaulted: false, Lag: 0, IsRebuilding: false,
            DateTimeOffset.MinValue, HasReported: false);
}

ALB9002 is free — ALB9001 (sharding) is the only ALB9xxx code in the source tree today. Confirm the allocation before merging.

ControlLoop changes

ControlLoopAssembler.Create (ControlLoopAssembler.cs:101) gains a ProcessorHealthState? healthState = null parameter and threads it into the ControlLoop constructor. Note the signature already ends with two optional parameters (ILogger<ControlLoop>? logger = null, TimeSpan? drainTimeout = null, the latter added in #71) — append after them or use named arguments at the call sites; do not insert in the middle.

Note on TimeProvider: ControlLoop does not hold a TimeProvider. ControlLoopAssembler holds one (ControlLoopAssembler.cs:42) and uses it exclusively to build the retry/dead-letter middleware chain; Create does not pass it to the ControlLoop constructor. Heartbeat timestamps therefore require adding TimeProvider as a new constructor parameter to ControlLoop itself and threading it through Create — an internal constructor change, not a re-use of an existing parameter.

Alongside each RecordProcessorLag call:

_healthState?.Report(ProcessorId, isFaulted: false, lag, _processor.IsRebuilding, _timeProvider.GetUtcNow());

At each fault site (:275, :417), immediately after IsFaulted = true:

_healthState?.Report(ProcessorId, isFaulted: true, lag: 0, isRebuilding: false, _timeProvider.GetUtcNow());

Shadow rebuild path

DcbModuleBuilderExtensions.cs:638 builds a second, independent ControlLoopAssembler for shadow rebuild loops and calls assembler.Create() separately from ControlLoopRegistration. Any new healthState parameter must be handled there too — either passing null (shadow loops do not report) or passing the module's state so they report under their distinct {processorId}::rebuild::{version} key (format at RebuildableProjection.cs:73).

This matters for the verdict rules: IsRebuilding is hardcoded to true on RebuildableProjection (:131), so it only ever applies to shadow-loop entries, never live ones. A live loop's _processor.IsRebuilding is false during normal operation, including while a rebuild is running elsewhere. The health check's lag suppression for IsRebuilding == true therefore only fires for ::rebuild:: entries. The verdict rules must document this explicitly rather than imply the live loop gets slack during a rebuild.

Health check verdict rules

for each processorId in state.ProcessorIds:
    snapshot = state.Get(processorId)
    if not snapshot.HasReported:
        skip  // standby (lease not held) or host just started
    if snapshot.IsFaulted:
        → Unhealthy("Processor '{id}' faulted and stopped")
    if now - snapshot.LastHeartbeatAt > StalenessThreshold:
        → Unhealthy("Processor '{id}' last reported {elapsed} ago — possible wedge")
    if not snapshot.IsRebuilding and snapshot.Lag > DegradedLagThreshold:
        → Degraded("Processor '{id}' is {lag} events behind")

aggregate: any Unhealthy → Unhealthy; any Degraded only → Degraded; else Healthy

Why skip HasReported == false: a replica in LeaseAwareControlLoopGroup that did not acquire the lease for a processor never calls loop.StartAsync, so it never reports — normal standby behaviour. A processor that is registered but never reaches its first poll (a silent DI failure) also produces no entry; that ambiguity is in the open questions.

Defaults:

• StalenessThreshold: 5 × PollingInterval resolved at registration, with a TimeSpan.FromSeconds(5) floor. The default PollingInterval is 250 ms (ControlLoopOptions.cs:9), so the floor dominates by default. Do not confuse it with HeadRefreshInterval (100 ms, ControlLoopOptions.cs:15) — a different knob.
• DegradedLagThreshold: long.MaxValue (effectively disabled). Lag is too noisy standalone — a quiet store sits at 0 lag when healthy, and a restarting host catches up legitimately.

Registration

services.AddKeyedSingleton<ProcessorHealthState>(moduleKey);

registered in ControlLoopRegistration.Register before the assembler factory, with the health check wired via the same services.Configure<HealthCheckServiceOptions> pattern as ShardingRegistration.cs:33.

Unlike ShardHealthCheck — registered only for modules using WithTenancy(...).AcrossPostgresDatabases() — unconditional registration here would affect every AddAlberto call, including modules on InMemoryEventStoreBackend. The check is inert when AddHealthChecks() is absent, but it still adds a keyed singleton and wires reporting into every ControlLoop. See the open question.

autoRestart

Out of scope. The health check observes; restarting changes the IHostedService lifecycle. Leave room in ProcessorHealthCheckOptions for a future bool AutoRestart so adding it later is not a breaking change.

Public API impact

The public-API gate is armed — #70 removed all four RS00xx suppressions and captured a baseline of 2870 entries across 11 packages, with the rules at error. So ProcessorHealthState, ProcessorSnapshot, ProcessorHealthCheck, and ProcessorHealthCheckOptions need src/Alberto.Dcb/PublicAPI.Unshipped.txt entries in the same PR or the build fails with RS0016. Same for any new property on ControlLoopOptions / ControlLoopOverrides (DrainTimeout in #71 is the worked example — four entries, get/init on the option plus get/set on the override).

All the new public types carry [Experimental("ALB9002")], matching the sharding pattern. ProcessorHealthCheck can be public to allow subclassing and direct testing.

Affected files

• src/Alberto.Dcb/Subscriptions/ControlLoop.cs
• src/Alberto.Dcb/Subscriptions/ControlLoopAssembler.cs
• src/Alberto.Dcb/ControlLoopRegistration.cs
• src/Alberto.Dcb/DcbModuleBuilderExtensions.cs (shadow rebuild path, :638)
• src/Alberto.Dcb/Subscriptions/ProcessorHealthState.cs (new)
• src/Alberto.Dcb/Subscriptions/ProcessorHealthCheck.cs (new)
• src/Alberto.Dcb/Subscriptions/ProcessorHealthCheckOptions.cs (new)
• src/Alberto.Dcb/PublicAPI.Unshipped.txt
• docs/configuration.md (new options table rows)

Migration impact

None. ProcessorHealthState is purely in-memory. No SQL schema changes, no new migration files, no checkpoint table changes.

Test plan

1. ControlLoop_Reports_Faulted_To_HealthState_After_UnrecoverableError — force an exception past the middleware; assert IsFaulted == true and HasReported == true. InMemoryEventStoreBackend, no Testcontainers.
2. ControlLoop_Reports_Heartbeat_On_Each_Poll_Cycle — LastHeartbeatAt advances on every successful poll (drive with FakeTimeProvider).
3. ControlLoop_Reports_Lag_Zero_When_Caught_Up.
4. ProcessorHealthCheck_Returns_Unhealthy_For_Faulted_Processor.
5. ProcessorHealthCheck_Returns_Unhealthy_For_Silent_Processor — HasReported=true, LastHeartbeatAt older than the threshold.
6. ProcessorHealthCheck_Returns_Degraded_When_Lag_Exceeds_Threshold.
7. ProcessorHealthCheck_Skips_Rebuilding_Processors_For_Lag.
8. ProcessorHealthCheck_Returns_Healthy_For_Never_Reported_Processor — standby.
9. LeaseAwareControlLoopGroup_Standby_Loop_Never_Reports — loop is in _allLoops but the lease is not acquired.
10. ProcessorHealthCheck_Aggregates_Correctly — faulted + healthy → Unhealthy; degraded + healthy → Degraded.
11. ProcessorHealthCheck_Reports_Unhealthy_For_Wedged_Handler — a handler that ignores cancellation stops heartbeating; assert Unhealthy before shutdown is ever requested. This is the live-detection counterpart to the shutdown bound added in Bound the shutdown drain so a stuck handler cannot stall the host #71.

Risks

• Staleness false positives under load: a temporarily slow database can push one polling cycle past the threshold. Keeping StalenessThreshold a multiple of PollingInterval (not a flat value) gives modules with longer intervals proportional slack.
• At startup every processor is HasReported=false until its first completed poll. A probe in that window sees everything as healthy with no data — acceptable and consistent with ShardHealth's fallback, but it means a loop that faults before its first heartbeat is invisible.
• Adding TimeProvider to ControlLoop's internal constructor means coordinating both factory call sites (ControlLoopRegistration and the shadow rebuild path in DcbModuleBuilderExtensions) in the same PR — the same two-site coordination Bound the shutdown drain so a stuck handler cannot stall the host #71 needed.
• Unconditional registration adds a keyed singleton plus reporting overhead to every test that calls AddAlberto, including in-memory tests with no health check consumer.

Open questions

• Register unconditionally (like ShardHealthCheck, inert without AddHealthChecks()) or opt in via WithControlLoop(...)? Blast radius is wider here because processors exist on every module, not just sharded ones.
• Should shadow loops report under their ::rebuild:: key (making rebuild progress visible), or always receive null?
• Should the health check suppress ::rebuild:: entries entirely, or surface them with a distinct informational note?
• Should DegradedLagThreshold stay disabled by default with a concrete recommended value in the docs (e.g. 10 000 events), to avoid false Degraded alerts during post-restart catch-up?