Persist processor stop-with-reason durably on alberto_processor_checkpoints

[VDBBjorn]

Problem

When ControlLoop faults it sets IsFaulted = true, logs a Critical message, and returns. Nothing durable records the cause. After a pod restart the reason is gone: an operator must mine structured logs by timestamp, correlate pod names, and hope log retention reaches back far enough.

ControlLoop is a public sealed class (ControlLoop.cs:17) with public bool IsFaulted { get; private set; } (ControlLoop.cs:39) and an internal constructor (ControlLoop.cs:42). In production DI the instance is wrapped in the internal ControlLoopGroup and registered only as IHostedService (ControlLoopRegistration.cs:80), so external callers cannot conveniently read IsFaulted without a cast. No production code reads it at all today — the only readers are tests (ControlLoopTests.cs:323/330, DiscoveredIssuesTests.cs:291/356/414, ControlLoopMiddlewareTests.cs).

The pipelined path compounds the problem. The worker catch block (ControlLoop.cs:471-481) has evt.GlobalPosition, evt.EventType.Id, and evt.TenantId in scope when it calls reportFailure(ex) (:479), but ReportFailure accepts only Action<Exception> (ControlLoop.cs:428), discarding all event context before the Critical log fires in the finally block (ControlLoop.cs:415-423).

In the sequential path the outer catch that sets IsFaulted = true (ControlLoop.cs:275) sits outside the foreach over relevantEvents, so the event position is no longer in scope at fault time.

The result: after any restart, operators cannot distinguish an infrastructure failure from an event-processing bug, cannot identify which event caused the stop, and cannot verify that the processor recovered cleanly.

Scope clarification

Dead-letter entries (DeadLetterEntry.cs) already capture GlobalPosition, EventType, TenantId, ErrorMessage, and StackTrace for every event that flows through the middleware chain and exhausts retries. That covers the per-event story. What is missing is the processor-level stopped-with-reason record: one durable row saying "this processor is halted, stopped at time T because of exception E, while processing the event at position P (if known)". The two records are complementary, not redundant.

Evidence

File	Line	What
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	39	IsFaulted property
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	275	IsFaulted = true in sequential outer catch — event position out of scope at this level
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	417	IsFaulted = true from pipelineFailure in the pipelined finally block
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	428	void ReportFailure(Exception failure) — event context is not carried through
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	430	Interlocked.CompareExchange — only the first failure wins
src/Alberto.Dcb/Subscriptions/ControlLoop.cs	479	reportFailure(ex) in the worker catch, with full event context in scope
src/Alberto.Dcb/Subscriptions/DeadLetterEntry.cs	36	Dead-letter captures position, type, tenant, error, stack — middleware chain only
src/Alberto.Dcb.Postgres/PostgresCheckpointStore.cs	39	SaveAsync — columns from Migrations/001_InitialSchema.sql (processor_id, last_position, updated_at) and Migrations/021_CheckpointFenceTokens.sql (fence_token); no fault columns
src/Alberto.Dcb/Subscriptions/CachingCheckpointStore.cs	156-165	RewindAsync writes directly to _inner and clears the cache — the bypass pattern for durable immediate writes

Proposal

Context available at fault time

Sequential path (RunAsync outer catch, :275): ProcessorId and the Exception are available; the event position is not in scope (the evt variable lives inside the foreach, inside the inner try). Capturing it needs long? _lastDispatchedPosition, string? _lastDispatchedEventType, and string? _lastDispatchedTenantId tracking fields set inside DispatchAsync before dispatching, then read in the catch.

Pipelined path (worker catch, :471-481): the worker has full event context in scope. Carrying it through means changing ReportFailure from Action<Exception> to Action<Exception, EventFaultContext?>:

internal readonly record struct EventFaultContext(
    long Position,
    string EventType,
    string? TenantId);

pipelineFailure becomes a pair (Exception, EventFaultContext?) — null context for infrastructure failures from the producer loop, populated for a worker failure.

