🐛 Read the live storage total without the events consumer

The cached path of the media-storage-bytes-per-team quota reads the
team_storage_usage counter, and that counter only stays true while
somebody marks the teams dirty. That somebody is the events consumer:
with its flag off nobody refreshes the counter and the quota reads a
frozen value, so it lets uploads over the limit pass without a word.
The failure is permissive, which is the one this work came to fix.

The cached path needs the :event-consumer flag too, next to
:storage-usage and :audit-log. Without it the quota falls back to the
live sum: slow on big teams, but true.

AI-assisted-by: minimax-m3.1-flash

.serena/memories/audit-log.md

@@ -57,7 +57,7 @@ Main columns (migration `backend/src/app/migrations/sql/0054-add-audit-log-table
 - `source` text: `backend` / `frontend` for full rows, `telemetry:backend` / `telemetry:frontend` for anonymized copies (see section 10).
 - `ip_addr` inet: real IP (or `0.0.0.0` on telemetry rows).
 - `props` / `context` jsonb holding transit-encoded maps: action data and metadata (versions, browser, request-id...).
-- No `archived_at`: the `audit_log_pending` queue (migration `0157`, consumer `1` = Nexus, `2` = admin console, `3` = storage-usage — see `mem:backend/subtleties` —, ids never reused) tracks what is left. Every record is born pending for the enabled non-selective consumers (today only Nexus; the producer gates on the flag), plus one pending per selective consumer only when the event qualifies (admin-console allowlist, storage-usage event list, both ingest paths): cheaper on single-consumer hosts and true to the design doc. Consequence, accepted on purpose: rows born with no send flag on get no pendings, so enabling archiving later does not pick them up — the sweep deletes them unarchived. Same for a consumer turned selective later: only events queued after the change drain.
+- No `archived_at`: the `audit_log_pending` queue (migration `0157`, consumer `1` = Nexus, `2` = admin console, `3` = events — see `mem:backend/subtleties` —, ids never reused) tracks what is left. Every record is born pending for the enabled non-selective consumers (today only Nexus; the producer gates on the flag), plus one pending per selective consumer only when the event qualifies (admin-console allowlist, storage-usage event list, both ingest paths): cheaper on single-consumer hosts and true to the design doc. Consequence, accepted on purpose: rows born with no send flag on get no pendings, so enabling archiving later does not pick them up — the sweep deletes them unarchived. Same for a consumer turned selective later: only events queued after the change drain.
 
 Indexes (each consumer has its own, so a slow consumer never blocks the others):
 
@@ -277,7 +277,7 @@ 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 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 <key>"` (`: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-events` on flag `:event-consumer`, 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 <key>"` (`: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`).
 
@@ -314,7 +314,8 @@ Flags (`common/src/app/common/flags.cljc`, varia; enabled with `PENPOT_FLAGS=ena
 | `:audit-log` | Stores full rows; turns on `wrap-audit` and frontend ingest |
 | `:nexus` | Send flag of the Nexus archive job `:audit-log-consumer-nexus` (cron `0 */5 * * * ?`; needs `:nexus-uri` (ingest path `api/audit` appended by the job)) |
 | `:admin-console` | Send flag of the admin-console archive job `:audit-log-consumer-admin-console` (cron `30 */5 * * * ?`, 30s after Nexus); also the module flag for the admin-console integration |
-| `:storage-usage` | Send flag of the storage-usage consumer `:audit-log-consumer-storage-usage` (cron `15 */5 * * * ?`) and of its scheduler `:schedule-team-storage` (cron `45 */10 * * * ?`, queue `:storage`); also gates the cached path of the `media-storage-bytes-per-team` quota |
+| `:event-consumer` | Send flag of the generalist consumer `:audit-log-consumer-events` (cron `15 */5 * * * ?`): it drains the whole stream and hands each event to the reactions that claim it, so a new reaction needs no change in the producer. Needs `:audit-log` to receive events |
+| `:storage-usage` | Send flag of the team-storage reaction (the one reaction registered today) and of its scheduler `:schedule-team-storage` (cron `45 */10 * * * ?`, queue `:storage`); also gates the cached path of the `media-storage-bytes-per-team` quota, which needs `:event-consumer` too (without it nobody marks the teams dirty and the counter stays frozen) |
 | `:audit-log-logger` | Structured `app.audit` log per event |
 | `:error-reporting` | Starts the error reporters (off by default); database reporter on flag alone, Mattermost on flag plus webhook URL |
 | `:telemetry` | Stores shadow rows and turns on shipping (auto with `:telemetry-enabled`) |

.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, 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.
+- 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` + `:event-consumer` flags are not all on (without the events consumer nobody marks the teams dirty, so the counter would be frozen and the quota would read a stale value), 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.

backend/src/app/rpc/quotes.clj

@@ -647,15 +647,19 @@
   [{:keys [::db/conn ::profile-id ::team-id ::target] :as cfg}]
   (assert (valid-media-storage-bytes-per-team-quote? cfg) "invalid quote parameters")
   (let [count-sql (if (and (contains? cf/flags :storage-usage)
-                           (contains? cf/flags :audit-log))
+                           (contains? cf/flags :audit-log)
+                           (contains? cf/flags :event-consumer))
                     ;; Cached path: the background recompute job keeps the
                     ;; counter. Fail open: a team with no row yet is allowed
                     ;; through and marked dirty for the next recompute pass.
                     (do
                       (ensure-team-storage-usage conn team-id)
                       [sql:get-cached-media-storage-bytes-per-team team-id])
-                    ;; Live path: full scan, only when the background
-                    ;; counter is off.
+                    ;; Live path: full scan, when the background counter is
+                    ;; off or nobody feeds it. The events consumer is what
+                    ;; marks the teams dirty, so without it the counter
+                    ;; stays frozen and the quota has to read the truth
+                    ;; the slow way.
                     [sql:get-media-storage-bytes-per-team
                      team-id team-id team-id team-id team-id team-id])]
     (-> cfg

backend/test/backend_tests/tasks_team_storage_test.clj

@@ -607,6 +607,46 @@
                                             :content mfile})]
             (t/is (nil? (:error out)))))))))
 
