diff --git a/.gitignore b/.gitignore index 23cc90c..fb1b19a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ +.claude/settings.json +.ressouces /simulator /gearman_publisher /rabbitmq_publisher diff --git a/CLAUDE.md b/CLAUDE.md index 83a61db..6798764 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -23,7 +23,7 @@ Every queue is wired end-to-end in `internal/queue/registry.go`'s `NewRouter` - - **Done** — State Changes (`statusngin_statechanges`) - **Done** — Core Restarts (`statusngin_core_restart`) - **Done** — Downtimes (`statusngin_downtimes`) - full ADD/LOAD/START/STOP/DELETE lifecycle across the `scheduleddowntimes`/`downtimehistory` table pairs; doesn't use the BulkInserter abstraction (see `.claude/specs/downtime_ablauf.txt` for the processing matrix and why) -- **Done** — Prometheus metrics exporter (`internal/metrics`, served on its own port, default `:9105/metrics`). **Every labeled series is pre-created at zero on startup**, so a fresh worker exports all 12 `queue_name`, 18 `table` (14 of them also on `db_flushes_total`) and 4 `component` series instead of growing them as traffic arrives - otherwise a panel reads "No data" exactly while everything is healthy, and an alert on it never evaluates. `metrics.InitQueue`/`InitTable`/`InitFlushes` do that; the calls sit where the label values are actually known (`db.NewBulkInserter` for its own table, `queue.NewRouter` for the router's keys and the four downtime tables that bypass BulkInserter, the metrics package's own `init` for the components), so a new inserter or queue is covered without a second list to keep in sync. `InitFlushes` is deliberately *not* called for the four downtime tables: they write one row per statement without batching, and a permanent 0 denominator under a climbing `db_events_written_total` would make the rows-per-flush ratio read `+Inf` rather than "not applicable". Only safe because all these label sets are small and fixed - never do this for an open-ended label like a client id or hostname. +- **Done** — Prometheus metrics exporter (`internal/metrics`, served on its own port, default `:9105/metrics`). **Every labeled series is pre-created at zero on startup**, so a fresh worker exports all 12 `queue_name`, 18 `table` (14 of them also on `db_flushes_total`) and 4 `component` series instead of growing them as traffic arrives - otherwise a panel reads "No data" exactly while everything is healthy, and an alert on it never evaluates. `metrics.InitQueue`/`InitTable`/`InitFlushes` do that; the calls sit where the label values are actually known (`db.NewBulkInserter` for its own table, `queue.NewRouter` for the router's keys and the four downtime tables that bypass BulkInserter, the metrics package's own `init` for the components), so a new inserter or queue is covered without a second list to keep in sync. `InitFlushes` is deliberately *not* called for the four downtime tables: they write one row per statement without batching, and a permanent 0 denominator under a climbing `db_events_written_total` would make the rows-per-flush ratio read `+Inf` rather than "not applicable". Only safe because all these label sets are small and fixed - never do this for an open-ended label like a client id or hostname. **`queue_connected` is the one deliberate exception to all of this and must stay out of `InitQueue`**: a pre-created gauge sits at 0, and 0 there is not "nothing has happened yet" but "this queue has no connection" - the exact state its alert fires on - so pre-creating it would report an outage on every worker between `NewRouter` and the first dial. Each consumer sets it once a connection is genuinely up (`Ready` for Gearman, `connect` for RabbitMQ); `TestInitQueueLeavesQueueConnectedAlone` fails if someone "fixes" the omission. - **Done** — Data retention (`internal/cleanup` + `cmd/db_cleanup`) - a separate one-shot binary for cron/systemd timers, not part of the worker process ## Core Architecture Rules @@ -37,6 +37,9 @@ Every queue is wired end-to-end in `internal/queue/registry.go`'s `NewRouter` - - **Every RabbitMQ queue is declared durable, and the events in it stay transient.** Those are two different properties and conflating them is the whole point of this entry. `durable=true` is not a preference: a queue that is neither durable nor exclusive is RabbitMQ's deprecated `transient_nonexcl_queues` feature, and on 4.x it is not merely warned about but **`denied_by_default`** - measured against 4.3.5, where every `queue.declare` is refused with a connection exception, so the worker declares nothing, consumes nothing, and 5 of 8 RabbitMQ tests fail. `durable` has worked since AMQP 0-9-1, which makes it the only setting that works on *every* broker version, and the CI matrix now runs the suite against `3-alpine` and `4-alpine` so the next such break is red rather than a line in a container log. What durable does **not** do is put events on disk: durability stores the queue *definition*, while a message is only written durably if its publisher marked it persistent, and neither the NEB broker (`amqp_basic_publish(..., properties=nullptr, ...)` in `src/MessageHandler/RabbitmqClient.cpp`) nor `cmd/rabbitmq_publisher` sets `delivery_mode`. So the queues still buffer in RAM while no worker is connected and a broker restart still empties them - verified end to end: 5 messages in a durable queue, `docker restart`, queue present and **0 messages**. Measured cost of the change: 20,000 events published in 0.22-0.23s either way, three runs each. The one thing that is *not* free is the migration - the queues already exist as non-durable, and AMQP answers a redeclare with different arguments with a **406 PRECONDITION_FAILED** rather than adapting, so both sides must be stopped and the queues deleted once. Which is also why the declaration must stay identical in all three places that make one (`internal/queue/rabbitmq.go`, `cmd/rabbitmq_publisher`, and `declareTestQueue` in the tests) **and** agree with the broker's `DurableQueues`, whose default was changed to `true` in the same breath. `STATUSENGINE_TEST_RABBITMQ_URL` points the suite at another broker, which is what makes "does it work on 4" a command instead of a source edit. - **`Stop` must close those connections in parallel** (`closeWorkers`). `Close` drains in-flight jobs before dropping the connection, bounded by `DrainTimeout` (30s); closing twelve one after another would make the worst case six minutes, long past the point a service manager escalates to SIGKILL - which would lose exactly the buffered rows the graceful shutdown exists to flush. `TestCloseWorkersDrainsInParallel` measures it against real stuck handlers, because that is the only state in which `Close` actually spends its timeout: 0.30s parallel vs 1.80s sequential at six workers. - **RabbitMQ's `Stop` cancels its consumers before it closes anything, which is the guarantee the Gearman fork already got.** `basic.cancel`, then wait for the delivery loops, then close the channels and the connection - the reverse of the order it had, where a Handler still running when `Stop` arrived finished, wrote its rows, and then acked into a channel that was already gone: `operation basic.ack caused a connection exception channel_error: "expected 'channel.open'"` at the broker, `rabbitmq: ack failed` here, and a message redelivered on the next start although it had been processed. Measured across three `-race` suite runs against a RabbitMQ 4.3.5 container: **7 failed acknowledgements here and 6 `channel_error` lines at the broker before, 0 and 0 after**. The two counts differ because the frame is often refused client-side before it is ever written, which is also why a run without `-race` can show neither - the client-side `rabbitmq: ack failed` line is the reliable observable, the broker log is the confirmation. It was only ever harmless because rule 6's writes are idempotent, which is why it survived so long - but it made the two backends promise different things for no reason. **The cancel also changes what happens to deliveries the broker has already pushed**, and that is the part to know before touching this: closing a channel makes amqp091-go *drop* them (`consumers.close`, "closed before drained, drop in-flight"), while cancelling hands them to the loop first (`consumers.cancel` closes only the input, and `buffer` empties its slice before closing the output). So a `Stop` under load now works through up to `-rabbitmq-prefetch` further messages per queue instead of leaving them to be redelivered - measured at 0.10s -> 2.61s for `TestRabbitMQConsumerStopUnderLoadClosesOutputSafely` (100 buffered x 25ms of Handler). That is bounded by `stopDrainTimeout` (5s) after a `cancelTimeout` (2s), so the RabbitMQ shutdown worst case is 22s against Gearman's 45s, and the cancels go out in parallel for the same reason `closeWorkers` does. Two details are load-bearing: the consumer tag had to become **explicit** (`statusengine-worker.`), because with an empty one the library invents a tag and never returns it, leaving nothing to cancel; and the channels are closed **even when the cancel timed out**, which is what still ends a loop parked on a delivery channel the broker never closed - `superviseReconnects` closes `out` only after those loops exit, so that fallback is what keeps the shutdown from hanging rather than being tidiness. +- **A lost Gearman connection is rebuilt per queue, because the library does not do it and does not look like it isn't.** On EOF `agent.work` hands a `*WorkerDisconnectError` to the `ErrorHandler` and returns; nothing sends on `worker.in` afterwards, and `Work()` is a plain `for range` over that channel which only `Close()` ends. So the worker stayed up, kept emitting its 30-second stats line, and consumed **nothing, permanently** - measured in production as twelve `error=EOF` WARN lines followed by `processed=12836` repeating unchanged until the process was restarted by hand. That is the worst shape a bug can have here: everything else looks healthy. `superviseReconnects` in `internal/queue/gearman.go` is one goroutine per queue that closes the dead generation and rebuilds the worker (`New`/`AddServer`/`AddFunc`/`Ready`/`Work`), retrying every `reconnectDelay` (2s, shared with RabbitMQ) until it succeeds or `Stop` is called. **Do not "simplify" this into `WorkerDisconnectError.Reconnect()`** - `agent.disconnect_error` calls the `ErrorHandler` while holding the agent's mutex and `reconnect()` takes that same mutex, so reconnecting where the disconnect is reported is a self-deadlock; and `reconnect()` additionally reuses the dead agent in place, overwriting `a.conn` without closing the old socket, taking the agent mutex before the worker mutex (the reverse of `AddFunc` -> `broadcast` -> `agent.Write`), and calling `agentWG.Add` on a WaitGroup `Close` may be waiting on. Rebuilding touches none of that and disposes of the socket via `Close`. Two properties are load-bearing: the `ErrorHandler`'s send onto `queueWorker.lost` is **non-blocking** (it runs under that agent mutex, so blocking there deadlocks the reconnect it is asking for), and `Stop` closes `c.stopping` **under `c.mu`**, which is the entire serialisation against a reconnect landing mid-shutdown - either the new worker gets into Stop's snapshot or the supervisor sees the shutdown and closes it itself, and without that a fresh connection survives `Stop` and feeds already-flushed BulkInserters. `TestGearmanConsumerReconnectsAfterConnectionDrop` publishes a job *after* severing the connection through a TCP proxy (gearman-go keeps its socket in an unexported field of an unexported agent, so there is nothing a test can close directly), and `TestGearmanConsumerStopsWhileReconnecting` pins that an unbounded retry loop never outlasts a shutdown. Jobs in flight at the drop lose their acknowledgement and come back - rule 6's upserts are what make that a non-event. **Watch `statusengine_queue_connected{queue_name=…}`**: it is the only series that separates "stopped consuming" from "idle", since both leave `messages_received_total` flat and `jobs_in_flight` at 0, and it is deliberately the one per-queue series *not* pre-created (a pre-created gauge sits at 0, which here reads as the outage it is meant to detect). Knowingly not covered: a half-open TCP connection produces no EOF at all, so the agent stays parked in `read()` - out of scope while gearmand is on `127.0.0.1`, and closing it would need an application-level heartbeat. +- **`statusngin_downtimes` is processed one message at a time, and no other queue is.** A downtime arrives as up to four separate jobs over its life - ADD, START, STOP, DELETE - and only ADD is an UPSERT: START and STOP are bare `UPDATE ... WHERE ` against the row ADD created, and STOP/DELETE are `DELETE FROM scheduleddowntimes`. Each of those is a silent no-op when it runs before the message it depends on, and `execDowntimeAction` used to ignore `RowsAffected`, so nothing failed and nothing was logged. With eight handlers per queue that ordering only holds by luck, and live traffic supplies the luck - a downtime's events are minutes apart. A backlog does not: after a two-hour job-server outage was drained in one burst, **6 of ~14 downtimes were corrupted**, each one explainable by the order its jobs happened to run in. START before ADD left `was_started=0, actual_start_time=0` on a row whose `actual_end_time` was correct (STOP got there after ADD); STOP before START left a `scheduleddowntimes` row behind for good, because STOP deleted it and START's UPSERT put it back. Found only by `cmd/db_verifier` diffing against the legacy PHP worker. `RequiresInOrderProcessing` (`registry.go`) now caps that queue at `gearman.New(1)` regardless of `-gearman-max-concurrent-jobs-per-queue`; reproduced end to end both ways - 8 lifecycles published in one burst give 8/8 correct rows and 0 leftovers with the cap, and without it the exact production signature comes back (one row at `ws=0, ast=0, aet=2400`, three leftover scheduled rows). **The cap costs nothing**: that queue sees a handful of messages an hour, and per-queue concurrency buys no throughput anyway (rule 2 - the bottleneck is the one `Run` goroutine per table). The legacy PHP worker forks one process per queue and the RabbitMQ consumer calls its Handler synchronously from a plain `for range`, so both were already in order; **only the Gearman path ever wasn't, and never deliberately**. Do not extend `inOrderQueues` casually - `TestInOrderQueuesAreExactlyTheOnesThatNeedIt` demands the same justification for any addition, since every other queue carries self-contained events that stand on their own. The two status queues come closest (an older snapshot could overwrite a newer one) but each is complete in itself and the next check interval repairs it, where a downtime's lost START is never re-sent by anything. +- **`statusengine_queue_downtime_updates_unmatched_total` is the backstop, and it is deliberately not a simple `RowsAffected != 0` check.** MySQL counts rows *changed*, not matched (`CLIENT_FOUND_ROWS` is off in `go-sql-driver`, and turning it on to answer this one question would change what every other statement reports), so a redelivered START rewriting the values it already wrote reports zero exactly like a START whose row does not exist - and redelivery is normal here (rule 6). A zero therefore triggers one follow-up `SELECT` (`db.DowntimeHistoryExistsQuery`) to tell the two apart, on a path that should never be taken. **DELETE is excluded on purpose**: STOP removes the `scheduleddowntimes` row and the DELETE that follows finds it already gone, so a zero-row DELETE happens on *every* ordinary downtime and counting it would bury the signal under one increment each. Honest limit, measured rather than assumed: if something writes the row between the UPDATE and the lookup, a real loss reads as a redelivery - against a deliberately unserialized queue the corruption appeared in MySQL while this counter stayed at 0. That is fine because the queue is serialized and a second worker process is advised against anyway (rule 2), but it means this counts a **missing ADD**, not a broken ordering guarantee. One legitimate source remains: a downtime whose ADD predates the database, since LOAD is a no-op by design in both workers. - **An empty Router is rejected by `Start`.** The connections are opened inside the per-queue loop, so with no queues `Start` would return success having connected to nothing and consumed nothing forever. Before the split, a single connection was opened up front and that case at least failed on an unreachable server. - **gearmand 2.0.0** (released 2026-07-22, previous release 1.1.22) changes **no wire protocol or packet format**, so `mikespook/gearman-go` stays compatible and needs no adaptation. Its three breaking changes are scheduling behaviour: round-robin job assignment is now the default and can no longer be disabled, job priority is now global rather than per function, and the client-side task list is FIFO instead of LIFO (that one affects publishers, not this worker). Round-robin changes only *which* pending job the server offers, and the Go client just sends `GRAB_JOB_UNIQ` and takes what it gets - so on the old single-connection shape it would have spread the shared budget around and softened the starvation above without removing it, since the coupling was in-process. With one connection per queue it is genuinely neutral, and correctness no longer depends on a server flag this worker can neither verify nor enforce (`-R` is off by default on gearmand 1.x). diff --git a/README.md b/README.md index 1b54682..f548cd3 100644 --- a/README.md +++ b/README.md @@ -285,6 +285,47 @@ Two consequences worth knowing: gearmand's `--round-robin` is a related but separate thing: it changes which queue the *server* offers next, and would have spread the shared budget around without removing the coupling, which lived in this process. It is also off by default on gearmand 1.x. Correctness here no longer depends on it. +## Reconnecting to the broker + +Neither client library recovers from a dropped connection on its own, so both consumers rebuild it themselves, retrying every 2 s until the broker is back. + +For Gearman that was not always true, and the failure was a quiet one. On EOF the library's agent goroutine reports a `WorkerDisconnectError` to the `ErrorHandler` and returns; nothing sends on the worker's internal job channel afterwards, and `Work()` is a plain `range` over that channel, which only `Close()` ever ends. So the worker stayed up, kept logging its stats line, and consumed nothing — one WARN line per queue and then silence until someone restarted the process: + +``` +level=WARN msg="gearman: worker error" queue=statusngin_hoststatus error=EOF +... +level=INFO msg="gearman: consumer stats" addr=127.0.0.1:4730 processed=12836 errors=0 +level=INFO msg="gearman: consumer stats" addr=127.0.0.1:4730 processed=12836 errors=0 +``` + +The consumer now treats that error as what it is. Per queue, it closes the dead connection, then rebuilds the worker from scratch — `New`/`AddServer`/`AddFunc`/`Ready`/`Work`, the same path every startup takes — rather than calling the library's `WorkerDisconnectError.Reconnect()`, which cannot be called from the `ErrorHandler` that hands you the error: `agent.disconnect_error` holds the agent's mutex while calling the handler, and `reconnect()` takes that same mutex. The twelve queues reconnect in parallel, so a job server restart costs seconds rather than twelve backoffs in a row. + +Jobs that were in flight when the connection dropped lose their acknowledgement and are handed out again afterwards. That is expected and harmless — it is the same at-least-once redelivery the upserts under [MySQL Write Behavior](#mysql-write-behavior) exist to absorb. + +**Watch `statusengine_queue_connected`.** It is the only series that distinguishes a queue that has stopped consuming from one that is merely idle — both leave `messages_received_total` flat and `jobs_in_flight` at 0. It is deliberately *not* pre-created at startup, unlike every other per-queue series: a pre-created gauge sits at 0, and 0 here would claim an outage for all twelve queues in the window between wiring the Router and dialling. `statusengine_queue_reconnects_total` counts how often the link had to be rebuilt — a short dip is a broker restart, a climbing counter is a flapping link. + +One case is knowingly not covered: a half-open TCP connection (a network partition with no FIN or RST) never produces an EOF, so the agent stays blocked in `read()` and nothing notices. In production gearmand is on `127.0.0.1:4730`, where that does not realistically happen; closing it would need an application-level heartbeat. + +## Downtimes are processed in order + +`statusngin_downtimes` is the one queue whose messages build on one another, and the only one the consumer handles strictly one at a time. + +A single downtime arrives as up to four separate jobs — ADD, START, STOP, DELETE — and only ADD is an UPSERT. START and STOP are bare `UPDATE ... WHERE ` against the row ADD created; STOP and DELETE are `DELETE FROM …_scheduleddowntimes`. Every one of those does nothing at all, successfully and silently, if it runs before the message it depends on. + +With eight handlers per queue that ordering held only by luck, and live traffic supplied the luck — a downtime's events are minutes apart, so they never overlap. A backlog does not. After the job-server outage described above was drained in one burst, 6 of ~14 downtimes came out wrong: + +| Symptom in the Go database | What happened | +|---|---| +| `was_started=0`, `actual_start_time=0`, but `actual_end_time` correct | START ran before ADD — its UPDATE matched nothing. STOP ran after ADD and landed. | +| history correct, but a `scheduleddowntimes` row left behind forever | STOP ran before START — STOP deleted the scheduled row, START's UPSERT recreated it. | +| everything 0 *and* a leftover scheduled row | START and STOP both ran before ADD. | + +Nothing failed and nothing was logged; it was found by `cmd/db_verifier` diffing against the legacy PHP worker. The legacy worker forks one process per queue and the RabbitMQ consumer here calls its Handler synchronously from a plain `for range`, so both were already in order — only the Gearman path was not, and never on purpose. + +The consumer now caps that queue at one handler regardless of `-gearman-max-concurrent-jobs-per-queue`. It costs nothing: the queue sees a handful of messages an hour, and per-queue concurrency buys no throughput anyway (see [One connection per queue](#one-connection-per-queue) — the bottleneck is the single `Run` goroutine per table). + +**Watch `statusengine_queue_downtime_updates_unmatched_total`.** It counts downtime UPDATEs that found no row, which is what a lost ADD looks like, and it should stay at 0. It costs one extra `SELECT` when that happens rather than trusting `RowsAffected`, because MySQL counts rows *changed* rather than matched — so a redelivered START rewriting identical values reports zero too, and redelivery is normal here. A DELETE matching nothing is deliberately not counted: STOP already removed the scheduled row, so that happens on every ordinary downtime. + ## RabbitMQ Queue Durability Every queue is declared **durable**, and the events inside it stay **transient**. Those are two different AMQP properties, and keeping them apart is the whole point of this section. diff --git a/docs/openapi.yaml b/docs/openapi.yaml index a51a0fc..462ee1c 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1232,6 +1232,9 @@ paths: | `statusengine_queue_events_discarded_stale_total` | counter | `queue_name` | Status events dropped for being older than `status_max_age` (default 5m), before reaching MySQL or any WebSocket client. Only `statusngin_hoststatus` and `statusngin_servicestatus` ever appear here - they carry superseded snapshots, every other queue carries history. A burst after a restart is the feature working; a value that keeps climbing while the worker is up means the monitoring core's clock and this worker's disagree, and both queues are being discarded wholesale. | | `statusengine_queue_jobs_in_flight` | gauge | `queue_name` | Messages currently being handled, per queue. Labeled because each queue has its own Gearman connection and its own concurrency budget: a queue pinned at `-gearman-max-concurrent-jobs-per-queue` is that queue falling behind, which an unlabeled total cannot distinguish from twelve queues sharing the load. `sum()` without the label gives the old process-wide number. | | `statusengine_queue_handler_duration_seconds` | histogram | `queue_name` | Time to handle one message end to end: decode, WebSocket publish and enqueueing every decoded item for insertion. | + | `statusengine_queue_connected` | gauge | `queue_name` | 1 while the consumer holds a working connection for that queue, 0 from the moment it is lost until one is re-established. **Alert on this.** It is the only series that can tell a queue that has stopped consuming from a queue that is merely idle - both leave `messages_received_total` flat and `jobs_in_flight` at 0. Deliberately not pre-created at startup, unlike every other per-queue series: a pre-created 0 would claim an outage for all twelve queues in the window between wiring the Router and dialling. On RabbitMQ one connection carries every queue, so all twelve move together; on Gearman each queue has its own and they move independently. | + | `statusengine_queue_reconnects_total` | counter | `queue_name` | Times the consumer rebuilt a lost connection for that queue. `queue_connected` says whether data is flowing right now, this says how stable the link has been - the difference between one broker restart overnight and a link that flaps every few minutes. | + | `statusengine_queue_downtime_updates_unmatched_total` | counter | `table` | Downtime UPDATEs that found no row to update, per downtimehistory table. **Should stay at 0.** A downtime's START and STOP are written as `UPDATE ... WHERE ` against the row its ADD created, so no match means that ADD never arrived and the event is lost - silently, since MySQL reports success. Only the two `downtimehistory` tables appear: `scheduleddowntimes` is only ever upserted or deleted. DELETE is deliberately not counted, because STOP already removed the scheduled row and the DELETE that follows legitimately matches nothing on every ordinary downtime. Expect a few right after a fresh installation, from downtimes whose ADD predates the database (LOAD is a no-op by design, as in the legacy worker), and none afterwards. | | `statusengine_db_events_written_total` | counter | `table` | Rows successfully persisted per destination table. | | `statusengine_db_batch_flush_duration_seconds` | histogram | - | Duration of each bulk-insert flush to MySQL. | | `statusengine_db_flushes_total` | counter | `table` | Successful bulk-insert statements per table. Exists to be the denominator of `db_events_written_total`: `rate(db_events_written_total[1m]) / rate(db_flushes_total[1m])` is the average rows per statement, per table, using only counters - `db_batch_size_at_flush` has the same information but is a histogram and carries no `table` label. Counted in the same branch as the row counter, so both always describe the same set of statements; a failed flush lands in `pipeline_errors_total{component="mysql"}` instead. The four downtime tables have no series here - they write one row per statement without batching, and a 0 denominator would make the ratio read +Inf rather than "not applicable". | @@ -1272,6 +1275,26 @@ paths: series below being present and can treat a missing one as the worker being down rather than merely idle. + `statusengine_queue_connected` is the one deliberate exception, and + for the same reason the rule exists: a pre-created gauge sits at 0, + and 0 on that one is not "nothing has happened yet" but "this queue + has no connection" - the exact state its alert fires on. It appears + once the consumer has actually connected, which for a healthy + worker is within the first moments of startup. + + ### Is the worker still connected? + + `queue_connected` at 0 is the answer, and it is the only one there + is: a consumer that has lost its broker leaves every other series + looking exactly like an idle one - `messages_received_total` flat, + `jobs_in_flight` at 0, no errors. Neither client library recovers on + its own, so both consumers rebuild the connection themselves, every + 2s until it comes back; `queue_reconnects_total` counts how often + that was needed. A gauge that goes to 0 and returns within seconds + is a broker restart. One that stays at 0 means the broker is still + gone and the backlog is accumulating there - which is recoverable, + unlike a worker that has quietly stopped consuming. + ### Is the worker keeping up? Three of these answer that together. `queue_jobs_in_flight` for a @@ -1309,6 +1332,18 @@ paths: # TYPE statusengine_queue_jobs_in_flight gauge statusengine_queue_jobs_in_flight{queue_name="statusngin_hoststatus"} 0 statusengine_queue_jobs_in_flight{queue_name="statusngin_servicestatus"} 3 + # HELP statusengine_queue_connected 1 while the consumer holds a working connection for this queue, 0 while it does not. + # TYPE statusengine_queue_connected gauge + statusengine_queue_connected{queue_name="statusngin_hoststatus"} 1 + statusengine_queue_connected{queue_name="statusngin_servicestatus"} 1 + # HELP statusengine_queue_reconnects_total Total number of times the consumer re-established a lost connection, per queue. + # TYPE statusengine_queue_reconnects_total counter + statusengine_queue_reconnects_total{queue_name="statusngin_hoststatus"} 0 + statusengine_queue_reconnects_total{queue_name="statusngin_servicestatus"} 0 + # HELP statusengine_queue_downtime_updates_unmatched_total Total number of downtime UPDATE statements that matched no row, per table. + # TYPE statusengine_queue_downtime_updates_unmatched_total counter + statusengine_queue_downtime_updates_unmatched_total{table="statusengine_host_downtimehistory"} 0 + statusengine_queue_downtime_updates_unmatched_total{table="statusengine_service_downtimehistory"} 0 # HELP statusengine_queue_handler_duration_seconds Duration of handling one message, per queue. # TYPE statusengine_queue_handler_duration_seconds histogram statusengine_queue_handler_duration_seconds_bucket{queue_name="statusngin_hoststatus",le="0.01"} 2 diff --git a/internal/db/downtime.go b/internal/db/downtime.go index 0107b73..54eb228 100644 --- a/internal/db/downtime.go +++ b/internal/db/downtime.go @@ -190,6 +190,25 @@ func UpdateDowntimeHistoryStoppedQuery(row DowntimeRow) (string, []any) { return query, args } +// DowntimeHistoryExistsQuery builds the SELECT that answers whether the +// downtimehistory row an UPDATE was aimed at exists at all. +// +// It is only ever run to disambiguate an UPDATE that reported zero affected +// rows, which MySQL says for two very different situations: the row is not +// there (the downtime's ADD is missing - a lost event), or the row is there +// and already held exactly these values (a redelivered message rewriting +// what it wrote before, which CLAUDE.md rule 6 makes a normal occurrence). +// go-sql-driver reports rows *changed* rather than rows matched, since +// CLIENT_FOUND_ROWS is off by default - and turning it on to tell these +// apart would change what every other statement in the worker reports, to +// answer a question that arises on a path that should never be taken. +func DowntimeHistoryExistsQuery(row DowntimeRow) (string, []any) { + table := downtimeTable("downtimehistory", row.IsHostDowntime) + where, whereArgs := downtimePrimaryKeyWhere(row) + + return "SELECT 1 FROM " + table + " WHERE " + where + " LIMIT 1", whereArgs +} + // DeleteDowntimeHistoryQuery builds the DELETE used only for the DELETE // Envelope.Type action's "wasNeverStarted" case (downtime_ablauf.txt // section 5): a downtime removed before its scheduled start_time was ever diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 1fd7a62..451d39d 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -48,15 +48,20 @@ var Components = []string{ComponentMySQL, ComponentWebSocket, ComponentGraphite, // id, a hostname), where pre-creating would be indistinguishable from a // cardinality leak. -// InitQueue pre-creates the three per-queue series for queueName, so a -// worker that has not yet received a message on that queue still reports -// zeros for it rather than nothing at all. Called from queue.NewRouter -// for every queue it wires up. +// InitQueue pre-creates the per-queue series for queueName, so a worker +// that has not yet received a message on that queue still reports zeros +// for it rather than nothing at all. Called from queue.NewRouter for every +// queue it wires up. +// +// QueueConnected is deliberately absent: it is the one series here whose +// zero value is a claim rather than an absence of data, so the consumers +// set it when a connection actually comes up. See its own comment. func InitQueue(queueName string) { QueueMessagesReceivedTotal.WithLabelValues(queueName) QueuePayloadsRepairedTotal.WithLabelValues(queueName) QueueHandlerDurationSeconds.WithLabelValues(queueName) QueueJobsInFlight.WithLabelValues(queueName) + QueueReconnectsTotal.WithLabelValues(queueName) } // InitStaleDiscards pre-creates the per-queue series on @@ -106,6 +111,19 @@ var ( CommandRejectReasons = []string{"auth", "malformed", "unknown_command", "denied", "too_large"} ) +// InitDowntimeUpdates pre-creates the per-table series on +// DowntimeUpdatesUnmatchedTotal. Called from queue.NewRouter for the two +// downtimehistory tables, which are the only ones a downtime UPDATE ever +// targets - the scheduleddowntimes pair is only ever upserted or deleted. +// +// Pre-created for the usual reason, which bites harder here than anywhere +// else: this counter is supposed to sit at 0 forever, so without this it +// would not exist at all on a healthy worker and an alert on it would never +// evaluate. +func InitDowntimeUpdates(table string) { + DowntimeUpdatesUnmatchedTotal.WithLabelValues(table) +} + // InitTable pre-creates the per-table series on DBEventsWrittenTotal. // Called from db.NewBulkInserter, so every table this worker can write to // is covered automatically - including one added later, which is the @@ -174,6 +192,79 @@ var ( Help: "Number of queue messages currently being handled, per queue.", }, []string{"queue_name"}) + // QueueConnected is 1 while the consumer holds a working connection + // for that queue and 0 from the moment it is lost until one is + // re-established. This is the metric to alert on, and it exists + // because nothing else could answer the question: a queue that has + // stopped consuming and a queue that is merely idle produce exactly + // the same flat messages_received_total and the same + // jobs_in_flight of 0. When the Gearman consumer lost its connections + // and never reconnected, the only evidence was twelve WARN lines in + // the log and a processed count that stopped climbing. + // + // Deliberately NOT pre-created by InitQueue, unlike every other series + // in this subsystem. Each consumer sets it to 1 itself after a + // connection is actually up (Ready for Gearman, connect for + // RabbitMQ), so a 1 means "connected" rather than "NewRouter ran". + // That is the opposite choice from DBAvailable, which is pre-set to 1 + // precisely because there is nothing to observe before the first flush + // - here there is, and it is the whole point of the metric. + // + // One series per queue on both backends, even though a RabbitMQ + // connection is shared by every queue and so moves all twelve at once: + // an alert written against one backend has to keep working on the + // other. + QueueConnected = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "statusengine", + Subsystem: "queue", + Name: "connected", + Help: "1 while the consumer holds a working connection for this queue, 0 while it does not.", + }, []string{"queue_name"}) + + // QueueReconnectsTotal counts how often the consumer had to rebuild a + // lost connection for that queue. QueueConnected says whether data is + // flowing right now; this says how unstable the link has been, which is + // the difference between "the broker restarted once at 03:00" and "the + // link flaps every few minutes". Pre-created at zero by InitQueue, + // because a counter that only appears once something has gone wrong + // cannot be graphed before it does. + QueueReconnectsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "statusengine", + Subsystem: "queue", + Name: "reconnects_total", + Help: "Total number of times the consumer re-established a lost connection, per queue.", + }, []string{"queue_name"}) + + // DowntimeUpdatesUnmatchedTotal counts downtime UPDATEs that matched no + // row, per destination table. A downtime's START and STOP arrive as + // separate messages and are written as bare UPDATE ... WHERE + // against the row its ADD created, so an UPDATE that matches nothing + // means that row was not there - the event is simply gone, with no + // error and nothing in the log to say so. + // + // That is not hypothetical: handling the downtime queue concurrently + // let a backlog execute START before ADD, which left was_started=0 and + // actual_start_time=0 on downtimes that had demonstrably run. The + // consumer now serializes that queue (queue.RequiresInOrderProcessing), + // so this should stay at 0; it exists because the failure it reports is + // otherwise invisible until someone diffs the database against another + // worker. + // + // Deliberately not counted for DELETE, which legitimately matches + // nothing on every ordinary downtime: STOP already removed the + // scheduleddowntimes row by the time DELETE arrives. + // + // One legitimate source remains: a downtime whose ADD predates this + // database entirely (LOAD is a no-op by design, mirroring the legacy + // worker), so its START finds no history row. Expect a few of these + // right after a fresh installation, and none afterwards. + DowntimeUpdatesUnmatchedTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: "statusengine", + Subsystem: "queue", + Name: "downtime_updates_unmatched_total", + Help: "Total number of downtime UPDATE statements that matched no row, per table.", + }, []string{"table"}) + // QueuePayloadsRepairedTotal counts payloads that were not valid // UTF-8 and had their invalid bytes reinterpreted as Windows-1252 // before decoding (see repairUTF8 in internal/queue). A non-zero diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index c606a89..732ceb4 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -99,6 +99,7 @@ var queueMetricNames = []string{ "statusengine_queue_payloads_repaired_total", "statusengine_queue_handler_duration_seconds", "statusengine_queue_jobs_in_flight", + "statusengine_queue_reconnects_total", } // TestComponentSeriesExistAtZero pins the package's init: all four @@ -128,6 +129,29 @@ func TestInitQueueCreatesEverySeries(t *testing.T) { } } +// TestInitQueueLeavesQueueConnectedAlone is the deliberate exception to +// every other pre-creation rule in this package, and it needs a guard +// precisely because it contradicts them: adding QueueConnected to InitQueue +// would look like fixing an oversight. +// +// A pre-created gauge sits at 0, and 0 on this one is not "nothing has +// happened yet" - it is the assertion "this queue has no connection", the +// exact state the alert fires on. Since NewRouter runs before any consumer +// dials, pre-creating it would make every worker report an outage for all +// twelve queues during startup, and a worker that failed to start at all +// would be indistinguishable from one that was running fine. The consumers +// set it themselves once a connection is genuinely up. +func TestInitQueueLeavesQueueConnectedAlone(t *testing.T) { + const queueName = "metrics_test_connected_queue" + + InitQueue(queueName) + + if _, ok := gatherFamily(t, "statusengine_queue_connected").find("queue_name", queueName); ok { + t.Errorf("InitQueue created a statusengine_queue_connected series for %q; "+ + "a pre-created 0 there claims the queue is disconnected, which is what the alert watches for", queueName) + } +} + func TestInitTableCreatesSeries(t *testing.T) { const table = "metrics_test_init_table" diff --git a/internal/queue/downtime_ordering_test.go b/internal/queue/downtime_ordering_test.go new file mode 100644 index 0000000..e1f111d --- /dev/null +++ b/internal/queue/downtime_ordering_test.go @@ -0,0 +1,156 @@ +package queue + +import ( + "context" + "sync" + "testing" + "time" + + gearmanClient "github.com/mikespook/gearman-go/client" +) + +// TestInOrderQueuesAreExactlyTheOnesThatNeedIt states the rule the whole +// serialization rests on, because the cost of getting it wrong runs in both +// directions and neither shows up as a failing test anywhere else. +// +// Adding a queue that does not need it silently throws away that queue's +// throughput headroom. Leaving out one that does costs data: the messages +// on such a queue are only meaningful in sequence, and the writes that +// implement them (a bare UPDATE, a DELETE) report success while doing +// nothing at all when they arrive early. +// +// statusngin_downtimes is the only one because it is the only queue whose +// messages refer to a row an *earlier message* was supposed to create. +// Every other queue carries self-contained events - a check result, a +// notification, a state change - written as an insert or an upsert that +// stands on its own. The two status queues come closest, since they upsert +// one row per object and an older snapshot could in principle overwrite a +// newer one, but each is complete in itself and the next check interval +// corrects it; a downtime's lost START is never re-sent by anything. +func TestInOrderQueuesAreExactlyTheOnesThatNeedIt(t *testing.T) { + if !RequiresInOrderProcessing(QueueDowntimes) { + t.Errorf("%s must be processed in order: its ADD/START/STOP/DELETE build on one another, "+ + "and START/STOP are UPDATEs that silently do nothing if they run before the ADD", QueueDowntimes) + } + + for queueName := range inOrderQueues { + if queueName != QueueDowntimes { + t.Errorf("%s was added to inOrderQueues: serializing a queue costs its concurrency, "+ + "so it needs the same justification statusngin_downtimes has - messages that are "+ + "only meaningful in sequence", queueName) + } + } +} + +// TestDowntimeQueueIsHandledOneAtATime is the regression test for six +// downtimes corrupted across a single job-server outage. +// +// The Gearman consumer dispatches up to -gearman-max-concurrent-jobs-per-queue +// handlers at once, and a downtime's ADD, START, STOP and DELETE arrive as +// separate jobs on one queue. Live traffic spaces them minutes apart so they +// never overlap; a backlog does not, and then START runs before ADD, its +// UPDATE matches nothing, and the downtime is recorded as never having +// started - with no error and no log line. +// +// Asserted from the handler's own side rather than through MySQL on +// purpose: what has to hold is that two handlers never run at the same +// time, and a database assertion would pass just as well on a run where the +// race happened not to be lost. +func TestDowntimeQueueIsHandledOneAtATime(t *testing.T) { + // Well above 1, so a consumer that ignored the rule would be caught + // rather than accidentally serialized by a low cap. + const perQueueLimit = 8 + + var mu sync.Mutex + concurrent, maxConcurrent := 0, 0 + handled := make(chan struct{}, 64) + + // Registered under the real queue name: that name is the input to + // RequiresInOrderProcessing, so a test function name would exercise the + // wrong branch. + router := Router{ + QueueDowntimes: func(_ context.Context, _ []byte) error { + mu.Lock() + concurrent++ + if concurrent > maxConcurrent { + maxConcurrent = concurrent + } + mu.Unlock() + + // Long enough that genuinely concurrent dispatch overlaps here + // with room to spare, short enough to keep the test quick. + time.Sleep(25 * time.Millisecond) + + mu.Lock() + concurrent-- + mu.Unlock() + + handled <- struct{}{} + return nil + }, + } + + cli, err := gearmanClient.New(gearmanClient.Network, gearmanAddr) + if err != nil { + skipOrFailService(t, "no reachable dev Gearman job server at %s: %v", gearmanAddr, err) + } + defer cli.Close() + + consumer := NewGearmanConsumer(gearmanAddr, router, perQueueLimit) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + if _, err := consumer.Start(ctx); err != nil { + skipOrFailService(t, "no reachable dev Gearman job server at %s: %v", gearmanAddr, err) + } + defer consumer.Stop() + + // Submitted in one burst with nothing draining them yet, which is what + // an outage leaves behind at the job server. + const jobs = 24 + for i := 0; i < jobs; i++ { + if _, err := cli.DoBg(QueueDowntimes, []byte(`{}`), gearmanClient.JobNormal); err != nil { + t.Fatalf("submit downtime job %d: %v", i, err) + } + } + + for i := 0; i < jobs; i++ { + select { + case <-handled: + case <-time.After(30 * time.Second): + t.Fatalf("only %d of %d downtime jobs were handled", i, jobs) + } + } + + mu.Lock() + got := maxConcurrent + mu.Unlock() + + if got != 1 { + t.Errorf("%d downtime handlers ran at once, want 1 - a downtime's ADD/START/STOP can then "+ + "overtake each other, and START's UPDATE silently does nothing when it wins", got) + } +} + +// TestOtherQueuesKeepTheirConcurrency is the other half: the serialization +// must apply to the downtime queue and nothing else. Without this, capping +// every queue at 1 would satisfy the test above while quietly undoing the +// per-queue budget the consumer exists to provide. +func TestOtherQueuesKeepTheirConcurrency(t *testing.T) { + const perQueueLimit = 4 + + c := NewGearmanConsumer(gearmanAddr, Router{}, perQueueLimit) + + if got := c.jobLimitFor(QueueDowntimes); got != 1 { + t.Errorf("job limit for %s = %d, want 1", QueueDowntimes, got) + } + for _, queueName := range []string{ + QueueHostStatus, QueueServiceStatus, QueueHostChecks, QueueServiceChecks, + QueueServicePerfdata, QueueStateChanges, QueueLogEntries, QueueNotifications, + QueueContactNotificationMethod, QueueAcknowledgements, QueueCoreRestart, + } { + if got := c.jobLimitFor(queueName); got != perQueueLimit { + t.Errorf("job limit for %s = %d, want the configured %d", queueName, got, perQueueLimit) + } + } +} diff --git a/internal/queue/downtime_unmatched_test.go b/internal/queue/downtime_unmatched_test.go new file mode 100644 index 0000000..8c27c15 --- /dev/null +++ b/internal/queue/downtime_unmatched_test.go @@ -0,0 +1,249 @@ +package queue + +import ( + "regexp" + "testing" + + sqlmock "github.com/DATA-DOG/go-sqlmock" + "github.com/prometheus/client_golang/prometheus" + + "statusengine-worker/internal/db" + "statusengine-worker/internal/types" +) + +// gatheredCounter reads one counter series back through the gatherer, +// returning 0 when the series does not exist. Read this way rather than +// from the collector so the test sees what a scrape would. +func gatheredCounter(t *testing.T, name, label, value string) float64 { + t.Helper() + + families, err := prometheus.DefaultGatherer.Gather() + if err != nil { + t.Fatalf("Gather: %v", err) + } + for _, family := range families { + if family.GetName() != name { + continue + } + for _, metric := range family.GetMetric() { + for _, pair := range metric.GetLabel() { + if pair.GetName() == label && pair.GetValue() == value { + return metric.GetCounter().GetValue() + } + } + } + } + return 0 +} + +// startMessage is a STARTed host downtime whose UPDATE targets +// statusengine_host_downtimehistory - the exact statement that went missing +// for six downtimes across one job-server outage. +func startMessage() types.DowntimeMessage { + return types.DowntimeMessage{ + Envelope: types.Envelope{Type: types.EventTypeDowntimeStart, Timestamp: 2000}, + Downtime: hostDowntimePayload, + } +} + +// expectExistsCheck queues the follow-up SELECT that disambiguates a +// zero-row UPDATE, answering with found or not-found. +func expectExistsCheck(mock sqlmock.Sqlmock, action DowntimeAction, found bool) { + query, args := db.DowntimeHistoryExistsQuery(toDBRow(action.Data)) + expectation := mock.ExpectQuery("^" + regexp.QuoteMeta(query) + "$").WithArgs(toDriverArgs(args)...) + + rows := sqlmock.NewRows([]string{"1"}) + if found { + rows.AddRow(1) + } + expectation.WillReturnRows(rows) +} + +// TestDowntimeUpdateWithoutItsRowIsReported is the visibility half of the +// downtime ordering fix. +// +// START and STOP are bare UPDATE ... WHERE against the row the +// downtime's ADD created. When that row is absent the statement succeeds, +// affects nothing, and the event is gone - was_started stays 0 on a +// downtime that demonstrably ran. That is how six downtimes were corrupted +// without a single log line, found only weeks later by diffing against the +// legacy PHP worker. Serializing the queue removes the cause; this counter +// is what would make a recurrence visible on its own. +func TestDowntimeUpdateWithoutItsRowIsReported(t *testing.T) { + const table = "statusengine_host_downtimehistory" + + handler, mock, _, ctx := setupDowntimeHandler(t) + msg := startMessage() + + before := gatheredCounter(t, "statusengine_queue_downtime_updates_unmatched_total", "table", table) + + for _, a := range DetermineDowntimeActions(msg, testDowntimeNodeName) { + query, args := buildDowntimeQuery(a) + // Zero rows affected: nothing was there to update. + mock.ExpectExec("^" + regexp.QuoteMeta(query) + "$"). + WithArgs(toDriverArgs(args)...). + WillReturnResult(sqlmock.NewResult(0, 0)) + + if a.Table == DowntimeHistoryTable && a.Action == DowntimeActionUpdateStarted { + expectExistsCheck(mock, a, false) + } + } + + if err := handler(ctx, marshalDowntimeMessage(t, msg)); err != nil { + t.Fatalf("handler: %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sqlmock expectations: %v", err) + } + + if got := gatheredCounter(t, "statusengine_queue_downtime_updates_unmatched_total", "table", table); got != before+1 { + t.Errorf("unmatched-update counter = %v, want %v - a START whose row does not exist is a lost "+ + "event and has to be countable", got, before+1) + } + +} + +// TestRedeliveredDowntimeUpdateIsNotReported is why that counter costs a +// second query instead of trusting RowsAffected. +// +// MySQL counts rows *changed*, not matched - go-sql-driver leaves +// CLIENT_FOUND_ROWS off - so a redelivered START rewriting the values it +// already wrote reports zero affected rows too. Redelivery is normal here +// (CLAUDE.md rule 6), so treating that as a lost event would put a steady +// trickle of false findings into the one series that is supposed to mean +// "data went missing". +func TestRedeliveredDowntimeUpdateIsNotReported(t *testing.T) { + const table = "statusengine_host_downtimehistory" + + handler, mock, _, ctx := setupDowntimeHandler(t) + msg := startMessage() + + before := gatheredCounter(t, "statusengine_queue_downtime_updates_unmatched_total", "table", table) + + for _, a := range DetermineDowntimeActions(msg, testDowntimeNodeName) { + query, args := buildDowntimeQuery(a) + mock.ExpectExec("^" + regexp.QuoteMeta(query) + "$"). + WithArgs(toDriverArgs(args)...). + WillReturnResult(sqlmock.NewResult(0, 0)) + + if a.Table == DowntimeHistoryTable && a.Action == DowntimeActionUpdateStarted { + // The row is there, it simply already held these values. + expectExistsCheck(mock, a, true) + } + } + + if err := handler(ctx, marshalDowntimeMessage(t, msg)); err != nil { + t.Fatalf("handler: %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sqlmock expectations: %v", err) + } + + if got := gatheredCounter(t, "statusengine_queue_downtime_updates_unmatched_total", "table", table); got != before { + t.Errorf("unmatched-update counter = %v, want it unchanged at %v - the row existed, so this was "+ + "a redelivery rewriting identical values, not a lost event", got, before) + } +} + +// TestDeleteMatchingNothingIsNotReported pins the exclusion that keeps the +// counter meaningful. Every ordinary downtime ends with STOP removing the +// scheduleddowntimes row and DELETE then finding it already gone, so a +// zero-row DELETE is the normal case, not a finding - and counting it would +// bury the real signal under one increment per downtime. +func TestDeleteMatchingNothingIsNotReported(t *testing.T) { + handler, mock, _, ctx := setupDowntimeHandler(t) + + // A DELETE for a downtime that had already started: scheduleddowntimes + // only, no downtimehistory delete (the "never started" case needs + // start_time > timestamp). + msg := types.DowntimeMessage{ + Envelope: types.Envelope{Type: types.EventTypeDowntimeDelete, Timestamp: 3000}, + Downtime: hostDowntimePayload, + } + + actions := DetermineDowntimeActions(msg, testDowntimeNodeName) + if len(actions) == 0 { + t.Fatal("test setup: a DELETE produced no actions") + } + + var before float64 + for _, table := range downtimeHistoryTables() { + before += gatheredCounter(t, "statusengine_queue_downtime_updates_unmatched_total", "table", table) + } + + for _, a := range actions { + if a.Action != DowntimeActionDelete { + t.Fatalf("test setup: expected only DELETEs from this message, got %s on %s", a.Action, a.Table) + } + query, args := buildDowntimeQuery(a) + mock.ExpectExec("^" + regexp.QuoteMeta(query) + "$"). + WithArgs(toDriverArgs(args)...). + WillReturnResult(sqlmock.NewResult(0, 0)) + } + + if err := handler(ctx, marshalDowntimeMessage(t, msg)); err != nil { + t.Fatalf("handler: %v", err) + } + // No follow-up SELECT may have been issued: ExpectationsWereMet is what + // proves the DELETE path asked the database nothing extra. + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sqlmock expectations: %v", err) + } + + var after float64 + for _, table := range downtimeHistoryTables() { + after += gatheredCounter(t, "statusengine_queue_downtime_updates_unmatched_total", "table", table) + } + if after != before { + t.Errorf("unmatched-update counter moved from %v to %v on a DELETE - a DELETE that matches "+ + "nothing happens on every ordinary downtime and is not a finding", before, after) + } +} + +// TestDowntimeHistoryTablesAreTheUpdatableSubset ties the pre-created +// series to the tables a downtime UPDATE can actually target. Only +// downtimehistory is ever updated; the scheduleddowntimes pair is upserted +// or deleted, so a series for those two would advertise a failure mode they +// cannot have. +func TestDowntimeHistoryTablesAreTheUpdatableSubset(t *testing.T) { + all := map[string]bool{} + for _, table := range downtimeMetricsTables() { + all[table] = true + } + + history := downtimeHistoryTables() + if len(history) != 2 { + t.Fatalf("got %d downtimehistory tables, want 2 (host and service)", len(history)) + } + for _, table := range history { + if !all[table] { + t.Errorf("%s is not among the downtime tables - the two lists have drifted apart", table) + } + } + + // Derived from the decision engine rather than asserted by name: every + // UPDATE it can produce must land on one of those two tables. + updatable := map[string]bool{} + for _, msg := range []types.DowntimeMessage{ + {Envelope: types.Envelope{Type: types.EventTypeDowntimeStart, Timestamp: 2000}, Downtime: hostDowntimePayload}, + {Envelope: types.Envelope{Type: types.EventTypeDowntimeStop, Timestamp: 2600}, Downtime: hostDowntimePayload}, + {Envelope: types.Envelope{Type: types.EventTypeDowntimeStart, Timestamp: 2000}, Downtime: serviceDowntimePayload}, + {Envelope: types.Envelope{Type: types.EventTypeDowntimeStop, Timestamp: 2600}, Downtime: serviceDowntimePayload}, + } { + for _, action := range DetermineDowntimeActions(msg, testDowntimeNodeName) { + if action.Action == DowntimeActionUpdateStarted || action.Action == DowntimeActionUpdateStopped { + updatable[downtimeMetricsTable(action)] = true + } + } + } + + if len(updatable) != len(history) { + t.Errorf("the decision engine produces UPDATEs on %d tables, but %d are pre-created: %v vs %v", + len(updatable), len(history), updatable, history) + } + for _, table := range history { + if !updatable[table] { + t.Errorf("%s has a pre-created unmatched-update series but no UPDATE ever targets it", table) + } + } +} diff --git a/internal/queue/gearman.go b/internal/queue/gearman.go index f6fb033..52ac143 100644 --- a/internal/queue/gearman.go +++ b/internal/queue/gearman.go @@ -60,11 +60,16 @@ type GearmanConsumer struct { // gearman.Unlimited. maxConcurrentJobsPerQueue int - mu sync.Mutex - workers []*gearman.Worker - out chan Message + // mu guards queues (the *gearman.Worker inside each entry is swapped + // by superviseReconnects) and out. Closing stopping happens under it + // too - see Stop for why that is the whole synchronisation between a + // reconnect landing and a shutdown starting. + mu sync.Mutex + queues []*queueWorker + out chan Message stopOnce sync.Once + stopping chan struct{} statsDone chan struct{} // handlerWG counts job handlers currently in flight, so Stop can wait @@ -80,11 +85,35 @@ type GearmanConsumer struct { stopped bool handlerWG sync.WaitGroup - // processed/errors count jobs handled since Start, for the periodic - // stats log. Incremented from per-connection job-handler goroutines, - // hence atomic. - processed atomic.Uint64 - errors atomic.Uint64 + // processed/errors/reconnects count activity since Start, for the + // periodic stats log. Incremented from per-connection job-handler + // goroutines and the per-queue reconnect supervisors, hence atomic. + processed atomic.Uint64 + errors atomic.Uint64 + reconnects atomic.Uint64 +} + +// queueWorker is one queue's connection to the job server, together with +// what it takes to rebuild it. The *gearman.Worker is a generation rather +// than an identity: superviseReconnects replaces it wholesale after a +// disconnect, which is why every reader takes GearmanConsumer.mu. +type queueWorker struct { + queue string + handle Handler + + // w is the current generation. Read and written only under + // GearmanConsumer.mu. + w *gearman.Worker + + // lost carries one reconnect request from the worker's ErrorHandler to + // this queue's supervisor. Capacity 1 and only ever written to with a + // non-blocking send: the ErrorHandler runs on the agent goroutine with + // the agent's own mutex held (agent.go's disconnect_error), so blocking + // there would deadlock anything that needs that mutex - including the + // reconnect itself. One pending request is all the supervisor needs, + // since it rebuilds the whole connection regardless of how many times + // it was told. + lost chan struct{} } // NewGearmanConsumer creates a consumer that will connect to the Gearman @@ -124,6 +153,7 @@ func NewGearmanConsumer(addr string, router Router, maxConcurrentJobsPerQueue in router: router, maxConcurrentJobsPerQueue: maxConcurrentJobsPerQueue, statsDone: make(chan struct{}), + stopping: make(chan struct{}), } } @@ -146,51 +176,45 @@ func (c *GearmanConsumer) Start(ctx context.Context) (<-chan Message, error) { } out := make(chan Message, outboundBufferSize) - workers := make([]*gearman.Worker, 0, len(c.router)) + queues := make([]*queueWorker, 0, len(c.router)) // Anything that fails partway through leaves the connections opened so - // far with nobody to close them, so unwind before returning. + // far with nobody to close them, so unwind before returning. Note that + // Close cannot actually drop them: the library returns from it early + // while running is false, and running is only set by Work, which has + // not been started yet. Those sockets go when the process does - which + // it always does, since a failed Start is fatal in cmd/app. fail := func(err error) (<-chan Message, error) { - for _, w := range workers { - w.Close() + for _, qw := range queues { + qw.w.Close() } return nil, err } for queueName, handle := range c.router { - // Not gearman.Unlimited - see NewGearmanConsumer. Note the library - // buffers limit-1 tokens (worker.go's New) and sends one only after - // spawning the job's goroutine, so New(n) permits exactly n - // concurrent handlers and New(1) serializes them. That is correct - // as written; it only looks like an off-by-one. - w := gearman.New(c.maxConcurrentJobsPerQueue) - if err := w.AddServer(gearman.Network, c.addr); err != nil { - return fail(fmt.Errorf("gearman: connect to %s for %q: %w", c.addr, queueName, err)) - } - w.ErrorHandler = func(err error) { - slog.Warn("gearman: worker error", "queue", queueName, "error", err) + qw := &queueWorker{ + queue: queueName, + handle: handle, + lost: make(chan struct{}, 1), } - // Exactly one function per worker: that is the whole point of the - // split (see the type comment). Registering a second here would - // silently restore the shared budget for those two queues. - if err := w.AddFunc(queueName, c.jobHandler(ctx, queueName, handle, out), 0); err != nil { - return fail(fmt.Errorf("gearman: register function %q: %w", queueName, err)) - } - if err := w.Ready(); err != nil { - return fail(fmt.Errorf("gearman: ready for %q: %w", queueName, err)) + w, err := c.newWorker(ctx, qw, out) + if err != nil { + return fail(err) } + qw.w = w - workers = append(workers, w) + queues = append(queues, qw) } c.mu.Lock() - c.workers = workers + c.queues = queues c.out = out c.mu.Unlock() - for _, w := range workers { - go w.Work() + for _, qw := range queues { + go qw.w.Work() + go c.superviseReconnects(ctx, qw, out) } go c.logStatsPeriodically(ctx) @@ -199,13 +223,218 @@ func (c *GearmanConsumer) Start(ctx context.Context) (<-chan Message, error) { c.Stop() }() + inOrder := 0 + for queueName := range c.router { + if RequiresInOrderProcessing(queueName) { + inOrder++ + } + } slog.Info("gearman: consumer started", "addr", c.addr, "queues", len(c.router), - "max_concurrent_jobs_per_queue", c.maxConcurrentJobsPerQueue) + "max_concurrent_jobs_per_queue", c.maxConcurrentJobsPerQueue, + "serialized_queues", inOrder) return out, nil } +// newWorker opens one connection to the job server for qw's queue, +// registers that queue's function on it and reports it ready to work. It +// does not start the Work loop - the caller does, because Start and +// superviseReconnects hang the worker into different places first. +// +// Shared by both of those on purpose: a connection rebuilt after a drop +// has to be configured identically to the one opened at startup, and the +// only way to guarantee that is for there to be one piece of code that +// configures it. +func (c *GearmanConsumer) newWorker(ctx context.Context, qw *queueWorker, out chan Message) (*gearman.Worker, error) { + // Not gearman.Unlimited - see NewGearmanConsumer. Note the library + // buffers limit-1 tokens (worker.go's New) and sends one only after + // spawning the job's goroutine, so New(n) permits exactly n concurrent + // handlers and New(1) serializes them. That is correct as written; it + // only looks like an off-by-one. + // + // A queue whose messages build on one another gets 1 regardless of the + // configured cap, which is what makes New(1)'s serialization + // load-bearing rather than a curiosity. See RequiresInOrderProcessing: + // statusngin_downtimes delivers a downtime's ADD/START/STOP/DELETE as + // separate jobs, and handling them concurrently silently corrupts the + // two downtime table pairs whenever a backlog makes them overlap. + // Costs nothing: that queue sees a handful of messages an hour, and + // per-queue concurrency buys no throughput anyway (CLAUDE.md rule 2 - + // the bottleneck is the single Run goroutine per table). + w := gearman.New(c.jobLimitFor(qw.queue)) + if err := w.AddServer(gearman.Network, c.addr); err != nil { + return nil, fmt.Errorf("gearman: connect to %s for %q: %w", c.addr, qw.queue, err) + } + w.ErrorHandler = c.errorHandler(qw) + + // Exactly one function per worker: that is the whole point of the + // split (see the type comment). Registering a second here would + // silently restore the shared budget for those two queues. + if err := w.AddFunc(qw.queue, c.jobHandler(ctx, qw.queue, qw.handle, out), 0); err != nil { + return nil, fmt.Errorf("gearman: register function %q: %w", qw.queue, err) + } + if err := w.Ready(); err != nil { + return nil, fmt.Errorf("gearman: ready for %q: %w", qw.queue, err) + } + + // Set here rather than pre-created at 1 in metrics.InitQueue, so the + // series means "this queue has a connection" and not "the Router was + // wired up". Ready has dialled and sent CAN_DO by this point. + metrics.QueueConnected.WithLabelValues(qw.queue).Set(1) + + return w, nil +} + +// jobLimitFor is the concurrency cap to open one queue's connection with: +// the configured per-queue cap, or 1 for a queue that must be processed in +// order. Split out so Start can report how many queues are serialized +// without repeating the rule. +func (c *GearmanConsumer) jobLimitFor(queueName string) int { + if RequiresInOrderProcessing(queueName) { + return 1 + } + return c.maxConcurrentJobsPerQueue +} + +// errorHandler builds the gearman.ErrorHandler for one queue. Its real job +// is to notice that this queue's only connection has died, because nothing +// else in the library will: agent.work reports the disconnect here and then +// returns, leaving Work() ranging over a channel no agent will ever send on +// again (worker.go's `for inpack = range worker.in`, which only Close +// terminates). The worker then sits there looking healthy and consumes +// nothing, for good - which is exactly what a lost job server used to cost +// this process. +// +// Only a *gearman.WorkerDisconnectError means that. Handler failures reach +// the same ErrorHandler (handleInPack routes exec's error through +// worker.err), and treating those as a disconnect would tear down a working +// connection every time MySQL rejected a batch. +func (c *GearmanConsumer) errorHandler(qw *queueWorker) func(error) { + return func(err error) { + var disconnected *gearman.WorkerDisconnectError + if !errors.As(err, &disconnected) { + slog.Warn("gearman: worker error", "queue", qw.queue, "error", err) + return + } + + metrics.QueueConnected.WithLabelValues(qw.queue).Set(0) + slog.Warn("gearman: connection lost, reconnecting", + "queue", qw.queue, "error", err, "retry_interval", reconnectDelay) + + // Non-blocking, and it has to be: this runs on the agent goroutine + // while agent.disconnect_error holds the agent's mutex, so blocking + // here would hold that mutex for as long as the supervisor took to + // answer. One queued request is enough - see queueWorker.lost. + select { + case qw.lost <- struct{}{}: + default: + } + } +} + +// superviseReconnects rebuilds one queue's connection after it is lost, +// retrying every reconnectDelay until it succeeds, ctx is cancelled or Stop +// is called. One goroutine per queue, so twelve queues dropped by a single +// job server restart come back in parallel rather than one after another - +// the same reason closeWorkers closes them in parallel. +// +// It builds a fresh gearman.Worker rather than calling the library's +// WorkerDisconnectError.Reconnect(), for two independent reasons. The first +// is decisive: agent.disconnect_error calls the ErrorHandler while holding +// the agent's mutex, and reconnect() takes that same mutex, so reconnecting +// from where the disconnect is reported is a self-deadlock - it has to +// happen on another goroutine anyway. The second is that reconnect() reuses +// the dead agent in place: it overwrites a.conn without closing the old +// socket (leaving the fd to the GC finalizer), it takes the agent mutex and +// then the worker mutex, which is the reverse of AddFunc -> broadcast -> +// agent.Write, and it calls agentWG.Add on a WaitGroup that Close may be +// waiting on. Rebuilding uses only New/AddServer/AddFunc/Ready/Work/Close, +// which run on every startup and in every test in this package, and closing +// the dead generation disposes of its socket properly. +// +// Jobs that were in flight when the connection dropped lose their +// acknowledgement, so the job server hands them out again after the +// reconnect. That is harmless here and it is not an accident: it is exactly +// the redelivery CLAUDE.md rule 6's upserts exist to absorb. +func (c *GearmanConsumer) superviseReconnects(ctx context.Context, qw *queueWorker, out chan Message) { + for { + select { + case <-ctx.Done(): + return + case <-c.stopping: + return + case <-qw.lost: + } + + c.reconnects.Add(1) + metrics.QueueReconnectsTotal.WithLabelValues(qw.queue).Inc() + + // End the dead generation before building the next one. Its Work + // loop is parked forever on a channel nothing will send to again, + // and its agent still holds the dead socket; Close is what releases + // both, after letting the handlers still running report their + // result (bounded by the fork's DrainTimeout, 30s). + // + // Before the retry loop rather than after, so that worst case falls + // inside an outage the worker is waiting out anyway instead of + // being added on top of a reconnect that would otherwise succeed. + c.mu.Lock() + dead := qw.w + c.mu.Unlock() + dead.Close() + + for { + select { + case <-ctx.Done(): + return + case <-c.stopping: + return + case <-time.After(reconnectDelay): + } + + fresh, err := c.newWorker(ctx, qw, out) + if err != nil { + slog.Warn("gearman: reconnect attempt failed", "queue", qw.queue, "error", err) + continue + } + + // Hanging the new generation in and observing that a shutdown + // has begun must not be able to interleave, which is why Stop + // closes c.stopping under this same mutex. Either this lands + // first and Stop's snapshot includes it, or Stop got there + // first and this generation is not in that snapshot - in which + // case it must not be allowed to consume, because Stop has + // already returned and the BulkInserters behind it are being + // flushed. + // + // What makes that safe is the order below, not the Close: a + // worker only asks for jobs once Work sends its first + // GRAB_JOB_UNIQ, and Work is started after this check. The + // Close is best-effort cleanup and known to be a no-op here, + // since the library returns early from it while running is + // false - the same reason Start's fail unwind above cannot + // close what it opened either. What is left behind is an idle + // socket that has sent CAN_DO and will never grab, released + // when the process exits. + c.mu.Lock() + select { + case <-c.stopping: + c.mu.Unlock() + fresh.Close() + return + default: + } + qw.w = fresh + c.mu.Unlock() + + go fresh.Work() + slog.Info("gearman: reconnected", "queue", qw.queue, "addr", c.addr) + break + } + } +} + // jobHandler builds the gearman.JobHandler for one queue. Extracted from // Start only so the per-queue setup loop above stays readable; it closes // over nothing that differs between workers except queueName and handle. @@ -293,7 +522,8 @@ func (c *GearmanConsumer) logStatsPeriodically(ctx context.Context) { select { case <-ticker.C: slog.Info("gearman: consumer stats", - "addr", c.addr, "processed", c.processed.Load(), "errors", c.errors.Load()) + "addr", c.addr, "processed", c.processed.Load(), "errors", c.errors.Load(), + "reconnects", c.reconnects.Load()) case <-c.statsDone: return case <-ctx.Done(): @@ -308,6 +538,11 @@ func (c *GearmanConsumer) logStatsPeriodically(ctx context.Context) { // might still be in progress. Safe to call multiple times and safe to call // without a prior Start. // +// It also tells the per-queue reconnect supervisors to stop retrying, so a +// shutdown that lands during a job-server outage is not held up by one - +// closing c.stopping is what ends both their backoff wait and their loop. +// See superviseReconnects for why that close happens under c.mu. +// // Three upstream defects made this unsafe; all are fixed in the patched // fork this module points at (see go.mod's replace directive). // @@ -347,8 +582,20 @@ func (c *GearmanConsumer) Stop() error { c.stopOnce.Do(func() { close(c.statsDone) + // Both under the mutex, and that is the entire synchronisation + // against a reconnect landing at the same moment: a supervisor + // hangs its new worker in under this same lock and checks + // c.stopping while holding it, so it either gets into the snapshot + // below or sees the shutdown and closes its own worker. Splitting + // these two lines would leave a window in which a fresh connection + // is created that nothing ever closes. c.mu.Lock() - workers, out := c.workers, c.out + close(c.stopping) + queues, out := c.queues, c.out + workers := make([]*gearman.Worker, 0, len(queues)) + for _, qw := range queues { + workers = append(workers, qw.w) + } c.mu.Unlock() if len(workers) == 0 { @@ -357,6 +604,10 @@ func (c *GearmanConsumer) Stop() error { closeWorkers(workers) + for _, qw := range queues { + metrics.QueueConnected.WithLabelValues(qw.queue).Set(0) + } + // Close the gate before waiting: after this write lock is // released, beginHandler can only ever return false, so the // WaitGroup counter can no longer rise and Wait is safe. Doing it @@ -373,7 +624,8 @@ func (c *GearmanConsumer) Stop() error { } slog.Info("gearman: consumer stopped", - "addr", c.addr, "processed", c.processed.Load(), "errors", c.errors.Load()) + "addr", c.addr, "processed", c.processed.Load(), "errors", c.errors.Load(), + "reconnects", c.reconnects.Load()) }) return nil } diff --git a/internal/queue/gearman_reconnect_test.go b/internal/queue/gearman_reconnect_test.go new file mode 100644 index 0000000..fe737cb --- /dev/null +++ b/internal/queue/gearman_reconnect_test.go @@ -0,0 +1,319 @@ +package queue + +import ( + "context" + "fmt" + "io" + "net" + "sync" + "testing" + "time" + + gearmanClient "github.com/mikespook/gearman-go/client" + "github.com/prometheus/client_golang/prometheus" +) + +// tcpBreaker is a minimal TCP proxy that a test can sit in front of the +// job server in order to sever the connection on demand. +// +// It exists because there is no other way to reach the connection from a +// test. The RabbitMQ reconnect test simply calls Close on the consumer's +// own *amqp.Connection; gearman-go keeps its socket in an unexported field +// of an unexported agent, with nothing exported that leads to it. The only +// alternatives would be restarting the real gearmand - which a test must +// not do to a shared dev service - or trusting that the code path is right +// because it looks right, which is what left this bug in place. +type tcpBreaker struct { + ln net.Listener + backend string + + mu sync.Mutex + conns []net.Conn + closed bool +} + +// newTCPBreaker starts a proxy forwarding to backend and returns it. Every +// connection it accepts, and every one it opens to the backend, is +// remembered so that breakAll can drop them all at once. +func newTCPBreaker(t *testing.T, backend string) *tcpBreaker { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + b := &tcpBreaker{ln: ln, backend: backend} + go b.acceptLoop() + t.Cleanup(b.close) + + return b +} + +func (b *tcpBreaker) addr() string { return b.ln.Addr().String() } + +func (b *tcpBreaker) acceptLoop() { + for { + client, err := b.ln.Accept() + if err != nil { + return // listener closed + } + + server, err := net.Dial("tcp", b.backend) + if err != nil { + client.Close() + continue + } + + if !b.track(client, server) { + client.Close() + server.Close() + return + } + + go func() { io.Copy(server, client); server.Close() }() + go func() { io.Copy(client, server); client.Close() }() + } +} + +// track remembers one connection pair, reporting false if the breaker has +// already been closed - in which case the caller must drop the pair itself +// rather than leaving it running past the end of the test. +func (b *tcpBreaker) track(conns ...net.Conn) bool { + b.mu.Lock() + defer b.mu.Unlock() + + if b.closed { + return false + } + b.conns = append(b.conns, conns...) + return true +} + +// waitForConnection blocks until the proxy has accepted and paired at +// least one connection. +// +// Necessary because a consumer's Start can return before this proxy has +// even accepted: the dial into the listener's backlog succeeds +// immediately, and the CAN_DO that follows goes into the socket buffer, so +// Ready reports success while acceptLoop is still dialling the backend. +// breakAll called at that moment finds an empty list, breaks nothing, and +// the test then waits out its timeout on a connection that was never +// severed - which is exactly how this failed roughly one run in six before +// the wait existed. +func (b *tcpBreaker) waitForConnection(t *testing.T) { + t.Helper() + + for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(10 * time.Millisecond) { + b.mu.Lock() + n := len(b.conns) + b.mu.Unlock() + + if n > 0 { + return + } + if time.Now().After(deadline) { + t.Fatal("the proxy never accepted a connection from the consumer") + } + } +} + +// breakAll drops every connection currently proxied, which is what the +// consumer sees as its job server going away. New connections are still +// accepted afterwards, so a reconnect can succeed. +func (b *tcpBreaker) breakAll() { + b.mu.Lock() + conns := b.conns + b.conns = nil + b.mu.Unlock() + + for _, conn := range conns { + conn.Close() + } +} + +// stopAccepting closes the listener while leaving the breaker usable, so a +// test can make every reconnect attempt fail rather than merely be slow. +func (b *tcpBreaker) stopAccepting() { b.ln.Close() } + +func (b *tcpBreaker) close() { + b.ln.Close() + + b.mu.Lock() + b.closed = true + conns := b.conns + b.conns = nil + b.mu.Unlock() + + for _, conn := range conns { + conn.Close() + } +} + +// TestGearmanConsumerReconnectsAfterConnectionDrop is the regression test +// for a worker that stopped consuming for good when it lost the job server. +// +// The library does not recover on its own and does not pretend to: on EOF +// agent.work hands a *WorkerDisconnectError to the ErrorHandler and +// returns, so nothing sends on worker.in any more and Work() - which is a +// plain range over that channel - parks there forever. The consumer stayed +// up, kept logging its stats line, and processed nothing: twelve WARN lines +// and then silence until someone restarted it by hand. +// +// So this asserts the only thing that distinguishes a fix from a +// well-argued comment: a job published *after* the drop still reaches its +// Handler. +func TestGearmanConsumerReconnectsAfterConnectionDrop(t *testing.T) { + breaker := newTCPBreaker(t, gearmanAddr) + + // Unique per run: gearmand is shared here, and a leftover job from an + // earlier run must not be able to satisfy the assertion below. + fnName := fmt.Sprintf("queue_pkg_test_reconnect_%d", time.Now().UnixNano()) + + // Buffered for the same reason as in TestGearmanConsumerEndToEnd, and + // with one extra consideration here: a job whose acknowledgement was + // lost in the drop is redelivered after the reconnect, so the handler + // can legitimately run more often than jobs were published. + received := make(chan []byte, 64) + router := Router{ + fnName: func(_ context.Context, payload []byte) error { + received <- payload + return nil + }, + } + + consumer := NewGearmanConsumer(breaker.addr(), router, 8) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + if _, err := consumer.Start(ctx); err != nil { + skipOrFailService(t, "no reachable dev Gearman job server at %s: %v", gearmanAddr, err) + } + defer consumer.Stop() + + cli, err := gearmanClient.New(gearmanClient.Network, gearmanAddr) + if err != nil { + t.Fatalf("gearman client: %v", err) + } + defer cli.Close() + + // Establish that it works at all before breaking it, so a failure + // below can only mean the reconnect. + if _, err := cli.DoBg(fnName, []byte(`{"before":"drop"}`), gearmanClient.JobNormal); err != nil { + t.Fatalf("submit job before the drop: %v", err) + } + select { + case <-received: + case <-time.After(10 * time.Second): + t.Fatal("handler never ran before the connection was dropped") + } + + breaker.waitForConnection(t) + breaker.breakAll() + + // Wait for the supervisor to have finished, not merely started: it + // counts the reconnect, closes the dead generation and only then + // rebuilds, so publishing on the strength of the counter alone would + // race the new CAN_DO. Poll for the connection instead, which is what + // the queue_connected gauge exists to expose. + deadline := time.Now().Add(30 * time.Second) + for { + if consumer.reconnects.Load() > 0 && gatheredQueueConnected(t, fnName) == 1 { + break + } + if time.Now().After(deadline) { + t.Fatalf("consumer did not reconnect within 30s of the connection being dropped "+ + "(reconnects=%d)", consumer.reconnects.Load()) + } + time.Sleep(50 * time.Millisecond) + } + + if _, err := cli.DoBg(fnName, []byte(`{"after":"drop"}`), gearmanClient.JobNormal); err != nil { + t.Fatalf("submit job after the drop: %v", err) + } + select { + case <-received: + case <-time.After(15 * time.Second): + t.Fatal("handler never ran after the reconnect - the consumer stopped consuming for good") + } +} + +// TestGearmanConsumerStopsWhileReconnecting pins the property the whole +// construction hangs on: the retry loop must never outlast a shutdown. +// +// It retries forever by design (the job server coming back is exactly the +// case worth waiting for), so the only thing keeping a Stop during an +// outage from running past systemd's TimeoutStopSec - which would SIGKILL +// the worker mid-flush and lose the buffered rows - is that closing +// c.stopping ends both the backoff wait and the loop itself. +func TestGearmanConsumerStopsWhileReconnecting(t *testing.T) { + breaker := newTCPBreaker(t, gearmanAddr) + + fnName := fmt.Sprintf("queue_pkg_test_stop_reconnecting_%d", time.Now().UnixNano()) + router := Router{fnName: func(context.Context, []byte) error { return nil }} + + consumer := NewGearmanConsumer(breaker.addr(), router, 8) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + if _, err := consumer.Start(ctx); err != nil { + skipOrFailService(t, "no reachable dev Gearman job server at %s: %v", gearmanAddr, err) + } + + // Drop the connection and make sure nothing can be dialled again, so + // the supervisor is genuinely stuck in its retry loop rather than + // recovering before Stop is called. + breaker.waitForConnection(t) + breaker.stopAccepting() + breaker.breakAll() + + for deadline := time.Now().Add(10 * time.Second); consumer.reconnects.Load() == 0; { + if time.Now().After(deadline) { + t.Fatal("consumer never noticed the connection had dropped") + } + time.Sleep(50 * time.Millisecond) + } + + stopped := make(chan error, 1) + start := time.Now() + go func() { stopped <- consumer.Stop() }() + + select { + case err := <-stopped: + if err != nil { + t.Fatalf("stop: %v", err) + } + t.Logf("Stop returned after %s while a reconnect was pending", time.Since(start).Round(time.Millisecond)) + case <-time.After(15 * time.Second): + t.Fatal("Stop did not return while a reconnect was pending - a job-server outage would " + + "hold the shutdown past TimeoutStopSec and cost the buffered rows") + } +} + +// gatheredQueueConnected reads statusengine_queue_connected for one queue +// through the gatherer, reporting -1 when the series does not exist at all +// (which is its state before any consumer has connected). Read this way +// rather than from the consumer, because there is no field to read - the +// gauge is the observable. +func gatheredQueueConnected(t *testing.T, queueName string) float64 { + t.Helper() + + families, err := prometheus.DefaultGatherer.Gather() + if err != nil { + t.Fatalf("Gather: %v", err) + } + + for _, family := range families { + if family.GetName() != "statusengine_queue_connected" { + continue + } + for _, metric := range family.GetMetric() { + for _, pair := range metric.GetLabel() { + if pair.GetName() == "queue_name" && pair.GetValue() == queueName { + return metric.GetGauge().GetValue() + } + } + } + } + return -1 +} diff --git a/internal/queue/metrics_init_test.go b/internal/queue/metrics_init_test.go index e0245c5..5bf039b 100644 --- a/internal/queue/metrics_init_test.go +++ b/internal/queue/metrics_init_test.go @@ -90,11 +90,17 @@ func TestNewRouterPreCreatesMetricSeries(t *testing.T) { router, _ := NewRouter(sqlDB, hub, graphite.NewClient("127.0.0.1:2003"), PerfdataRouteMySQL, "statusengine-test", "statusengine-test", false, noAgeFilter, testBatchSize) - // Every queue in the router, on all three per-queue metrics. + // Every queue in the router, on all four per-queue metrics. + // + // statusengine_queue_connected is deliberately not among them: it is + // set by the consumer once a connection is actually up, so that a 1 + // means "connected" rather than "the Router was wired up". See + // TestInitQueueLeavesQueueConnectedAlone (internal/metrics). for _, name := range []string{ "statusengine_queue_messages_received_total", "statusengine_queue_payloads_repaired_total", "statusengine_queue_handler_duration_seconds", + "statusengine_queue_reconnects_total", } { exported := gatheredLabelValues(t, name, "queue_name") for queueName := range router { diff --git a/internal/queue/rabbitmq.go b/internal/queue/rabbitmq.go index 82cb45b..63e78e4 100644 --- a/internal/queue/rabbitmq.go +++ b/internal/queue/rabbitmq.go @@ -15,13 +15,19 @@ import ( "statusengine-worker/internal/metrics" ) -// reconnectDelay is how long the consumer waits between reconnect attempts -// after an unexpected disconnect. Unlike the Gearman client library (see -// gearman.go's KNOWN ISSUE comment for the one thing it doesn't handle), -// amqp091-go never reconnects on its own - a dropped TCP connection just -// closes every delivery channel and stops there - so this consumer has to -// redial itself to satisfy CLAUDE.md rule 6 ("reconnect automatically to -// MySQL/Queues on connection drops"). +// reconnectDelay is how long either consumer waits between reconnect +// attempts after an unexpected disconnect. Neither backend's client library +// recovers on its own: amqp091-go closes every delivery channel and stops +// there, and gearman-go's agent goroutine reports the disconnect to the +// ErrorHandler and returns, leaving its Work loop parked on a channel +// nothing will ever send on again. So both consumers redial themselves, to +// satisfy CLAUDE.md rule 6 ("reconnect automatically to MySQL/Queues on +// connection drops"). +// +// This comment used to claim the opposite about Gearman - that the library +// handled it and only shutdown was a problem. That assumption is why a lost +// job server silently stopped the worker consuming until it was restarted +// by hand; see superviseReconnects in gearman.go. const reconnectDelay = 2 * time.Second // requeueDelay is how long a delivery loop pauses before requeueing a @@ -251,6 +257,15 @@ func (c *RabbitMQConsumer) connect(ctx context.Context, out chan<- Message) erro c.closeNotify = closeNotify c.mu.Unlock() + // One series per queue even though a single connection carries all of + // them, so the same alert works against either backend - see + // metrics.QueueConnected. Set only now, with every queue declared and + // consuming, because a partial connect returned above without ever + // getting here. + for queueName := range c.router { + metrics.QueueConnected.WithLabelValues(queueName).Set(1) + } + return nil } @@ -378,6 +393,12 @@ func (c *RabbitMQConsumer) superviseReconnects(ctx context.Context, out chan<- M default: } c.reconnects.Add(1) + // Every queue at once: they share the connection that just + // went away, so every one of them has stopped consuming. + for queueName := range c.router { + metrics.QueueConnected.WithLabelValues(queueName).Set(0) + metrics.QueueReconnectsTotal.WithLabelValues(queueName).Inc() + } slog.Warn("rabbitmq: connection lost, reconnecting", "error", amqpErr, "retry_interval", reconnectDelay) } @@ -528,6 +549,10 @@ func (c *RabbitMQConsumer) Stop() error { } } + for queueName := range c.router { + metrics.QueueConnected.WithLabelValues(queueName).Set(0) + } + slog.Info("rabbitmq: consumer stopped", "processed", c.processed.Load(), "errors", c.errors.Load(), "dropped", c.dropped.Load(), "reconnects", c.reconnects.Load()) diff --git a/internal/queue/registry.go b/internal/queue/registry.go index 40b1036..33506bb 100644 --- a/internal/queue/registry.go +++ b/internal/queue/registry.go @@ -3,6 +3,7 @@ package queue import ( "context" "database/sql" + "errors" "fmt" "log/slog" "time" @@ -33,6 +34,49 @@ const ( QueueCoreRestart = "statusngin_core_restart" ) +// inOrderQueues names the queues whose messages build on one another, so a +// later message must never be handled before an earlier one. A consumer is +// required to process these strictly one at a time. +// +// statusngin_downtimes is the only one, and it is not a preference. A +// single downtime arrives as up to four separate messages over its +// lifetime - ADD, START, STOP, DELETE (see +// .claude/specs/downtime_ablauf.txt) - and only ADD is an UPSERT. START and +// STOP are bare UPDATE ... WHERE against the row ADD created, and +// STOP/DELETE are DELETE FROM scheduleddowntimes. Every one of those is a +// silent no-op if it runs before the message it depends on. +// +// This is not theoretical. Measured against a shadow-tested database after +// the worker had been disconnected from the job server for two hours and +// was restarted with the whole backlog waiting: of ~14 downtimes, 6 were +// corrupted, each one explainable by the order its messages happened to be +// executed in. START before ADD left was_started=0 and actual_start_time=0 +// on a row whose actual_end_time was correct, because STOP got there after +// ADD. STOP before START left a scheduleddowntimes row behind for good, +// because STOP deleted it and START's UPSERT put it back. Nothing was +// logged, nothing failed, and the only reason it was found at all is that +// cmd/db_verifier diffs against the legacy PHP worker. +// +// Live traffic hides this completely - a downtime's events are minutes +// apart, so they never overlap. It takes a backlog, which is exactly what +// an outage produces. +// +// The legacy PHP worker forks one process per queue and is sequential, and +// the RabbitMQ consumer here calls its Handler synchronously from a plain +// for range (TestRabbitMQPrefetchDoesNotAddConcurrency pins that), so both +// of those are already in order. Only the Gearman consumer dispatches +// concurrently, and it never meant to make this queue an exception. +var inOrderQueues = map[string]bool{ + QueueDowntimes: true, +} + +// RequiresInOrderProcessing reports whether queueName's messages must be +// handled strictly one at a time. Consumers that dispatch concurrently have +// to honour it; see inOrderQueues for why. +func RequiresInOrderProcessing(queueName string) bool { + return inOrderQueues[queueName] +} + // isHardState maps the standard Nagios/Icinga/Naemon state_type convention // (0 = SOFT, 1 = HARD) to the tinyint(1) is_hardstate column. func isHardState(stateType int) int { @@ -451,19 +495,12 @@ func downtimeMetricsTables() []string { return tables } -// execDowntimeAction turns one DowntimeAction into its concrete (query, -// args) pair via the matching internal/db builder (Schritt 3), executes it, -// and reports the outcome through the same db-write instrumentation -// BulkInserter.flushBuffer uses for every other table - done by hand here -// since downtime writes deliberately bypass BulkInserter entirely (see -// .claude/specs/downtime_ablauf.txt section 6: a single downtime message -// can require an UPSERT, UPDATE or DELETE, not just an INSERT). Unlike -// BulkInserter's batch histograms (which track 100-item/250ms batching -// behaviour that plainly doesn't apply here, per downtime_ablauf.txt -// section 6), only DBEventsWrittenTotal/PipelineErrorsTotal apply to a -// single-row ExecContext like this one. -func execDowntimeAction(ctx context.Context, sqlDB *sql.DB, action DowntimeAction) error { - row := db.DowntimeRow{ +// downtimeRowFor projects a DowntimeAction's payload onto the db-layer row +// every downtime query builder takes. Shared by the statement itself and by +// the follow-up existence check, so both can never disagree about which row +// they mean. +func downtimeRowFor(action DowntimeAction) db.DowntimeRow { + return db.DowntimeRow{ IsHostDowntime: action.Data.IsHostDowntime, HostName: action.Data.HostName, ServiceDescription: action.Data.ServiceDescription, @@ -483,6 +520,113 @@ func execDowntimeAction(ctx context.Context, sqlDB *sql.DB, action DowntimeActio ActualEndTime: action.Data.ActualEndTime, WasCancelled: action.Data.WasCancelled, } +} + +// reportUnmatchedDowntimeUpdate says so when a downtime UPDATE matched no +// row, which MySQL reports as success and which nothing else would ever +// surface. +// +// START and STOP are written as bare UPDATE ... WHERE against the row +// the downtime's ADD created (see .claude/specs/downtime_ablauf.txt). If +// that row is not there the statement affects nothing, returns no error, +// and the event is lost - was_started stays 0 on a downtime that ran, or +// actual_end_time stays 0 on one that ended. That is precisely how +// concurrent handling of the downtime queue corrupted six downtimes across +// one job-server outage without producing a single log line; the queue is +// serialized now (see inOrderQueues), and this is what makes a recurrence +// visible rather than something cmd/db_verifier finds weeks later. +// +// Only the two UPDATE actions are checked. A DELETE matching nothing is +// normal and happens on every ordinary downtime - STOP removes the +// scheduleddowntimes row, and the DELETE that follows finds it already +// gone - so counting that would bury the real signal in noise. An UPSERT +// always matches by definition. +// +// Zero affected rows is not on its own enough to report, which is why this +// costs a second query. MySQL counts rows *changed*, not matched +// (CLIENT_FOUND_ROWS is off in go-sql-driver by default), so a redelivered +// START rewriting the values it already wrote also reports zero - and +// redelivery is normal here, not exceptional (CLAUDE.md rule 6). Asking +// whether the row exists separates the two, and it only ever runs on a path +// that is supposed to be dead, so its cost is irrelevant. +// +// The answer is only as good as the serialization it rests on, and that is +// worth knowing rather than assuming. If something else writes that row +// between the UPDATE and this lookup, a genuinely lost event reads as a +// redelivery and goes uncounted - measured, by running this check against a +// deliberately unserialized downtime queue: the ADD racing behind its own +// START landed in exactly that window, the corruption appeared in MySQL and +// this counter stayed at 0. Not a gap in practice, because the queue that +// produces these is serialized (see inOrderQueues) and the only other +// writer would be a second worker process, which CLAUDE.md rule 2 already +// advises against for unrelated reasons. It does mean this counter +// backstops a missing ADD rather than a broken ordering guarantee - +// TestDowntimeQueueIsHandledOneAtATime is what covers the latter. +// +// A driver that cannot report RowsAffected is not treated as a finding: the +// point is to detect a missing row, not to complain about the driver. +func reportUnmatchedDowntimeUpdate(ctx context.Context, sqlDB *sql.DB, result sql.Result, action DowntimeAction, table string) { + if action.Action != DowntimeActionUpdateStarted && action.Action != DowntimeActionUpdateStopped { + return + } + + affected, err := result.RowsAffected() + if err != nil || affected != 0 { + return + } + + query, args := db.DowntimeHistoryExistsQuery(downtimeRowFor(action)) + var exists int + switch err := sqlDB.QueryRowContext(ctx, query, args...).Scan(&exists); { + case err == nil: + // The row is there and already held these values - a redelivery + // doing its job, which is the whole reason for this second look. + return + case !errors.Is(err, sql.ErrNoRows): + // Could not find out. Reporting a lost event on a failed lookup + // would be worse than staying quiet, since this counter is only + // useful if a non-zero value is trustworthy. + slog.Debug("queue: could not check whether the downtime row exists", + "table", table, "action", action.Action, "error", err) + return + } + + metrics.DowntimeUpdatesUnmatchedTotal.WithLabelValues(table).Inc() + slog.Warn("queue: downtime update matched no row, the event is lost", + "table", table, "action", action.Action, + "host", action.Data.HostName, "service", action.Data.ServiceDescription, + "internal_downtime_id", action.Data.InternalDowntimeID, + "scheduled_start_time", action.Data.ScheduledStartTime, + "note", "no row for this downtime existed to update - its ADD is missing or arrived later") +} + +// downtimeHistoryTables enumerates the two tables a downtime UPDATE can +// target. A strict subset of downtimeMetricsTables: the scheduleddowntimes +// pair is only ever upserted or deleted, never updated, so pre-creating an +// unmatched-update series for those two would advertise a failure mode they +// do not have. Built from the same helper as everything else here so the +// four names cannot drift apart; a test pins the subset relationship. +func downtimeHistoryTables() []string { + tables := make([]string, 0, 2) + for _, scope := range []string{"host", "service"} { + tables = append(tables, downtimeTableName(scope, DowntimeHistoryTable.String())) + } + return tables +} + +// execDowntimeAction turns one DowntimeAction into its concrete (query, +// args) pair via the matching internal/db builder (Schritt 3), executes it, +// and reports the outcome through the same db-write instrumentation +// BulkInserter.flushBuffer uses for every other table - done by hand here +// since downtime writes deliberately bypass BulkInserter entirely (see +// .claude/specs/downtime_ablauf.txt section 6: a single downtime message +// can require an UPSERT, UPDATE or DELETE, not just an INSERT). Unlike +// BulkInserter's batch histograms (which track 100-item/250ms batching +// behaviour that plainly doesn't apply here, per downtime_ablauf.txt +// section 6), only DBEventsWrittenTotal/PipelineErrorsTotal apply to a +// single-row ExecContext like this one. +func execDowntimeAction(ctx context.Context, sqlDB *sql.DB, action DowntimeAction) error { + row := downtimeRowFor(action) var query string var args []any @@ -508,7 +652,7 @@ func execDowntimeAction(ctx context.Context, sqlDB *sql.DB, action DowntimeActio table := downtimeMetricsTable(action) start := time.Now() - _, err := sqlDB.ExecContext(ctx, query, args...) + result, err := sqlDB.ExecContext(ctx, query, args...) duration := time.Since(start) if err != nil { @@ -517,6 +661,8 @@ func execDowntimeAction(ctx context.Context, sqlDB *sql.DB, action DowntimeActio return fmt.Errorf("%s %s: %w", action.Action, table, err) } + reportUnmatchedDowntimeUpdate(ctx, sqlDB, result, action, table) + metrics.DBEventsWrittenTotal.WithLabelValues(table).Add(1) // Debug, for the same reason as the bulk-insert flush line: one entry // per row written. Downtimes are normally low-volume, but scheduling @@ -857,6 +1003,9 @@ func NewRouter(sqlDB *sql.DB, hub *websocket.Hub, gc *graphite.Client, perfdataR for _, table := range downtimeMetricsTables() { metrics.InitTable(table) } + for _, table := range downtimeHistoryTables() { + metrics.InitDowntimeUpdates(table) + } runners := []Runner{ hostStatus, serviceStatus,