From 8a192dc68f86225d0488e829c9657a5daf8d74fb Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Fri, 23 Oct 2026 17:34:54 +0000 Subject: [PATCH] :recycle: Lock each consumer's pendings instead of the audit-log rows The drain locked the audit_log rows it reads, but the rows it writes are the pendings: each consumer owns its queue and deletes its own pendings. Locking the shared table made the consumers wait for each other, and the events consumer had to skip the locked rows with SKIP LOCKED to avoid that wait. The lock moves to audit_log_pending, so every consumer locks its own queue and no consumer waits for another one. The rows of audit_log are only read now, and every delete that takes them away (the drained chunk and the orphan sweep) skips the ones another transaction holds instead of waiting for them. The SKIP LOCKED variant of the chunk query is gone: the events have to be executed in order, like an event log, so a second run (another backend, or a retry overlapping the previous one) waits for the pendings of the run in flight and takes the next events. Skipping the locked ones would hand the newer events to the second run while the older ones were still in flight. The lock is FOR NO KEY UPDATE, not FOR UPDATE: deleting an audit_log row makes the foreign key check take FOR KEY SHARE on the pending rows that reference it, and those belong to the other consumers. FOR KEY SHARE does not conflict with FOR NO KEY UPDATE, so the check never waits for a drain in flight and the two consumers cannot deadlock. It still conflicts with itself, so two runs of the same consumer take turns. AI-assisted-by: minimax-m3.1-flash --- .serena/memories/audit-log.md | 8 +- .serena/memories/backend/subtleties.md | 2 +- backend/src/app/loggers/audit/consumers.clj | 76 ++++++++++--------- .../backend_tests/tasks_team_storage_test.clj | 45 +++++++++-- 4 files changed, 81 insertions(+), 50 deletions(-) diff --git a/.serena/memories/audit-log.md b/.serena/memories/audit-log.md index 34c45f9373..fbd8a87d18 100644 --- a/.serena/memories/audit-log.md +++ b/.serena/memories/audit-log.md @@ -277,15 +277,15 @@ Nexus dispatches on `:type`: `"telemetry-events"` batches are coerced and insert ### How it works today -Every 5 min each archive job drains its own queue (`:audit-log-consumer-nexus` on flag `:nexus`, cron `0 */5 * * * ?`; `:audit-log-consumer-admin-console` on flag `:admin-console`, cron `30 */5 * * * ?`; `:audit-log-consumer-storage-usage` on flag `:storage-usage`, cron `15 */5 * * * ?`; the offsets keep the three drains from starting on the same second), in 128-row chunks taken oldest-first (`ORDER BY created_at ASC, id ASC`) with `FOR UPDATE OF audit_log` (`SET LOCAL lock_timeout = '30s'`) inside the same transaction that sends and completes them, with 100 ms pauses between chunks. Batches arrive at Nexus in chronological order (it numbers by arrival for order-sensitive consumers). Nexus receives transit `{:events [...]}` with `x-shared-key: "nexus "` (`:nexus-shared-key`, or derived from the instance secret) and `origin` = public-uri; flag on but no URI raises `:task-not-configured`. The admin console receives only its allowlisted event names (enriched with team names and emails); anything filtered out counts as done for the admin-console queue. A chunk whose destination refuses it (Nexus non-204 raises `:consumer-chunk-refused`) or throws (transport failure, admin-console error) rolls back and stays for the next tick, and the job aborts as failed: it archives what it can, and the cron retries the rest. Committed chunks stay sent. The send fns return counts or throw carrying the attempted count as `:events`; `run-single-consumer` records the `failed` metric from it, outside every chunk transaction (a local DB failure carries no `:events` and stays uncounted — it is not a destination failure). No `warn` before the throw: the runner reports the aborted job. Each job touches only its own queue, so a fallen Nexus never starves the admin console (nor vice versa). +Every 5 min each archive job drains its own queue (`:audit-log-consumer-nexus` on flag `:nexus`, cron `0 */5 * * * ?`; `:audit-log-consumer-admin-console` on flag `:admin-console`, cron `30 */5 * * * ?`; `:audit-log-consumer-storage-usage` on flag `:storage-usage`, cron `15 */5 * * * ?`; the offsets keep the three drains from starting on the same second), in 128-row chunks taken oldest-first (`ORDER BY created_at ASC, id ASC`) with `FOR NO KEY UPDATE OF audit_log_pending` (`SET LOCAL lock_timeout = '30s'`) inside the same transaction that sends and completes them, with 100 ms pauses between chunks. Batches arrive at Nexus in chronological order (it numbers by arrival for order-sensitive consumers). Nexus receives transit `{:events [...]}` with `x-shared-key: "nexus "` (`:nexus-shared-key`, or derived from the instance secret) and `origin` = public-uri; flag on but no URI raises `:task-not-configured`. The admin console receives only its allowlisted event names (enriched with team names and emails); anything filtered out counts as done for the admin-console queue. A chunk whose destination refuses it (Nexus non-204 raises `:consumer-chunk-refused`) or throws (transport failure, admin-console error) rolls back and stays for the next tick, and the job aborts as failed: it archives what it can, and the cron retries the rest. Committed chunks stay sent. The send fns return counts or throw carrying the attempted count as `:events`; `run-single-consumer` records the `failed` metric from it, outside every chunk transaction (a local DB failure carries no `:events` and stays uncounted — it is not a destination failure). No `warn` before the throw: the runner reports the aborted job. Each job touches only its own queue, so a fallen Nexus never starves the admin console (nor vice versa). A chunk is **one POST** and the task advances only on 204, so a single oversized event wedges the whole queue: a crash report whose message embeds a whole payload (a serialized file, for example) can push the request past the proxy body limit and the rejection blocks every later event until it is pruned by hand. Frontend crash reports are therefore size-bounded at the source (`mem:frontend/routing-app-shell-subtleties`): `app.common.exceptions/get-hint` and the CLJS `format-throwable` clamp exception messages (`max-message-length`). On the other side, Nexus checks `origin` against its allowlist (403 if not allowed), coerces the events, and inserts them into its own `audit_log` (partitioned by `received_at`, with an incrementing sequence number and `source`). Watch the rename: Penpot's `created-at` is stored as **`received-at`** in Nexus; Nexus's `created-at` is when it was created there. `tracked-at` keeps the origin time. Telemetry batches arriving at `telemetry.penpot.app` land in the same table through the telemetry endpoint above. -Concurrent jobs are safe: the two archive consumers lock rows in `created_at ASC, id ASC` order (one total order, deliberately blocking — no `SKIP LOCKED`, so archive arrival stays chronological), so the second waits for the first to commit and then sees whether it was last. The storage-usage consumer is the exception: it archives nothing and its dirty mark is idempotent, so it takes its chunk with `FOR UPDATE … SKIP LOCKED` and never waits on an archive job holding the oldest rows (a wait could hit `lock_timeout` and abort the run). Only acknowledged work deletes pendings; an admin-console failure never blocks the Nexus send of the same rows (each job owns its queue). +Concurrent jobs are safe: every consumer locks its own pendings (`audit_log_pending`, the rows it owns and deletes) with `FOR NO KEY UPDATE OF p`, in `created_at ASC, id ASC` order — one total order for the whole stream, deliberately blocking, no `SKIP LOCKED`. A second run of the same consumer (another backend, or a retry overlapping the previous one) waits for the run in flight and then takes the next events: the events are executed in order, like an event log. No consumer waits for another one, because the queues are disjoint; the `audit_log` rows are only read, and every delete that takes them away — the drained chunk and the orphan sweep — skips the ones another transaction holds instead of waiting for them (the sweep runs outside the chunk transaction and without a `lock_timeout`, so waiting there would hang the whole drain). The lock mode is `FOR NO KEY UPDATE` and not `FOR UPDATE` on purpose: deleting an `audit_log` row makes the foreign key check take `FOR KEY SHARE` on the pending rows that reference it (they belong to the other consumers), and `FOR KEY SHARE` does not conflict with `FOR NO KEY UPDATE` — with `FOR UPDATE` the two consumers deadlocked. Only acknowledged work deletes pendings; an admin-console failure never blocks the Nexus send of the same rows (each job owns its queue). -Observability (`backend/src/app/loggers/audit/consumer_task.clj`, definitions in `backend/src/app/main.clj:default-metrics`): one `dbg` summary per job run (`:consumer` label, sent/skipped/chunks, reaped disabled pendings, swept orphans, `:elapsed` as `"1.5s"` via `ct/format-duration`, frozen in a strict binding first because log props evaluate lazily in the logger thread), one `dbg` per admin-console send call (`ready` with totals, `sent`/`failed` with counts), plus `dbg` when the queue is empty or the job's send flag is off. Queue depth is Grafana's business: it queries the database directly, so the run performs no `GROUP BY` for the logs. No `warn` before a throw: the runner reports the aborted job (refusal details travel in the error data). Metrics, all recorded with bounded labels only: `penpot_audit_consumer_events_total{ destination = nexus|admin-console|storage-usage, outcome = archived|processed|failed|skipped }` (`skipped` = filtered by the admin-console allowlist; storage-usage reports `processed` instead of `archived`: it marks teams dirty, it archives nothing) and `penpot_audit_consumer_discarded_total{ reason = reaped|swept }` (entries discarded without archiving: `reaped` counts pendings dropped for disabled consumers, `swept` counts rows deleted while nobody waited for them — the silent R3 loss, now dashboard-visible instead of log-only). Timing needs no custom histogram: the jobs substrate already records execution time per job (`jobs-execution-timing`, split by job name), so each archive job is timed as itself. No queue-depth gauge: pending depth is read off the replica database directly. The bulk frontend ingest counts `penpot_audit_ingest_events_total{ kind = full|shadow, outcome = inserted|failed }` (both faces share one transaction, so a failure drops the batch on both). Archive success counts are recorded by `run-single-consumer` on the drain totals: a chunk that returns normally committed its transaction (a rollback always throws instead), so they cannot inflate. Logs never carry event data (`props`, `context`, `profile-id`, `ip-addr`, `team-name`, emails): only counts, statuses and the uri host. +Observability (`backend/src/app/loggers/audit/consumers.clj`, definitions in `backend/src/app/main.clj:default-metrics`): one `dbg` summary per job run (`:consumer` label, sent/skipped/chunks, reaped disabled pendings, swept orphans, `:elapsed` as `"1.5s"` via `ct/format-duration`, frozen in a strict binding first because log props evaluate lazily in the logger thread), one `dbg` per admin-console send call (`ready` with totals, `sent`/`failed` with counts), plus `dbg` when the queue is empty or the job's send flag is off. Queue depth is Grafana's business: it queries the database directly, so the run performs no `GROUP BY` for the logs. No `warn` before a throw: the runner reports the aborted job (refusal details travel in the error data). Metrics, all recorded with bounded labels only: `penpot_audit_consumer_events_total{ destination = nexus|admin-console|events, outcome = archived|processed|failed|skipped }` (`skipped` = filtered by the admin-console allowlist, or nobody claimed the event; the events consumer reports `processed` instead of `archived`: it dispatches the stream, it archives nothing) and `penpot_audit_consumer_discarded_total{ reason = reaped|swept }` (entries discarded without archiving: `reaped` counts pendings dropped for disabled consumers, `swept` counts rows deleted while nobody waited for them — the silent R3 loss, now dashboard-visible instead of log-only). Timing needs no custom histogram: the jobs substrate already records execution time per job (`jobs-execution-timing`, split by job name), so each archive job is timed as itself. No queue-depth gauge: pending depth is read off the replica database directly. The bulk frontend ingest counts `penpot_audit_ingest_events_total{ kind = full|shadow, outcome = inserted|failed }` (both faces share one transaction, so a failure drops the batch on both). Archive success counts are recorded by `drain-consumer` on the drain totals: a chunk that returns normally committed its transaction (a rollback always throws instead), so they cannot inflate. Logs never carry event data (`props`, `context`, `profile-id`, `ip-addr`, `team-name`, emails): only counts, statuses and the uri host. ### How Nexus consumes the data (and forwards to PostHog) @@ -338,6 +338,6 @@ Config (`backend/src/app/config.clj`): `:nexus-uri`, `:telemetry-uri` (default ` - Penpot frontend producer: `frontend/src/app/main/data/event.cljs` (collector, buffer, send), `frontend/src/app/main/repo.cljs` + `frontend/src/app/util/http.cljs` (headers), `frontend/src/app/main/data/nitrate.cljs` (Nitrate membership events). - Admin-console producer (private repo, no paths cited here): server-side organization/subscription events through the management `push-audit-events`, browser-side UI events through the public `push-audit-events`. - Consumers: `backend/src/app/loggers/webhooks.clj`, `backend/src/app/loggers/database.clj`, `backend/src/app/loggers/mattermost.clj`, `backend/src/app/tasks/telemetry.clj`. -- Archive (three jobs sharing one namespace + pending queue, no GC): `backend/src/app/loggers/audit/consumer_task.clj` (`:audit-log-consumer-nexus` / `:audit-log-consumer-admin-console` / `:audit-log-consumer-storage-usage` job-defs, shared drain/reap/sweep helpers); the storage-usage counter itself is recomputed by `backend/src/app/tasks/team_storage.clj` (`:schedule-team-storage` on cron `45 */10 * * * ?` + `:recompute-team-storage` on the `:storage` queue) and read by `app.rpc.quotes`; its table is `team_storage_usage` (`team_id` PK/FK `ON DELETE CASCADE`, `total_bytes`, `dirty_at`, `updated_at`, partial index on `dirty_at IS NOT NULL`), created by migration `0158`, producer hook in `backend/src/app/loggers/audit.clj` (`append-audit-entry`, `queue-pendings` with explicit consumers per selectivity: Nexus default-gated on its flag, storage-usage and admin-console only for qualifying events) plus the bulk ingest in `backend/src/app/rpc/commands/audit.clj` (`handle-events`, same selectivity), metrics/cron wiring in `backend/src/app/main.clj` (one cron entry per job, Nexus at second 0, admin console at second 30, storage drain at second 15 and its scheduler every 10 min at second 45, each gated by its own send flag), single migration `0157` (queue, drop of `archived_at`, covering `(created_at, id)` index for the chronological scan; no backfill — pre-existing rows are swept unsent, an accepted loss). +- Archive (three jobs sharing one namespace + pending queue, no GC): `backend/src/app/loggers/audit/consumers.clj` (`:audit-log-consumer-nexus` / `:audit-log-consumer-admin-console` / `:audit-log-consumer-events` job-defs, shared drain/reap/sweep helpers and the registry of reactions); the storage-usage counter itself is recomputed by `backend/src/app/tasks/team_storage.clj` (`:schedule-team-storage` on cron `45 */10 * * * ?` + `:recompute-team-storage` on the `:storage` queue) and read by `app.rpc.quotes`; its table is `team_storage_usage` (`team_id` PK/FK `ON DELETE CASCADE`, `total_bytes`, `dirty_at`, `updated_at`, partial index on `dirty_at IS NOT NULL`), created by migration `0158`, producer hook in `backend/src/app/loggers/audit.clj` (`append-audit-entry`, `queue-pendings` with explicit consumers: Nexus default-gated on its flag, the events consumer takes the whole stream gated on `:event-consumer`, admin-console only for qualifying events) plus the bulk ingest in `backend/src/app/rpc/commands/audit.clj` (`handle-events`, same rules), metrics/cron wiring in `backend/src/app/main.clj` (one cron entry per job, Nexus at second 0, admin console at second 30, events drain at second 15 and the team-storage scheduler every 10 min at second 45, each gated by its own send flag), single migration `0157` (queue, drop of `archived_at`, covering `(created_at, id)` index for the chronological scan; no backfill — pre-existing rows are swept unsent, an accepted loss). - Nexus (private repo, no paths cited here): archive ingest, ordered consumers (local profile mirror, PostHog SaaS events, PostHog telemetry), telemetry endpoint, analytics views. - Tests: `backend_tests/rpc-audit-test.clj`, `rpc-team-test/accept-organization-invitation-audit-event`. diff --git a/.serena/memories/backend/subtleties.md b/.serena/memories/backend/subtleties.md index e59e7de48c..3b3e297054 100644 --- a/.serena/memories/backend/subtleties.md +++ b/.serena/memories/backend/subtleties.md @@ -83,7 +83,7 @@ ## Storage and media -- The `media-storage-bytes-per-team` quota is checked on the three commands that write media bytes: `upload-file-media-object` (multipart), `assemble-file-media-object` (chunked) and `create-file-media-object-from-url`. All three read the cached `team_storage_usage` counter (fail open + mark dirty when the row is missing; the ensure reads first and only inserts when the row is absent) instead of scanning `storage_object` per upload; the live `SUM` only runs when the `:storage-usage` + `:audit-log` flags are off, and in the background recompute job. A third audit-log consumer (id 3, selective pendings) marks dirty teams; `:schedule-team-storage` (cron `45 */10 * * * ?`) streams every dirty team oldest-first through a plan cursor inside one read-only transaction and enqueues one `:recompute-team-storage` per team on the dedicated `:storage` queue (never the default one, which serves the delete cascades), in chunks of 50 that commit in their own transaction via `db/own-transaction`, with a per-chunk stagger (5 s step, 60 s cap) and `skip-if-active` labelled by team id. Its query also returns the rows nobody refreshed for a week (`updated_at` older than 7 days), so a mark lost on the way heals at a bounded rate. The recompute upsert only clears dirty marks older than its read, so a mark landing mid-recompute survives for the next tick. The counter lives in `team_storage_usage` (migration `0158`): `team_id` PK/FK `ON DELETE CASCADE`, `total_bytes`, `dirty_at` (set = pending recompute), `updated_at`, partial index on `dirty_at IS NOT NULL`. `file-gc` emits `file-gc` so trims mark the team dirty; `delete-object-with-cascade` carries flat `:team-ids` so the drain never joins production tables (`hard-delete-objects` carries only its entity key and is not consumed: a physical purge cannot change the live total). The `upload-sessions-per-profile` quota is unaffected. +- The `media-storage-bytes-per-team` quota is checked on the three commands that write media bytes: `upload-file-media-object` (multipart), `assemble-file-media-object` (chunked) and `create-file-media-object-from-url`. All three read the cached `team_storage_usage` counter (fail open + mark dirty when the row is missing; the ensure reads first and only inserts when the row is absent) instead of scanning `storage_object` per upload; the live `SUM` only runs when the `:storage-usage` + `:audit-log` flags are off, and in the background recompute job. A third audit-log consumer (id 3, the events consumer, not selective: it takes the whole stream and each reaction decides what concerns it) marks dirty teams; `:schedule-team-storage` (cron `45 */10 * * * ?`) streams every dirty team oldest-first through a plan cursor inside one read-only transaction and enqueues one `:recompute-team-storage` per team on the dedicated `:storage` queue (never the default one, which serves the delete cascades), in chunks of 50 that commit in their own transaction via `db/own-transaction`, with a per-chunk stagger (5 s step, 60 s cap) and `skip-if-active` labelled by team id. Its query also returns the rows nobody refreshed for a week (`updated_at` older than 7 days), so a mark lost on the way heals at a bounded rate. The recompute upsert only clears dirty marks older than its read, so a mark landing mid-recompute survives for the next tick. The counter lives in `team_storage_usage` (migration `0158`): `team_id` PK/FK `ON DELETE CASCADE`, `total_bytes`, `dirty_at` (set = pending recompute), `updated_at`, partial index on `dirty_at IS NOT NULL`. `file-gc` emits `file-gc` so trims mark the team dirty; `delete-object-with-cascade` carries flat `:team-ids` so the drain never joins production tables (`hard-delete-objects` carries only its entity key and is not consumed: a physical purge cannot change the live total). The `upload-sessions-per-profile` quota is unaffected. - Storage abstraction, backend configuration, logical buckets, object lifecycle, deduplication, access rules, and garbage collection: `mem:backend/storage`. - SVG validation strips DOCTYPE and uses secure SAX parsing. Basic SVG info falls back to 100x100 dimensions when width/height/viewBox are missing. - Raster metadata is shell-derived with ImageMagick `identify`, verifies detected MIME against the supplied MIME, and swaps dimensions for EXIF orientations 6/8. diff --git a/backend/src/app/loggers/audit/consumers.clj b/backend/src/app/loggers/audit/consumers.clj index c8ed9015fe..024e7d9408 100644 --- a/backend/src/app/loggers/audit/consumers.clj +++ b/backend/src/app/loggers/audit/consumers.clj @@ -66,10 +66,10 @@ (def consumer-events 3) (def ^:private consumers - [{:id consumer-nexus :flag :nexus :label "nexus" :lock :ordered :outcome "archived"} - {:id consumer-admin-console :flag :admin-console :label "admin-console" :lock :ordered :outcome "archived"} + [{:id consumer-nexus :flag :nexus :label "nexus" :outcome "archived"} + {:id consumer-admin-console :flag :admin-console :label "admin-console" :outcome "archived"} ;; dispatches the events to the reactions, it archives nothing - {:id consumer-events :flag :event-consumer :label "events" :lock :unordered :outcome "processed" + {:id consumer-events :flag :event-consumer :label "events" :outcome "processed" :summary "events dispatched"}]) (def ^:private xf:enabled-consumers @@ -240,34 +240,35 @@ (subs s 0 limit) s)))) -;; NOTE: deliberately no SKIP LOCKED here. Batches must reach the -;; destinations in chronological order (Nexus numbers by arrival), -;; so every consumer takes the same total order and a second -;; overlapping run waits for the first to commit, then sees whether -;; it was last. The brief blocking is accepted; skipping rows could -;; reorder archive arrival. (def ^:private sql:pending-chunk - "SELECT a.* - FROM audit_log_pending p - JOIN audit_log a ON a.id = p.audit_log_id - WHERE p.consumer = ?::smallint - ORDER BY a.created_at ASC, a.id ASC - LIMIT 128 - FOR UPDATE OF a") + "The next chunk of events for one consumer. -(def ^:private sql:pending-chunk-unordered - "The same chunk for a consumer that does not archive and does not - care about order (the events consumer dispatches the stream, and the - reactions are idempotent). SKIP LOCKED keeps it from waiting on an - archive job holding the oldest rows: that wait could hit lock_timeout - and abort the run." + The lock is on `audit_log_pending`, the rows this consumer owns and + the ones it deletes: every consumer locks its own queue, so no + consumer waits for another one. The `audit_log` rows are only read. + + The lock is FOR NO KEY UPDATE, not FOR UPDATE: deleting an + `audit_log` row makes the foreign key check take FOR KEY SHARE on + the pending rows that reference it, and those belong to the other + consumers. FOR KEY SHARE does not conflict with FOR NO KEY UPDATE, + so the check never waits for a drain in flight and the two consumers + cannot deadlock. It still conflicts with itself, so two runs of the + same consumer take turns. + + No SKIP LOCKED, on purpose: the events have to be executed in order, + like an event log. Two runs of the same consumer (two backends, or a + retry overlapping the previous run) take the same total order and the + second one waits for the first to commit, then finds its pendings + gone and takes the next events. Skipping the locked ones would hand + the newer events to the second run while the older ones are still in + flight." "SELECT a.* FROM audit_log_pending p JOIN audit_log a ON a.id = p.audit_log_id WHERE p.consumer = ?::smallint ORDER BY a.created_at ASC, a.id ASC LIMIT 128 - FOR UPDATE OF a SKIP LOCKED") + FOR NO KEY UPDATE OF p") (def ^:private sql:delete-pendings "DELETE FROM audit_log_pending @@ -275,27 +276,22 @@ AND audit_log_id = ANY(?)") (def ^:private sql:delete-drained + "The rows nobody waits for anymore are deleted. The inner select + locks the rows with SKIP LOCKED: a consumer never waits for another + one that happens to hold the row, and the row it skips is left for + the orphan sweep." "DELETE FROM audit_log a - WHERE a.id = ANY(?) + WHERE a.id IN (SELECT id FROM audit_log + WHERE id = ANY(?) + FOR UPDATE SKIP LOCKED) AND NOT EXISTS ( SELECT 1 FROM audit_log_pending p WHERE p.audit_log_id = a.id )") -(defn- pending-rows-sql - "The chunk query of a consumer. The ones that archive block on the - oldest rows (the arrival order matters to the destination); the ones - that do not archive take the rows with SKIP LOCKED, so they never wait - on an archive job holding them (that wait could hit lock_timeout and - abort the run)." - [consumer] - (if (= :ordered (:lock consumer)) - sql:pending-chunk - sql:pending-chunk-unordered)) - (defn- get-pending-rows [{:keys [::db/conn]} consumer] - (->> (db/exec! conn [(pending-rows-sql consumer) (:id consumer)]) + (->> (db/exec! conn [sql:pending-chunk (:id consumer)]) (not-empty))) (defn- complete-batch @@ -614,11 +610,17 @@ " ORDER BY audit_log_id LIMIT 5000)")) (def ^:private sql:sweep-orphans + "The records nobody waits for anymore are deleted. The inner select + locks the rows with SKIP LOCKED, like the delete of a drained chunk: + the sweep never waits for a transaction that holds one of them (a + producer inserting a pending, another consumer draining), and the rows + it skips stay for the next sweep." "DELETE FROM audit_log a WHERE a.id IN (SELECT a2.id FROM audit_log a2 WHERE NOT EXISTS (SELECT 1 FROM audit_log_pending p WHERE p.audit_log_id = a2.id) - ORDER BY a2.id LIMIT 5000)") + ORDER BY a2.id LIMIT 5000 + FOR UPDATE SKIP LOCKED)") (defn- delete-batched "Run a batched DELETE until it removes nothing. Returns the total." diff --git a/backend/test/backend_tests/tasks_team_storage_test.clj b/backend/test/backend_tests/tasks_team_storage_test.clj index 434069fa90..3818ce9831 100644 --- a/backend/test/backend_tests/tasks_team_storage_test.clj +++ b/backend/test/backend_tests/tasks_team_storage_test.clj @@ -350,11 +350,10 @@ (get (::l/props %) :hint)) @records))))) -(t/deftest storage-usage-drain-skips-rows-locked-by-an-archive - ;; The drain does not archive and its mark is idempotent, so it must - ;; not wait on an archive job holding the oldest rows: it skips them - ;; (SKIP LOCKED) and processes the rest, instead of hitting - ;; lock_timeout and aborting the run. +(t/deftest events-drain-does-not-wait-on-a-locked-audit-log-row + ;; The drain locks its own pendings, not the audit_log rows it only + ;; reads: another transaction holding one of those rows must not stop + ;; it. The row it cannot lock is left for the orphan sweep. (let [prof (th/create-profile* 85 {:is-active true}) team-a (:default-team-id prof) team-b (:id (th/create-team* 85 {:profile-id (:id prof)}))] @@ -371,11 +370,41 @@ (th/run-task! :audit-log-consumer-events {}) (finally (db/exec! conn ["ROLLBACK"]))) - ;; The locked row stayed pending; the other one drained. - (t/is (= 1 (count (events-pendings)))) - (t/is (nil? (:dirty-at (usage-row team-a)))) + ;; Both events were dispatched: the drain never waits for the + ;; row another transaction holds. + (t/is (= 0 (count (events-pendings)))) + (t/is (some? (:dirty-at (usage-row team-a)))) (t/is (some? (:dirty-at (usage-row team-b))))))))) +(t/deftest events-drain-waits-for-the-run-in-flight + ;; The events have to be executed in order, like an event log: a + ;; second run (another backend, or a retry overlapping the previous + ;; one) waits for the pendings of the run in flight instead of taking + ;; the newer events while the older ones are still being dispatched. + (let [prof (th/create-profile* 86 {:is-active true}) + team (:default-team-id prof)] + (with-redefs [cf/flags test-flags] + (append-row "upload-file-media-object" {:team-id team}) + (with-open [conn (db/open th/*system*)] + (db/exec! conn ["BEGIN"]) + ;; Lock the pendings of the events consumer, as a run in flight + ;; holds them. + (db/exec! conn ["SELECT audit_log_id FROM audit_log_pending + WHERE consumer = 3 FOR UPDATE"]) + ;; The future needs the system: the dynamic var is not + ;; inherited by the thread. + (let [system th/*system* + job (future (binding [th/*system* system] + (th/run-task! :audit-log-consumer-events {})))] + ;; The run is blocked on the pending: it dispatched nothing. + (Thread/sleep 500) + (t/is (nil? (:dirty-at (usage-row team)))) + (db/exec! conn ["COMMIT"]) + ;; The run finishes once the pending is free: it did not + ;; time out waiting for it. + (t/is (not= :timeout (deref job 10000 :timeout))) + (t/is (some? (:dirty-at (usage-row team))))))))) + (t/deftest events-run-summary-names-its-work ;; The run summary must not say "events archived": this consumer ;; dispatches the stream and has no destination.