+(t/deftest media-quota-falls-back-to-the-live-sum-without-the-events-consumer
+  ;; The cached counter is only true while somebody marks the teams
+  ;; dirty, and that is the events consumer. Without it the counter is
+  ;; frozen, so the quota has to read the live sum: slow, but true.
+  ;; A frozen counter (no row at all) with a live sum above the limit
+  ;; must reject the upload.
+  (doseq [[i flags] (map-indexed vector [#{:quotes :audit-log :storage-usage}
+                                         #{:quotes :audit-log}])]
+    (let [prof (th/create-profile* (+ 82 i) {:is-active true})
+          team (:default-team-id prof)
+          proj (th/create-project* (+ 82 i) {:profile-id (:id prof) :team-id team})
+          file (th/create-file* (+ 82 i) {:profile-id (:id prof) :project-id (:id proj) :is-shared false})
+          mfile {:filename "sample.jpg"
+                 :path (th/tempfile "backend_tests/test_files/sample.jpg")
+                 :mtype "image/jpeg"
+                 :size 312043}]
+      (with-redefs [cf/flags flags]
+        (let [out (th/multipart-command! {::th/type :upload-file-media-object
+                                          ::rpc/profile-id (:id prof)
+                                          :file-id (:id file)
+                                          :is-local true
+                                          :name "testfile"
+                                          :content mfile})]
+          (t/is (nil? (:error out))))
+        ;; The live path never creates the counter row: nobody is
+        ;; keeping it up to date, so there is nothing to read.
+        (t/is (nil? (usage-row team)))
+        ;; Limit above one upload but below two: the live SUM (already
+        ;; ~312 KB) rejects the second one.
+        (with-mocks [mock {:target 'app.config/get
+                           :return (th/config-get-mock {:quotes-media-storage-bytes-per-team 400000})}]
+          (let [out (th/multipart-command! {::th/type :upload-file-media-object
+                                            ::rpc/profile-id (:id prof)
+                                            :file-id (:id file)
+                                            :is-local true
+                                            :name "testfile2"
+                                            :content mfile})]
+            (t/is (some? (:error out)))
+            (t/is (= :max-quote-reached (:code (ex-data (:error out)))))))))))
+
 (t/deftest storage-usage-scheduler-skips-live-teams
   ;; A second scheduler run while the workers are still queued
   ;; enqueues nothing: skip-if-active on the team id as label.