Fix/gearman reconnect

.gitignore

@@ -1,3 +1,5 @@
+.claude/settings.json
+.ressouces
 /simulator
 /gearman_publisher
 /rabbitmq_publisher

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 <PK>` 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.

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 <PK>` 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

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