mirror of
https://github.com/civitai/civitai.git
synced 2026-09-20 22:08:18 +08:00
Justin Maier
d4dc1058e6
* fix(notifications): bust the unread-count cache for rows cleanup deletes cleanupNotifications batch-deleted UserNotification rows without touching the per-user unread-count cache, so a badge kept reporting notifications whose rows were already gone — "Updates 2" over a list saying "All caught up" — until a mark-read or the one-week TTL cleared it. The delete now RETURNs "userId", viewed and busts the count cache for the users whose deleted row was unread. Read rows are skipped: the cached hash only holds unread counts, so deleting a read row cannot make it wrong (~35% of a batch on prod data). bustUser rather than decrementUser so the next count re-derives from the DB instead of drifting. The lag window is flagged before each bust, as markReadImpl does, so a count landing before the replica catches up cannot cache the not-yet-deleted rows for another week. Measured on the prod replica: a 10k-row batch carries ~8.4k distinct users, ~5.5k of them with an unread row, so the bust side costs ~0.55 Redis DELs per deleted row. Busts are deduped within a batch only — not across the sweep, because a user who reads their bell mid-sweep re-populates the cache from rows a later batch then deletes, and a sweep-wide dedupe would skip that second bust and re-create this bug. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(notifications): say what the lag flag actually closes, and count the busts that landed Review findings on the previous commit. The comment claimed the lag flag stops a count from caching the pre-delete number. It does not: a count that has already picked the replica pool is unaffected, and its setUser still lands after the bust. The flag narrows the window to counts that start after it — the same exposure markReadImpl carries — so the comment now says that instead of claiming closure. The sweep log reported users attempted rather than keys dropped, so a redis outage that dropped none would still log a full sweep. It now counts the busts that resolved. Tests: the redis-failure case rejected only bustUser, so the swallow on the lag flag — the call that fails FIRST in an outage, and would otherwise abort the whole sweep — was untested. It now rejects both and asserts the later user is still busted. Added a batch wider than CLEANUP_BUST_CONCURRENCY so the worker pool's cursor takes its second iteration in a test rather than never, and an assertion on the logged bust count. beforeEach resets the mocks instead of clearing them: a mockRejectedValueOnce survives mockClear and would poison the next test's first bust. Also dropped the undated prod measurements from the comments (they belong in the PR body, where they cannot rot silently) and stopped defending resp.rows against an undefined node-pg never returns. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(notifications): say what the sweep's bust count actually counts, and pin the concurrency cap Second review round on the same change. The log field claimed to count keys dropped. It counts DELs redis acknowledged: bustUser discards the reply, redis acks a DEL for an absent key, and most swept users have no cached count at all — so the number is acks, and it is now named bustsAcked and says so. It still does the job it was added for, which is to stop a redis outage reporting a fully-busted sweep. The lag paragraph asserted the primary-read narrowing unconditionally and then mentioned four lines later that the flag no-ops when REPLICATION_LAG_DELAY is unset. Those cannot both describe production: with the tracker disabled the narrowing is zero and the bust is the only thing working. Said so, and pointed at L5 in docs/plans/notifications-review-action-items.md, which owns deciding that value. Tests: added one that counts busts in flight and asserts the peak is exactly CLEANUP_BUST_CONCURRENCY. Nothing else in the file could see the cap — replace the pool with an unbounded Promise.all and every other assertion still passed, while the cap is what stands between one batch and thousands of simultaneous redis round-trips. Sorted the ids in the redis-rejection assertion, which depended on worker continuation order. Corrected the mock comment, which described a protection the bare vi.fn() stubs do not provide. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test(notifications): close nine mutants the cleanup suite let through Round 3 asked two lanes for mutants that go GREEN rather than for prose. They produced nine distinct ones and all nine passed the suite as it stood. None of them is a hypothetical: the SQL predicate, the RETURNING list, the dedupe order and the cross-batch accumulation were all unasserted, so the tests verified the bust machinery thoroughly and the delete itself not at all. Closed, each verified by applying the mutant and watching a named assertion fail: - The sweep could delete the NEWEST rows (`"createdAt" >`), or none at all (`LIMIT 0`), and every test passed — the fake pool answers any query with the rows the test programmed. Now pins the predicate and the limit. - `RETURNING "userId", viewed IS FALSE AS viewed` CONTAINS the old toContain string, and inverts the filter so cleanup busts the read users and nobody else — the original bug, restored, green. The guard is anchored now. - Deduping before the unread filter drops any user whose first returned row is read. DELETE ... RETURNING yields rows in physical order, so that was a coin flip per affected user. The fixture now puts the read row first. - `bustsAcked` accumulates across batches; one busting batch could not tell `=` from `+=`. The log assertion now spans two batches. - Flagging `userIds[0]` for the whole batch left every other user without the primary-read narrowing; the ordering test has one user, so it could not see "each user". The wide test now pins the flag's argument per user. - The concurrency cap is per sweep and covers both round-trips. Hoisting the lag flags out of the pool, or dropping the await on the bust pass so batches overlap, both left the DEL half at 25 and passed. The peak test now counts both calls across two batches. - A bust pass skipped after the first full batch passed everything, because no fixture came near CLEANUP_BATCH_SIZE. It is exported for test visibility and there is a full-batch test. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test(notifications): pin the whole DELETE, not three fragments of it Round 4 produced six more green mutants, three of them created by round 3's own closures — which is the argument for the change this commit makes. The fragment guards were each substring-blind in a different direction. `LIMIT 10000` passed `LIMIT 100000`. `"createdAt" < $1` passed `"createdAt" < $1 AND viewed`, a sweep that deletes only READ rows and busts nobody while reporting a healthy `deleted`. Neither guard saw the table name, so the subquery could be aimed at "Notification" and delete UserNotification rows by id collision. And the limit guard interpolated CLEANUP_BATCH_SIZE, so both sides moved together and the constant could be set to 1 — a sweep that cannot finish inside the client's timeout — with every assertion still agreeing. So the statement is now asserted whole, by equality, with every literal spelled out. That is a golden string rather than a behavioural check, and the comment says so: the fake pool executes no SQL, so this is the only thing between the suite and a cleanup aimed at the wrong rows. Two behavioural closures beside it: - The full-batch fixture now carries CLEANUP_BATCH_SIZE DISTINCT users and asserts the bust count. All-one-user dedupes to a single bust, which let a truncated bust list — `userIds.splice(1000)`, the shape of a plausible per-batch cap — pass while stranding ~88% of a real batch. - A batch wider than the pool with EVERY bust rejecting. The narrow fixtures gave each user its own worker, so no user was ever reached after a failure; a worker that gave up on error retired and the users past the 25th were never attempted. Also pinned: the lag flag repeats across batches for the same user. Its TTL is REPLICATION_LAG_DELAY seconds and a sweep runs for minutes, so a sweep-wide "already flagged" cache would leave every later batch of that user's rows reading the replica. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test(notifications): pin two decisions a later reader would correctly undo Both are things a reviewer raised as tempting to "fix", and neither is recoverable from the code. The peak test hardcodes 25 while its title names CLEANUP_BUST_CONCURRENCY, which reads as an inconsistency — but asserting toBe(CLEANUP_BUST_CONCURRENCY) would agree with the source at every width, including no cap at all. Same shape as the LIMIT guard that let a tenfold batch size through: a guard must not read its expected value from the thing under test. And the two fullness tests look redundant. They are not: one covers a gate that stops on a FULL batch, the other a gate that stops on a short one, and deleting either opens its half. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test(notifications): let the fake database fail, and tell the two pools apart Round 5 went after surface rather than spelling, and found the seams the SQL assertion cannot reach: the fake pool could not fail, and it could not be told apart from the read pool. - `notifDbWrite()` -> `notifDbRead()` was a one-word green mutation, because the mock returned the SAME object for both. In an environment with NOTIFICATION_DB_REPLICA_URL set, that is every batch failing with "cannot execute DELETE in a read-only transaction" — and it is a silent no-op anywhere without a replica, so it works wherever you would test it and dies where it matters. The mock now gives the read pool its own object, which throws. - The fake could not reject, so the whole query error path was unobserved: swallowing a transient postgres error would end a sweep early, log a healthy `deleted`, and hand the admin endpoint a 200 while rows accumulated nightly. There is now a failure queue and a test that the error comes out. - Only `captured[0]` was ever asserted, so batches 2..N were unconstrained in both statement and cutoff. Now asserted across every call of a 60-batch sweep, which also puts a floor under a "runaway" pass cap — the loop's real guarantee is that it exits on an empty batch, and a cap above 60 is still invisible. - The logger can now reject, pinning the `.catch` on a call made without await: an unhandled rejection there takes the process down AFTER the deletes have happened, so the caller sees a transport failure and retries the whole sweep. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
apps/notifications
The notification domain, in-repo. Folds the external civitai/notification-server repo into the
monorepo as a peer of apps/auth / apps/orchestrator-gateway, consuming the shared @civitai/*
packages instead of its own forked DB/redis/axiom plumbing. See the plan:
docs/plans/notifications-monorepo-migration.md.
What it does
One process owns the whole notification write side:
- (A) Producer API —
POST /notifications, authed (shared secret, internal-only ingress). Validates the@civitai/notificationsschema, filters opted-out recipients against the main DBUserNotificationSettings(reachable now that we're in-repo), and UPSERTs thePendingNotificationqueue row. This is net-new surface — the external server was GET-only. - (A2) Bulk producer —
POST /notifications/bulk, same auth. Takes pre-resolved recipients and applies no opt-out filter; eachsend-notificationsprocessor filters in its own SQL. This is the pathsrc/server/jobs/send-notifications.tsuses, and it is why a processor missing itsNOT EXISTS "UserNotificationSettings"clause is unmuteable. - (B) Fan-out worker — the ported ~5s poll loop. Claims
PendingNotificationrows, fans each intoNotification+UserNotificationrows (normal / debounced), deletes/reschedules the pending row, and POSTs a realtimenotification:newsignal per affected user while bumping the redis unread counter. - Ops routes —
GET /health(no-dep liveness),GET /pool-stats(notif pool snapshots),GET /metrics(Prometheus, private-by-XFF).
The producer↔consumer contract is the PendingNotification table, not HTTP — so the monolith's
existing direct writes and this app's producer API can coexist during the transition (plan R3).
Shape
Node + Fastify + tsup, same as apps/orchestrator-gateway. src/app.ts is the testable Fastify
factory; src/server.ts adds listen() + startWorker(). Clients are thin shims over the shared
packages (src/lib/server/clients/{db,redis,axiom}.ts).
Dev
cp .env.example .env # fill in the DB / redis / signals values
pnpm --filter @civitai/notifications-app dev # tsx watch (API + worker)
pnpm --filter @civitai/notifications-app typecheck
pnpm --filter @civitai/notifications-app build # tsup → dist/server.js
pnpm --filter @civitai/notifications-app test
Unlike the auth/gateway images, this app imports @civitai/db's Prisma-backed helper, so a Docker
build needs the generated Prisma client — the Dockerfile does a full frozen install so the root
postinstall runs db:generate (see the Dockerfile header).
Not done here (follow the plan)
Deploy wiring (datapacket-talos app dir, Tekton APP_CONFIG entry, lockfile refresh, worker cutover
soak/canary) and the read-path move (C) are separate, sequenced steps — see §5 of the plan.