Storage: new nullable columns on alberto_processor_checkpoints

-- Migrations/035_ProcessorFaultRecord.sql (multi-tenant)
-- Migrations/SingleTenant/035_ProcessorFaultRecord.sql (single-tenant)
ALTER TABLE $schema_prefix$alberto_processor_checkpoints
    ADD COLUMN IF NOT EXISTS faulted_at        TIMESTAMPTZ DEFAULT NULL,
    ADD COLUMN IF NOT EXISTS fault_message     TEXT        DEFAULT NULL,
    ADD COLUMN IF NOT EXISTS fault_stack_trace TEXT        DEFAULT NULL,
    ADD COLUMN IF NOT EXISTS fault_position    BIGINT      DEFAULT NULL,
    ADD COLUMN IF NOT EXISTS fault_event_type  TEXT        DEFAULT NULL,
    ADD COLUMN IF NOT EXISTS fault_tenant_id   TEXT        DEFAULT NULL;

All six are NULL when the processor is healthy or has never faulted. Non-NULL faulted_at is the canonical stopped signal.

A write path upserts only the fault columns, leaving last_position, fence_token, and updated_at untouched. The ON CONFLICT DO UPDATE arm is correct for the normal case (a checkpoint row already exists). The INSERT arm for the no-existing-row case must not use last_position = 0: if a processor faults on its very first run before saving any checkpoint, inserting a row at position 0 makes a restart re-stream from 0. Use INSERT ... ON CONFLICT DO NOTHING plus a separate UPDATE ... WHERE processor_id = @processor_id in one function, or guard the INSERT so it fires only when a row already exists.

A separate clear path NULLs all six columns on recovery.

IProcessorFaultStore

internal interface IProcessorFaultStore
{
    Task RecordFaultAsync(string processorId, ProcessorFaultRecord fault, CancellationToken ct = default);
    Task ClearFaultAsync(string processorId, CancellationToken ct = default);
}

internal sealed record ProcessorFaultRecord(
    string Message,
    string? StackTrace,
    long? Position,
    string? EventType,
    string? TenantId);

Internal to Alberto.Dcb, reachable from Alberto.Dcb.InMemory and Alberto.Dcb.Postgres through the existing InternalsVisibleTo grants — no new grants needed.

• PostgresCheckpointStore implements it; writes go straight to the database, bypassing CachingCheckpointStore's flush buffer (same pattern as RewindAsync), since fault records must be immediately durable.
• InMemoryCheckpointStore implements it with a Dictionary<string, ProcessorFaultRecord?> under the existing lock.
• CachingCheckpointStore passes fault calls straight through to _inner (_inner is IProcessorFaultStore), bypassing the dirty buffer, and clears the fault on ResetAsync (:145) and RewindAsync (:156) via the same passthrough.

ControlLoop changes

_faultStore = checkpointStore is IProcessorFaultStore f ? f : null;

Sequential outer catch (:275):

IsFaulted = true;
if (_faultStore is not null)
    await _faultStore.RecordFaultAsync(ProcessorId,
        new ProcessorFaultRecord(ex.Message, ex.StackTrace,
            _lastDispatchedPosition, _lastDispatchedEventType, _lastDispatchedTenantId),
        CancellationToken.None);

Pipelined finally block (:415-423), after the Critical log:

if (pipelineFailure is not null && _faultStore is not null)
    await _faultStore.RecordFaultAsync(ProcessorId,
        new ProcessorFaultRecord(
            pipelineFailure.Value.Ex.Message,
            pipelineFailure.Value.Ex.StackTrace,
            pipelineFailure.Value.Ctx?.Position,
            pipelineFailure.Value.Ctx?.EventType,
            pipelineFailure.Value.Ctx?.TenantId),
        CancellationToken.None);

The tracking fields are written only by RunAsync and read only in its own outer catch, so there is no cross-path race — the pipelined path uses the pipelineFailure pair and never touches them.

