| D-M2-01-1 |
2026-10-12 |
No migration seeds event_consumer_cursors. In the roles worker and all, ConsumerService.Ensure inserts the row of every registered consumer in one unit of work before the runner starts, with INSERT ... SELECT $1, COALESCE(max(seq), 0) FROM event_log ON CONFLICT (name) DO NOTHING; unregistered names have no row and no lag series. The column default 0 of 03 stays. Place: 00007_event_consumer_cursors.sql, queries/consumers.sql, 02 11.5 item 5. |
A seeded row for a consumer that arrives in a later release (push, M3-19) would make it replay every event retained since the migration; a new consumer starts at the head like the first connection of an agent consumer (04 15.1 item 2). Reading max(seq) in one statement is safe because a reader that sees seq = k sees every smaller number (04 3.2 item 3); two workers starting together run the same statement and one row wins. |
| D-M2-01-2 |
2026-10-12 |
domain.PlanRound: numbers contiguous from last_seq + 1 are processed; when last_seq + 1 lies below min(seq) and the first number read equals min(seq), the events in between were removed by retention, the round processes from min(seq), logs event consumer lost events to retention once at error level with bothy.consumer.name and bothy.event.seq and continues; every other jump, hole or repeat is ErrConsumerGap, the round rolls back and the consumer stalls. min(seq) is read in the round's transaction only when the first number does not follow the cursor. Place: domain/consumer.go, app/consumers.go, 02 11.5 item 6. |
The removed events cannot come back and stalling would not help; a webhook subscription detects its own gap through last_delivered_seq (11 6), and the alarm fires long before. 04 3.2 items 3 and 4 rule out other holes, so they are bugs; unlike the tail's ErrGap the worker does not exit, the cursor just never passes the hole. The round holds ACCESS SHARE on event_log, so no partition is detached between the reads. |
| D-M2-01-3 |
2026-10-12 |
A handler error, a handler panic (recovered into kind internal, the panic value is not kept), a database error and ErrConsumerGap roll back the whole round: no job, no cursor move, the next attempt at the next wake or poll; no event is ever skipped. The runner logs the first failure after a success at error level (event consumer round failed, bothy.consumer.name, error.type as SQLSTATE, consumer_gap, timeout, canceled or the kernel/errs kind) and the next success at info level (event consumer recovered). Each consumer has its own goroutine, so a stalled one never holds back another. Place: app/consumer_runner.go, 02 11.5 item 7. |
One failing event must stall only its consumer, visibly through the gauge and the alarm, rather than lose an event; edge-triggered lines keep a database outage, during which the worker pauses (02 14.4), from flooding the log. |
| D-M2-01-4 |
2026-10-12 |
Before the cursor moves, the round asks the appender of sync/app (its per-unit txn.Key buffer, bufferedEvents) whether the unit buffered durable events; if so it fails with kind internal and the unit rolls back. Place: app/appender.go, app/consumers.go, 02 11.5 item 8. |
The consumer would read its own output, and the round would take the event_seq counter lock and lengthen the serialized commit phase (02 11.7); ADR 0051 lets a round insert jobs only. |
| D-M2-01-5 |
2026-10-12 |
Two observable gauges, bothy.event.consumer.lag ({event}, head_seq - last_seq, floored at zero) and bothy.event.consumer.pending_age (s, now from kernel/clock minus occurred_at of the oldest event after the cursor, zero at the head), with the only attribute bothy.consumer.name from the closed set sync.ConsumerNames(); one callback calls Consumers.States (one read of all rows with the head, then one read per row, bounded by database.statement_timeout), and a failed read drops the points. Every role registers them in start step 6b. The registry internal/platform/otel/metrics of M1-68 is not merged, so they live in internal/sync/adapters/otelmetrics with the meter of the global provider; sync already had the otel vendor group. bothy.consumer.name is in otel/attr, the collector allowlist and signals.md 5.3. Place: adapters/otelmetrics/consumers.go, signals.md 4.2. |
The plan row asks for head_seq minus last_seq, but sync.event_retention is a time window, so the alarm needs the age too. A stopped or missing worker must still show a growing lag, so the api role exports it as well; telemetry never blocks (17 rule 4). |
| D-M2-01-6 |
2026-10-12 |
Alarm BothyEventConsumerLag, severity page, max by (bothy_consumer_name) (bothy_event_consumer_pending_age_seconds{job="bothy/bothy-server"}) > 900 for 5m, on the new key consumer.pending_alert_after (place Server, duration, default 15m, at least 1s). The start check consumer_alert_below_retention requires it strictly below sync.event_retention and names both keys and no value. The rule file holds the number; alert-thresholds keeps it equal to the default. Place: bothy-alerts.yaml, alerting.md 3.2 and 4, 02 13 item 4 and 22, configkeys_consumer.go, configchecks.go. |
A normal worker restart takes at most shutdown.total_timeout (60s) plus a few consumer.poll_interval (1s), far below 15m, and the for of the neighbouring lag alarm adds five minutes; 15m is far below the 30d retention, so the operator has weeks before any event is lost. page, because a standing consumer stops pushes and outgoing webhooks, which people notice at once (alerting.md 1). |
| D-M2-01-7 |
2026-10-12 |
One LISTEN event_log connection per process for all worker consumers, opened with pg.Pool.DialDedicated(ctx, "bothy-consumer-wake"), apart from the tail's connection; the listening loop of the M1-16 tail moved into the shared type listener in adapters/postgres, which both use. A notification wakes every consumer through a channel of capacity one with a non-blocking send; the payload is not read. The connection is reconnected with exponential backoff from consumer.poll_interval up to database.listen_reconnect_backoff_cap, and every consumer also polls every consumer.poll_interval. Without a registered consumer the runner starts nothing and opens no connection. Place: adapters/postgres/listen.go, waker.go, 02 11.5 item 9. |
In role all the ordered loop of the tail must never depend on the consumers; polling covers a lost connection and nothing is lost because the cursor lives in the database (02 14.4). The production list is empty in M2-01, and an idle connection per worker would serve nobody. |
| D-M2-01-8 |
2026-10-12 |
After a full batch (consumer.batch_size events) the next round of that consumer starts at once (drain); after a short batch, an idle round, a locked row or a failed round the consumer waits for the next wake or poll. Place: app/consumer_runner.go. |
A backlog drains without waiting a poll per batch, while consumer.batch_size keeps each round, and with it the ACCESS SHARE lock that a detach of the partition maintenance waits for, short. |
| D-M2-01-9 |
2026-10-12 |
The stop step event_consumers lies in lifecycle.PhaseConsumers (before River's phase) with budget 0, the rest of shutdown.total_timeout: the runner starts no new round and waits for the running ones; when the budget runs out the round's context is cancelled and the round rolls back. An unexpected end of the runner (the waker ends early, a panic outside a handler) goes through Failed() to lifecycle.Runner.FailOn, and the process exits 1 like after an unexpected stop of the River client. Place: wire/sync.go, wire/serve.go, 02 14.5, 15.2 and 15.3. |
02 15.3 gives this step no own budget; either way no job is lost or doubled, because cursor and jobs commit together. A process whose consumers ended would look healthy while pushes and webhooks stop. |
| D-M2-01-10 |
2026-10-12 |
Branch feat/M2-01-worker-event-consumers in its own worktree from the head of int/m1-wave6 (785916c); this session does not merge it locally and does not remove the worktree, the PR lane ports it. New text in the German chapters (02, alerting.md, signals.md, ADR 0051, threat model) is written in English. The tests take two hooks of wire.Deps: ConsumerHandlers (replaces the empty production list in worker and all) and ConsumerGauges (replaces the registration on the global meter provider), because otel.Setup installs one global provider per process and parallel serve tests would overwrite it. |
The launcher's task overrides the merge step of the brief ("Definition of done") and requires English output. The hooks follow the pattern of Deps.Registrations and let the serve tests prove that worker and all run a registered consumer and that role api exports the gauges without a runner. |
| D-M2-01-11 |
2026-10-13 |
Port onto main: the migration is 00009_event_consumer_cursors.sql (00007 and 00008 are M1-20 and M1-33), the start step identity of M1-33 stands in components() right after authz and before routes, held by TestStartSequenceModuleSteps, and closeTimeout lives only in listen.go, which the log tail and the waker share. |
M2-01 rebuilt the start sequence as components(); a step left in the old list would start in no role, and the device handlers must exist when the routes are mounted. The same renumbering as the integration (ADR 0007 rule 4). |