Fix/statusengine schema primary keys

CLAUDE.md

@@ -9,7 +9,7 @@ A highly performant, concurrent Go-based event pipeline. It consumes bulk JSON m
 - **Legacy Reference:** https://github.com/statusengine/worker (Reference for domain logic, event types, and processing rules)
 
 ## System Context & Specs
-- **Database Schema:** Read `/.claude/specs/mysql_schema.sql`
+- **Database Schema:** Read `/.claude/specs/mysql_schema.sql` - **this is openITCOCKPIT's schema, not standard Statusengine's, and the PRIMARY KEYs differ.** openITCOCKPIT stores UUIDs in `hostname`/`service_description`, so a service description is unique on its own and four service tables key on it without `hostname`: `statusengine_servicechecks`, `statusengine_service_statehistory`, `statusengine_service_notifications` and `statusengine_service_acknowledgements`. Standard Statusengine keeps the plain description and leads all four keys with `hostname` (`setPrimaryKey` in [lib/mysql.php](https://github.com/statusengine/worker/blob/master/lib/mysql.php)); it also keys `statusengine_logentries` on `id` alone where the dump has `(id, entry_time)`, which is partitioning rather than UUIDs. Everything else matches, no UNIQUE index exists in either, and every one of those tables carries a `hostname` column in both - which is what makes the difference survivable. Two consequences, both already handled: rule 6's upsert columns must be in the key under *either* schema, and anything matching a row across two databases (`cmd/db_verifier`) must key on `hostname` **and** `service_description`, which is unique in both
 - **Queue Payload Examples:** Read JSON dumps in `/.claude/specs/` (Note: Each queue delivers a specific type, but payloads arrive as a JSON bulk array).
 - **Queue Payload Bulk Exceptions:** The Queues `statusngin_acknowledgements`, `statusngin_contactnotificationmethod.json`, `statusngin_core_restart.json` and `statusngin_downtimes` do not use bulk payloads.
 - **WebSocket/Metrics API Reference:** `docs/openapi.yaml` (OpenAPI 3.1, viewable via `docs/index.html` with Scalar) - the `/ws` message envelope and a real captured example for every event topic, including the `statusngin_downtimes` `type`/`attr` lifecycle table.
@@ -88,7 +88,7 @@ Every queue is wired end-to-end in `internal/queue/registry.go`'s `NewRouter` -
 ### 6. Stability & Lifecycle
 - Explicitly handle all errors, reconnect automatically to MySQL/Queues on connection drops.
 - Implement full **Graceful Shutdown**: Catch OS signals (SIGINT/SIGTERM), stop the consumer, flush all remaining items from the 250ms DB buffers, cleanly close active WebSocket connections, and then exit.
-- **Delivery is at-least-once, so every write that can collide is idempotent.** This is not a tuning detail, it is what stops a restart from losing data. A worker that is killed, OOM-killed or loses power between finishing a job and its acknowledgement reaching the broker gets that job again on restart, with its rows already in MySQL. Before this was handled, one SIGTERM under load lost **3,300 of 300,000 events (1.1%)**, measured with `cmd/losstest`: the redelivered rows collided on the PRIMARY KEY, MySQL aborted the *entire* multi-row INSERT with `Error 1062`, and `flushBuffer` dropped the batch - taking with it every fresh row that happened to share it, since batches are cut at `MaxBatchSize` regardless of job boundaries. 97 batches failed, all `rows=100`: 9,700 dropped rows = 6,400 harmless duplicates + 3,300 unrecoverable fresh events. The shutdown flush was never the cause; two uninterrupted control runs over the same path lost nothing. The fix is `newRedeliverySafeInserter` in `internal/queue/registry.go`: the ten tables with a natural PRIMARY KEY are built as upserts whose update clause names the **first column of that key**, which makes it a genuine no-op (the row only matched because that column is already equal). `redeliverySafePKColumn` is the single source of that mapping and is cross-checked against `.claude/specs/mysql_schema.sql` by `TestRedeliverySafePKColumnsMatchSchema`; a table wired up without an entry panics at construction rather than silently reverting to a plain INSERT. **The no-op argument has one unstated premise, and `TestUpsertTablesHaveNoSecondaryUniqueIndex` is what states it: the PRIMARY KEY must be the table's only unique index.** `ON DUPLICATE KEY UPDATE` fires on a violation of *any* unique index, so a secondary UNIQUE one would make it match a row the primary key does not identify - dropping the incoming row and overwriting an unrelated row's key column, both without an error and without a log line (and for the two status tables, whose update clause covers every data column, writing one object's status over another's). The schema carries no UNIQUE index today, which is exactly why the guard has to exist before one appears: adding an index is an ordinary schema change nobody would connect to this file. The test derives its 16 tables from `NewRouter`'s own inserters (`BulkInserter.IsUpsert`) plus `downtimeMetricsTables()` for the four that bypass BulkInserter but still upsert, so a table added later is covered without a second list. `INSERT IGNORE` was deliberately not used - it downgrades *every* error to a warning, including truncation and NOT NULL violations. Re-measured after the fix: 300,000 of 300,000, zero `Error 1062`.
+- **Delivery is at-least-once, so every write that can collide is idempotent.** This is not a tuning detail, it is what stops a restart from losing data. A worker that is killed, OOM-killed or loses power between finishing a job and its acknowledgement reaching the broker gets that job again on restart, with its rows already in MySQL. Before this was handled, one SIGTERM under load lost **3,300 of 300,000 events (1.1%)**, measured with `cmd/losstest`: the redelivered rows collided on the PRIMARY KEY, MySQL aborted the *entire* multi-row INSERT with `Error 1062`, and `flushBuffer` dropped the batch - taking with it every fresh row that happened to share it, since batches are cut at `MaxBatchSize` regardless of job boundaries. 97 batches failed, all `rows=100`: 9,700 dropped rows = 6,400 harmless duplicates + 3,300 unrecoverable fresh events. The shutdown flush was never the cause; two uninterrupted control runs over the same path lost nothing. The fix is `newRedeliverySafeInserter` in `internal/queue/registry.go`: the ten tables with a natural PRIMARY KEY are built as upserts whose update clause names **a column of that key**, which makes it a genuine no-op (the row only matched because every key column is already equal). Membership, not position - and the distinction is load-bearing rather than pedantic, because the two schemas order these keys differently (see the schema note above). The four service tables therefore name `service_description`, the one column inside the key under both; `hostname` reads as the natural choice from the standard-Statusengine keys and is **not** in openITCOCKPIT's, where it would turn every redelivered row into a real write. `redeliverySafePKColumn` is the single source of that mapping, and `TestRedeliverySafePKColumnsMatchSchema` checks every entry for membership in both keys - against `.claude/specs/mysql_schema.sql` for openITCOCKPIT's and against the standard keys transcribed beside it - so it fails on exactly that mistake. A table wired up without an entry panics at construction rather than silently reverting to a plain INSERT. **The no-op argument has one unstated premise, and `TestUpsertTablesHaveNoSecondaryUniqueIndex` is what states it: the PRIMARY KEY must be the table's only unique index.** `ON DUPLICATE KEY UPDATE` fires on a violation of *any* unique index, so a secondary UNIQUE one would make it match a row the primary key does not identify - dropping the incoming row and overwriting an unrelated row's key column, both without an error and without a log line (and for the two status tables, whose update clause covers every data column, writing one object's status over another's). The schema carries no UNIQUE index today, which is exactly why the guard has to exist before one appears: adding an index is an ordinary schema change nobody would connect to this file. The test derives its 16 tables from `NewRouter`'s own inserters (`BulkInserter.IsUpsert`) plus `downtimeMetricsTables()` for the four that bypass BulkInserter but still upsert, so a table added later is covered without a second list. `INSERT IGNORE` was deliberately not used - it downgrades *every* error to a warning, including truncation and NOT NULL violations. Re-measured after the fix: 300,000 of 300,000, zero `Error 1062`.
 - **The redelivery itself is gone too, as of gearman-go v1.1.1 - and that does not make the upserts redundant.** The 64 re-queued jobs were not the broker being pessimistic: the library cleared its `running` flag and dropped the connections before waiting for in-flight handlers, and it writes a job's WORK_COMPLETE only while that flag is set, so those handlers wrote their rows and then silently skipped the acknowledgement. Fixed in the fork by draining before disconnecting (see rule 1). Verified by arithmetic rather than by absence of errors: `processed` plus the jobs left at the broker now sums to exactly what was published (3,000), where it was reliably 3,064 - the 64 being the concurrency cap. The two fixes cover different halves and neither replaces the other: this one stops redelivery on an *orderly* shutdown (the RabbitMQ consumer reached the same place from the other direction, by cancelling its consumers before closing anything - see rule 1), the upserts keep it harmless when it happens anyway - a crash, an OOM-kill, a lost acknowledgement on the network. Exactly-once is not achievable here, so the idempotent write is the load-bearing part; removing it because "redelivery no longer happens" would reintroduce the 1.1% loss on the first hard kill.
 - **Two tables are knowingly not covered:** `statusengine_logentries` (AUTO_INCREMENT key) and `statusengine_perfdata` (no PRIMARY KEY at all) cannot collide, so a redelivery inserts their rows a **second time, silently** - no error, nothing in the log. Accepted rather than fixed: both are retention-managed history, where a duplicate row is less harmful than a missing event, and closing it would need a UNIQUE index - a schema change with its own migration and index cost on the two highest-volume tables in the database. `TestExcludedTablesReallyCannotCollide` fails if that ever changes.
 - **Known metric side effect:** `DBEventsWrittenTotal` counts every buffered row as written, including duplicates the upsert skipped, so `db_events_written_total` briefly overstates by up to 6,400 after a restart under load.
@@ -104,4 +104,4 @@ Every queue is wired end-to-end in `internal/queue/registry.go`'s `NewRouter` -
 - **Deployment:** unit files in `packaging/systemd/` (`make install`, `make install-systemd`). One value there is load-bearing: **`TimeoutStopSec` must stay above 45s** - 30s Gearman drain (`DrainTimeout`, closed in parallel) + 10s `shutdownFlushTimeout` + 5s HTTP shutdown. systemd `SIGKILL`s when it expires, so a shorter value kills the worker mid-flush and loses exactly the buffered rows the graceful shutdown exists to write, plus the acknowledgements, which then makes rule 6's upserts the only thing preventing duplicates. It is set to 90s explicitly rather than left to systemd's default, so raising `DrainTimeout` has a place to be reflected. The units run with `ProtectSystem=strict`, which the worker tolerates because it writes nothing to disk. API keys come from an `EnvironmentFile`, never `ExecStart` - a command line is world-readable via `/proc`. The cleanup timer uses `Persistent=true` so a missed day is caught up; enable it on **one** node in a cluster
 - **Build DB Cleanup:** `go build -o bin/db_cleanup cmd/db_cleanup/main.go` - one-shot retention tool for cron/systemd timers that deletes rows older than a configured number of days from the 14 history tables; same config mechanism and same YAML file as the worker (`-config`), retention per table in days via the legacy `age_*` keys (`0` disables a table), plus `-cleanup-batch-size` (5000) and `-cleanup-batch-pause` (0s); stops cleanly between batches on SIGTERM, exits non-zero only if a table failed; run with `go run ./cmd/db_cleanup -config config.example.yaml -mysql-dsn "statusengine-dev:statusengine-dev@tcp(127.0.0.1:3306)/statusengine-dev"`
 - **Build Loss Test:** `go build -o losstest cmd/losstest/main.go` - proves whether the worker loses events when stopped under load, which is the one thing about CLAUDE.md rule 6 that reading the code cannot establish. Publishes `statusngin_hostchecks` events whose hostname is a unique marker (`lt-<run-id>-<seq>`, and hostname is the first column of that table's PRIMARY KEY), so a missing sequence number is proof of a lost event rather than a deduplication artifact. Three modes (`-mode publish|verify|cleanup`, shared `-run-id`); `verify` reports missing sequence numbers as compressed ranges, asks the job server how many jobs are still queued (a backlog there is not loss), and exits 1 if anything is missing. Run the worker as a **built binary, not `go run`** - `go run` spawns the worker as a child process, so SIGTERM hits the parent and never reaches the worker. This tool found the redelivery data loss described under rule 6 below, and is what proves it stays fixed; re-run it before a release.
-- **Build DB Shadow Verifier:** `go build -o bin/db_verifier cmd/db_verifier/main.go` - read-only CLI that diffs the most recent rows of the legacy PHP worker's MySQL database against this Go worker's, column by column, to prove shadow-testing data parity (`-dsn-php`, `-dsn-go`, `-tables`, `-limit`, default `-tables` covers every Status/Check/History/Notification/Acknowledgement/Downtime table (excludes `statusengine_dbversion`/`statusengine_nodes`/`statusengine_perfdata`/`statusengine_tasks`/`statusengine_users`/`statusengine_logentries`, still selectable explicitly), default `-limit` 5000); run with `go run cmd/db_verifier/main.go -dsn-php "statusengine-dev:statusengine-dev@tcp(127.0.0.1:3306)/statusengine_php" -dsn-go "statusengine-dev:statusengine-dev@tcp(127.0.0.1:3306)/statusengine_go"`
+- **Build DB Shadow Verifier:** `go build -o bin/db_verifier cmd/db_verifier/main.go` - read-only CLI that diffs the most recent rows of the legacy PHP worker's MySQL database against this Go worker's, column by column, to prove shadow-testing data parity. Its `pkColumns` are a unique key rather than a copy of the PRIMARY KEY, because the two schemas disagree (see the schema note above): the four service tables key on `hostname` **and** `service_description`, a superset of openITCOCKPIT's key and exactly standard Statusengine's, so one binary works on both without a schema switch. Keying on `service_description` alone would collapse "PING on host A" and "PING on host B" onto one entry on a standard install and report mismatches between unrelated services (`-dsn-php`, `-dsn-go`, `-tables`, `-limit`, default `-tables` covers every Status/Check/History/Notification/Acknowledgement/Downtime table (excludes `statusengine_dbversion`/`statusengine_nodes`/`statusengine_perfdata`/`statusengine_tasks`/`statusengine_users`/`statusengine_logentries`, still selectable explicitly), default `-limit` 5000); run with `go run cmd/db_verifier/main.go -dsn-php "statusengine-dev:statusengine-dev@tcp(127.0.0.1:3306)/statusengine_php" -dsn-go "statusengine-dev:statusengine-dev@tcp(127.0.0.1:3306)/statusengine_go"`

cmd/db_verifier/main.go

@@ -30,11 +30,25 @@ const (
 )
 
 // tableSpec describes how to page through a table's most recent rows and
-// which columns uniquely identify a row across both databases: pkColumns
-// must match the table's actual PRIMARY KEY (see .claude/specs/mysql_schema.sql)
-// so that a row found in one database can be matched against its counterpart
-// in the other, and orderBy picks the columns that make "most recent" mean
-// something for that table (usually the PK's own time columns).
+// which columns uniquely identify a row across both databases: pkColumns has
+// to identify a row uniquely so that one found in one database can be matched
+// against its counterpart in the other, and orderBy picks the columns that
+// make "most recent" mean something for that table (usually the PK's own time
+// columns).
+//
+// pkColumns is a unique key rather than the PRIMARY KEY, because there are two
+// schemas and they disagree. .claude/specs/mysql_schema.sql is openITCOCKPIT's
+// variant, where service_description holds a UUID and is therefore unique on
+// its own, so four service tables key on it without hostname. Standard
+// Statusengine (lib/mysql.php in statusengine/worker, the setPrimaryKey calls)
+// keeps the plain service description there and leads those PRIMARY KEYs with
+// hostname. Keying on service_description alone would then collapse "PING on
+// host A" and "PING on host B" onto one map entry: one row silently replaces
+// the other and the comparison reports a mismatch against a row from a
+// different service. Naming hostname *and* service_description is a superset
+// of openITCOCKPIT's key and exactly Statusengine's, so it is unique under
+// both and needs no switch - the column exists in every one of these tables in
+// either schema.
 type tableSpec struct {
 	pkColumns []string
 	orderBy   []string
@@ -62,15 +76,15 @@ var tableSpecs = map[string]tableSpec{
 		orderBy:   []string{"start_time", "start_time_usec"},
 	},
 	"statusengine_servicechecks": {
-		pkColumns: []string{"service_description", "start_time", "start_time_usec"},
+		pkColumns: []string{"hostname", "service_description", "start_time", "start_time_usec"},
 		orderBy:   []string{"start_time", "start_time_usec"},
 	},
 	"statusengine_host_statehistory": {
 		pkColumns: []string{"hostname", "state_time", "state_time_usec"},
 		orderBy:   []string{"state_time", "state_time_usec"},
 	},
 	"statusengine_service_statehistory": {
-		pkColumns: []string{"service_description", "state_time", "state_time_usec"},
+		pkColumns: []string{"hostname", "service_description", "state_time", "state_time_usec"},
 		orderBy:   []string{"state_time", "state_time_usec"},
 	},
 	"statusengine_host_downtimehistory": {
@@ -90,7 +104,7 @@ var tableSpecs = map[string]tableSpec{
 		orderBy:   []string{"start_time", "start_time_usec"},
 	},
 	"statusengine_service_notifications": {
-		pkColumns: []string{"service_description", "start_time", "start_time_usec"},
+		pkColumns: []string{"hostname", "service_description", "start_time", "start_time_usec"},
 		orderBy:   []string{"start_time", "start_time_usec"},
 	},
 	"statusengine_service_notifications_log": {
@@ -102,7 +116,7 @@ var tableSpecs = map[string]tableSpec{
 		orderBy:   []string{"entry_time", "entry_time_usec"},
 	},
 	"statusengine_service_acknowledgements": {
-		pkColumns: []string{"service_description", "entry_time", "entry_time_usec"},
+		pkColumns: []string{"hostname", "service_description", "entry_time", "entry_time_usec"},
 		orderBy:   []string{"entry_time", "entry_time_usec"},
 	},
 	"statusengine_host_scheduleddowntimes": {