Ordering note (post-#71): the pipelined finally block now also runs the bounded-drain logic added in #71 (ControlLoop.cs:400-413) and may hand an abandoned worker set to DisposeWhenDrainedAsync. The fault write must sit after SaveWatermarkCheckpointAsync and must not be gated on the drain succeeding — a processor that both faulted and failed to drain is exactly the case an operator most needs a durable record for.

Recovery / clear: the loop exits immediately after recording the fault — there is no next iteration in the same instance. ClearFaultAsync therefore belongs in the new ControlLoop created on the next host start, fired after the first successful SaveAsync proves the processor can advance. ResetAsync and RewindAsync also clear via the store passthrough.

Admin surface

CheckpointInfo lives at src/Alberto.Dcb.Admin/AdminTypes.cs:17:

public sealed record CheckpointInfo(string ProcessorId, long LastPosition, DateTimeOffset? UpdatedAt);

PostgresAdminDataAccess moved to its own project in #70 — it is now src/Alberto.Dcb.Postgres.Admin/PostgresAdminDataAccess.cs, and it constructs CheckpointInfo positionally at :103 (GetCheckpointsAsync) and :129 (GetSingleCheckpointAsync). Both sites need updating.

The CLI consumer is alberto status, not alberto processors list — tools/Alberto.Cli/Commands/StatusCommand.cs:59-64 projects ProcessorId, LastPosition, and UpdatedAt. That is where FaultedAt / FaultMessage should surface when non-null.

Note that both Alberto.Dcb.Admin and Alberto.Dcb.Postgres.Admin were parked (IsPackable=false) in #70, and every alberto ops mutation now routes through IAdminOperator. This work extends a parked surface — worth confirming that is intended before starting.

Migration impact

Two new files:

• src/Alberto.Dcb.Postgres/Migrations/035_ProcessorFaultRecord.sql (multi-tenant)
• src/Alberto.Dcb.Postgres/Migrations/SingleTenant/035_ProcessorFaultRecord.sql (single-tenant)

Both sets currently top out at 034_OutboxRetentionIndex.sql, so 035 is the next number for each. (The SingleTenant set is missing 012 — it jumps 011 → 013 — but that does not affect the next allocation.) The Migrations/Catalog/ set holds only alberto_tenant_shards and needs no change.

Non-breaking: all six columns are nullable with DEFAULT NULL, existing rows unaffected, and the GREATEST-guarded SaveAsync upsert uses an explicit column list that does not touch them. A zero-downtime rolling deploy is safe.

Public API impact

The public-API gate is armed. #70 removed the root .editorconfig suppressions, publicapi-silence.globalconfig, the project-local src/Alberto.Dcb/.editorconfig, and the <NoWarn> group; RS00xx is now at error, with a captured baseline of 2870 entries across 11 packages.

Alberto.Dcb.Admin has both PublicAPI.Shipped.txt and PublicAPI.Unshipped.txt (327 entries). CheckpointInfo and its positional constructor are baselined at PublicAPI.Unshipped.txt:149-156:

Alberto.Dcb.Admin.CheckpointInfo.CheckpointInfo(string! ProcessorId, long LastPosition, System.DateTimeOffset? UpdatedAt) -> void

Adding parameters to that record therefore breaks the build (RS0016 for the new members, RS0017 for the now-stale constructor entry) until the file is updated in the same PR. This is a build-time requirement now, not something deferred to a v1 gate.

IProcessorFaultStore and ProcessorFaultRecord are internal — no entries needed for Alberto.Dcb, Alberto.Dcb.Postgres, or Alberto.Dcb.InMemory. Alberto.Dcb.Postgres.Admin has no PublicAPI files of its own.

Affected files

• src/Alberto.Dcb/Subscriptions/ControlLoop.cs
• src/Alberto.Dcb/Subscriptions/CachingCheckpointStore.cs
• src/Alberto.Dcb.Postgres/PostgresCheckpointStore.cs
• src/Alberto.Dcb.InMemory/InMemoryCheckpointStore.cs
• src/Alberto.Dcb.Admin/AdminTypes.cs
• src/Alberto.Dcb.Admin/IAdminReader.cs
• src/Alberto.Dcb.Admin/PublicAPI.Unshipped.txt
• src/Alberto.Dcb.Postgres.Admin/PostgresAdminDataAccess.cs
• tools/Alberto.Cli/Commands/StatusCommand.cs
• src/Alberto.Dcb.Postgres/Migrations/035_ProcessorFaultRecord.sql
• src/Alberto.Dcb.Postgres/Migrations/SingleTenant/035_ProcessorFaultRecord.sql

Test plan

1. ControlLoopFaultsAndRecordsReason_RunAsync — force an exception past the middleware on a loop backed by an InMemoryCheckpointStore implementing IProcessorFaultStore; assert the fault message is recorded. No Testcontainers needed.
2. ControlLoopFaultsAndRecordsEventContext_RunPipelinedAsync — MaxConcurrency=2; inject a worker exception; assert fault_position and fault_event_type match the offending event.
3. ControlLoopClearsFaultOnRecovery — record a fault, create a new loop, let one successful checkpoint save complete, assert all fault fields are NULL.
4. ControlLoopFault_InfrastructureError_NoEventContext — make StreamAllAsync throw before any dispatch; assert fault_position is NULL and the message is not.
5. ControlLoopFault_RecordedEvenWhenDrainIsAbandoned — pipelined loop with both a faulting worker and a worker that ignores cancellation; assert the fault row is written despite the abandoned drain (guards the Bound the shutdown drain so a stuck handler cannot stall the host #71 interaction).
6. InMemoryCheckpointStore_RecordFault_RoundTrip — record, read back, clear, assert null.
7. CachingCheckpointStore_FaultWriteBypassesFlushBuffer — assert the inner store receives the write immediately.
8. AdminReader_GetCheckpoints_IncludesFaultFields — after a fault, both read methods return non-null FaultedAt / FaultMessage. (Testcontainers.)
9. Migration035_IsIdempotent — run twice on a live database; no error, columns exist with the right nullability. (Testcontainers.)

Risks

• CheckpointInfo is a positional record with a baselined constructor entry. Adding parameters is source-breaking for positional callers and an RS0017 build error until PublicAPI.Unshipped.txt is updated. Decide up front: stay positional (update the two call sites plus the baseline) or add a separate GetProcessorFaultAsync to IAdminReader and leave CheckpointInfo alone.
• The fault write uses CancellationToken.None (the loop is already exiting). If the database is unreachable — plausibly the very cause of the fault — RecordFaultAsync hangs until the Npgsql connection timeout. Give the fault write its own short command timeout. Note this now compounds with the drain budget from Bound the shutdown drain so a stuck handler cannot stall the host #71: a hung fault write inside the finally block eats into host shutdown.
• The pipelined path records only the first failure's event context (Interlocked.CompareExchange, ControlLoop.cs:430). Concurrent worker failures lose the second context. Dead letters already cover the multi-event case.
• Both admin projects are parked (IsPackable=false) as of Park the admin surface, audit every CLI mutation, expire delivered outbox entries #70. Extending them is fine, but confirm the surface is meant to grow rather than stay frozen.

Open questions

• Clear the fault record on the first successful SaveAsync of the new run, or only on an explicit operator reset/rewind? Automatic clearing is ergonomic but silently wipes evidence of a transient blip that self-healed.
• Truncate fault_stack_trace to a fixed length (e.g. 4 000 chars, matching the dead-letter schema) rather than storing it in full? Deep stacks can run to hundreds of KB per row.
• Extend CheckpointInfo, or add GetProcessorFaultAsync to IAdminReader to keep the checkpoint contract and its PublicAPI baseline stable?