diff --git a/backend/src/app/loggers/audit.clj b/backend/src/app/loggers/audit.clj index c4498ca50c..6304ccbde7 100644 --- a/backend/src/app/loggers/audit.clj +++ b/backend/src/app/loggers/audit.clj @@ -26,7 +26,6 @@ [app.loggers.webhooks :as-alias webhooks] [app.rpc :as-alias rpc] [app.setup :as-alias setup] - [app.tasks.team-storage :as team-storage] [app.util.inet :as inet] [app.util.services :as-alias sv] [cuerdas.core :as str] @@ -259,19 +258,17 @@ ;; The consumer queue follows the row in the same transaction: a ;; rollback takes both on its way out, so a row never outlives its ;; queue. Nexus takes every row; the two `when`s below add the - ;; pendings of the selective consumers (storage-usage, - ;; admin-console), each gated on its own flag. + ;; pendings of the events consumer (the whole stream) and of the + ;; selective admin-console one, each gated on its own flag. (consumer/queue-nexus-pendings cfg [(:id row)]) - ;; The storage-usage consumer is selective: only the events that - ;; can change per-team media storage queue its pending, and never - ;; for telemetry shadows (the full row already queues). The list of - ;; names belongs to the reaction that consumes them (see - ;; `app.tasks.team-storage`). - (when (and (contains? cf/flags :storage-usage) - (contains? team-storage/team-storage-event-names (:name params)) + ;; The events consumer is not selective: it takes the whole stream + ;; and each reaction decides what concerns it. Never a telemetry + ;; shadow (the full row already queues, and the shadow keeps the + ;; name, so it would be dispatched twice). + (when (and (contains? cf/flags :event-consumer) (not (telemetry-shadow-source? (:source params)))) - (consumer/queue-pendings cfg [(:id row)] consumer/consumer-storage-usage)) + (consumer/queue-events-pendings cfg [(:id row)])) ;; The admin-console consumer is selective too: only allowlisted ;; names queue its pending, and never a telemetry shadow. diff --git a/backend/src/app/loggers/audit/consumer_task.clj b/backend/src/app/loggers/audit/consumer_task.clj index e298c5a315..54d490951f 100644 --- a/backend/src/app/loggers/audit/consumer_task.clj +++ b/backend/src/app/loggers/audit/consumer_task.clj @@ -21,20 +21,23 @@ [app.metrics :as mtx] [app.nitrate :as nitrate] [app.setup :as-alias setup] - [app.tasks.event-consumer :as events] [integrant.core :as ig] [promesa.exec :as px])) -;; This task drains the audit log buffer into the long-term archives. +;; This task drains the audit log buffer: two of the jobs archive the +;; events in a long-term destination, and the third one dispatches them +;; to the reactions. ;; -;; Three separate jobs share this namespace and the queue helpers, one -;; per destination: `:audit-log-consumer-nexus`, -;; `:audit-log-consumer-admin-console` and -;; `:audit-log-consumer-storage-usage`. Each job drains only its own queue -;; and deletes its pendings on completion; whoever deletes the last -;; pending also deletes the record ("last one out turns off the -;; light"), so no separate GC task is needed. Housekeeping (reaping -;; disabled pendings, sweeping orphans) lives in every job on purpose: +;; Two separate jobs live in this namespace, one per destination: +;; `:audit-log-consumer-nexus` and +;; `:audit-log-consumer-admin-console`. The third one, +;; `:audit-log-consumer-events`, lives in `app.tasks.event-consumer` +;; (the generalist that decides what to do with each event). Each job +;; drains only its own queue and deletes its pendings on completion; +;; whoever deletes the last pending also deletes the record ("last one +;; out turns off the light"), so no separate GC task is needed. +;; Housekeeping (reaping disabled pendings, sweeping orphans) lives in +;; every job on purpose: ;; only the enabled job runs, so it must clean up for the rest. ;; Consumer ids are fixed constants, never reuse one: retired @@ -42,14 +45,14 @@ ;; previous owner's queue. (def consumer-nexus 1) (def consumer-admin-console 2) -(def consumer-storage-usage 3) +(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"} - ;; marks teams dirty, it archives nothing - {:id consumer-storage-usage :flag :storage-usage :label "storage-usage" :lock :unordered :outcome "processed" - :summary "teams marked dirty"}]) + [{: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"} + ;; dispatches the events to the reactions, it archives nothing + {:id consumer-events :flag :event-consumer :label "events" :lock :unordered :outcome "processed" + :summary "events dispatched"}]) (def ^:private xf:enabled-consumers "Static transducer: keep the consumers whose send flag is on. @@ -100,6 +103,16 @@ (when (contains? cf/flags :nexus) (queue-pendings cfg ids consumer-nexus))) +(defn queue-events-pendings + "Queue records for the events consumer, the one that dispatches the + stream to the reactions. It is not selective: it takes every event + and each reaction decides what concerns it. Gated on its flag, so the + callers never rely on what the `queue-pendings` default happens to + exclude today." + [cfg ids] + (when (contains? cf/flags :event-consumer) + (queue-pendings cfg ids consumer-events))) + (def ^:private consumers-by-id (into {} (map (juxt :id identity)) consumers)) @@ -108,8 +121,8 @@ (get-in consumers-by-id [consumer :label])) (defn- consumer-outcome - "Success outcome of a consumer in the events metric: storage-usage - marks teams dirty, it archives nothing." + "Success outcome of a consumer in the events metric: the events + consumer dispatches the stream, it archives nothing." [consumer] (get-in consumers-by-id [consumer :outcome])) @@ -146,7 +159,7 @@ [row] (select-keys row event-keys)) -(defn- row->reaction-event +(defn row->reaction-event "The view of an audit-log row that a reaction sees: the decoded event plus the id of the row, so a reaction can correlate what it did with the row it came from. The id does not travel to the archives: the @@ -183,10 +196,10 @@ (def ^:private sql:pending-chunk-unordered "The same chunk for a consumer that does not archive and does not - care about order (storage-usage only marks teams dirty, and the mark - is 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." + 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." "SELECT a.* FROM audit_log_pending p JOIN audit_log a ON a.id = p.audit_log_id @@ -487,18 +500,6 @@ :events (count decoded))] (send-to-admin-console cfg decoded))) -(defn- process-storage-usage-chunk - "Mark dirty the teams changed in one pending chunk. Returns - {:sent :skipped}: the events a reaction claims count as sent, the - rest as skipped. Same contract as the other chunk processors." - [cfg rows] - (let [batch (mapv row->reaction-event rows) - sent (events/dispatch-reactions cfg batch)] - (l/trc :hint "consumer events chunk" - :consumer "storage-usage" - :events sent) - {:sent sent :skipped (- (count batch) sent)})) - (defn- drain-chunks "Drain one consumer queue chunk by chunk, one transaction per chunk. The processor receives the rows of each chunk, already locked, and @@ -580,7 +581,6 @@ (declare execute-audit-log-consumer-nexus) (declare execute-audit-log-consumer-admin-console) -(declare execute-audit-log-consumer-storage-usage) (def schema:audit-log-consumer-nexus-params "Overrides for the nexus job repl invocation defaults. @@ -642,29 +642,6 @@ ::jobs/decoder (sm/decoder schema:audit-log-consumer-admin-console-params sm/json-transformer) ::jobs/validator (sm/validator schema:audit-log-consumer-admin-console-params)}) -(def schema:audit-log-consumer-storage-usage-params - "Params of the storage-usage job: none. Cron submits {}; system deps - (pool) live on the job-def cfg closed over by the handler, not in - job params. Closed on purpose, same as the admin-console schema." - [:map {:closed true}]) - -(defmethod ig/assert-key ::storage-usage-job-def - [_ cfg] - (assert (db/pool? (::db/pool cfg)) "expected a valid database pool") - (assert (mtx/metrics? (::mtx/metrics cfg)) "expected a valid metrics instance")) - -(defmethod ig/init-key ::storage-usage-job-def - [_ cfg] - (l/dbg :hint "audit log consumer job initialized" - :consumer "storage-usage") - {::jobs/name :audit-log-consumer-storage-usage - ::jobs/schema schema:audit-log-consumer-storage-usage-params - ::jobs/handler - (fn [_context params] - (execute-audit-log-consumer-storage-usage cfg params)) - ::jobs/decoder (sm/decoder schema:audit-log-consumer-storage-usage-params sm/json-transformer) - ::jobs/validator (sm/validator schema:audit-log-consumer-storage-usage-params)}) - (defn drain-consumer "Run one consumer: drain its queue chunk by chunk (one transaction per chunk), then the housekeeping (reap of the disabled queues and sweep @@ -760,11 +737,3 @@ (l/dbg :hint "audit log consumer (admin-console) skipped, flag off") (drain-consumer cfg consumer-admin-console process-admin-console-chunk))) -(defn execute-audit-log-consumer-storage-usage - "Storage-usage job handler: drain the storage-usage queue in chunks, - marking dirty the teams whose media storage may have changed - (heartbeat per iteration: a burst can span many chunks)." - [cfg _params] - (if-not (contains? cf/flags :storage-usage) - (l/dbg :hint "audit log consumer (storage-usage) skipped, flag off") - (drain-consumer cfg consumer-storage-usage process-storage-usage-chunk))) diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index 9680185cc5..f7fdd47831 100644 --- a/backend/src/app/main.clj +++ b/backend/src/app/main.clj @@ -662,7 +662,7 @@ :file-gc-scheduler (ig/ref :app.tasks.file-gc-scheduler/job-def) :audit-log-consumer-nexus (ig/ref :app.loggers.audit.consumer-task/nexus-job-def) :audit-log-consumer-admin-console (ig/ref :app.loggers.audit.consumer-task/admin-console-job-def) - :audit-log-consumer-storage-usage (ig/ref :app.loggers.audit.consumer-task/storage-usage-job-def)} + :audit-log-consumer-events (ig/ref :app.tasks.event-consumer/events-job-def)} ::jobs/request-pool {::rds/client (ig/ref ::rds/client) @@ -770,7 +770,7 @@ :app.nitrate/client (ig/ref :app.nitrate/client) ::mtx/metrics (ig/ref ::mtx/metrics)} - :app.loggers.audit.consumer-task/storage-usage-job-def + :app.tasks.event-consumer/events-job-def {::db/pool (ig/ref ::db/pool) ::mtx/metrics (ig/ref ::mtx/metrics)} @@ -904,10 +904,12 @@ {:cron #penpot/cron "30 */5 * * * ?" ;; every 5m, offset 30s from nexus :job :audit-log-consumer-admin-console}) - (when (contains? cf/flags :storage-usage) + (when (contains? cf/flags :event-consumer) [{:cron #penpot/cron "15 */5 * * * ?" ;; every 5m, offset 15s - :job :audit-log-consumer-storage-usage} - {:cron #penpot/cron "45 */10 * * * ?" ;; every 10m, after the drain + :job :audit-log-consumer-events}]) + + (when (contains? cf/flags :storage-usage) + [{:cron #penpot/cron "45 */10 * * * ?" ;; every 10m, after the drain :job :schedule-team-storage}])])} ::wrk/dispatcher diff --git a/backend/src/app/rpc/commands/audit.clj b/backend/src/app/rpc/commands/audit.clj index f90e83ed25..ec9a7497b7 100644 --- a/backend/src/app/rpc/commands/audit.clj +++ b/backend/src/app/rpc/commands/audit.clj @@ -128,19 +128,25 @@ (and (contains? consumer/event-names-for-admin-console name) (not (audit/telemetry-shadow-source? source)))) +(defn- telemetry-shadow-row? + "Rows are positional, in `event-columns` order: [:id :name :source ...]." + [[_ _ source]] + (audit/telemetry-shadow-source? source)) + (defn- insert-rows! [tx rows] (when (seq rows) (db/insert-many! tx :audit-log event-columns rows) ;; Same transaction as the rows: every ingested record is born ;; pending for every enabled non-selective consumer (see - ;; consumer-task). The storage-usage consumer is fed only by - ;; backend events (append-audit-entry): no frontend event name can - ;; change per-team media storage, so bulk ingest queues nothing - ;; for it. - ;; The nexus consumer takes every row (gated on its flag like - ;; the backend producer does in append-audit-entry). + ;; consumer-task). The nexus consumer takes every row (gated on its + ;; flag like the backend producer does in append-audit-entry). (consumer/queue-nexus-pendings tx (map first rows)) + ;; The events consumer is not selective either: it takes the whole + ;; stream and each reaction decides what concerns it. Never a + ;; telemetry shadow (the full row already queues; the shadow keeps + ;; the name, so it would be dispatched twice). + (consumer/queue-events-pendings tx (map first (remove telemetry-shadow-row? rows))) ;; The admin-console consumer is selective: only allowlisted names ;; queue its pending (this keeps its browser events working), and ;; never a telemetry shadow (the full row already queues; the diff --git a/backend/src/app/tasks/event_consumer.clj b/backend/src/app/tasks/event_consumer.clj index 652b6bca39..7a1c450de2 100644 --- a/backend/src/app/tasks/event_consumer.clj +++ b/backend/src/app/tasks/event_consumer.clj @@ -5,50 +5,64 @@ ;; Copyright (c) KALEIDOS SUBSIDIARY SL (ns app.tasks.event-consumer - "Registro de reacciones del consumidor de eventos del audit log. + "The generalist of the audit-log event consumer: it reads the stream + of events and lets each reaction decide what concerns it. - El consumidor de eventos (el id 3 de `audit_log_pending`) es el único - que no archiva nada: lee el flujo de eventos y deja que cada reacción - decida qué le interesa. Una reacción es un mapa: + The consumer itself (the id 3 of `audit_log_pending`) is not selective: + it queues every event and hands it to the reactions. A reaction is a + map: - {:id :team-storage-usage ;; quién es - :flag :storage-usage ;; la flag que la enciende - :events #{...} ;; los nombres de evento que le importan - :apply (fn [cfg events] ...)} ;; qué hace con ellos + {:id :team-storage-usage ;; who it is + :flag :storage-usage ;; the flag that turns it on + :events #{...} ;; the event names it cares about + :apply (fn [cfg events] ...)} ;; what it does with them - Cada dominio aporta la suya: quien tiene la tabla y el SQL de su marca - es quien sabe qué eventos le afectan y de dónde sale su clave. + Each domain contributes its own: the one that owns the table and the + SQL of its mark is the one that knows which events affect it and + where its key lives. - La reacción recibe el lote de eventos que declara, una sola vez: una - reacción que agrupa trabajo (marcar un equipo sucio) no debe escribir - una vez por evento. El despacho no captura excepciones: la reacción - decide qué errores son de datos (se salta el evento) y cuáles abortan - el lote (el evento se reintenta en el siguiente tick)." + A reaction receives the whole batch of events it declares, once: a + reaction that groups work (marking a team dirty) must not write once + per event. The dispatch does not catch exceptions: the reaction + decides which errors are data (it skips the event) and which abort + the batch (the event is retried on the next tick)." (:require + [app.common.logging :as l] + [app.common.schema :as sm] [app.config :as cf] - [app.tasks.team-storage :as team-storage])) + [app.db :as db] + [app.jobs :as jobs] + [app.loggers.audit.consumer-task :as consumer] + [app.metrics :as mtx] + [app.tasks.team-storage :as team-storage] + [integrant.core :as ig])) (def reactions - "Las reacciones registradas, en el orden en que se aplican." + "The registered reactions, in the order they are applied." [team-storage/team-storage-reaction]) (defn enabled-reactions - "Las reacciones cuya flag está encendida." + "The reactions whose flag is on." [] (filterv #(contains? cf/flags (:flag %)) reactions)) +(defn- claimed-names + "The names the enabled reactions claim." + [] + (into #{} (mapcat :events) (enabled-reactions))) + (defn- claimed-by - "Las reacciones activas que declaran el nombre de este evento." + "The enabled reactions that declare the name of this event." [reactions event] (filterv #(contains? (:events %) (:name event)) reactions)) (defn dispatch-reactions - "Aplica las reacciones activas a los eventos. Cada reacción recibe solo - los eventos que declara, y los recibe en un único lote. + "Applies the enabled reactions to the events. Each reaction receives + only the events it declares, and it receives them in a single batch. - Devuelve el número de eventos que reclamó alguna reacción: los que no - reclamó nadie se descartan, porque la cola del consumidor es no - selectiva y ese evento no le interesa a nadie." + Returns the number of events some reaction claimed: the ones nobody + claimed are discarded, because the queue of the consumer is not + selective and that event concerns nobody." [cfg events] (let [reactions (enabled-reactions) batches (keep (fn [reaction] @@ -61,3 +75,56 @@ (count (filter (fn [event] (seq (claimed-by reactions event))) events)))) + +(defn- process-events-chunk + "Dispatch one pending chunk to the reactions. Returns {:sent :skipped}: + the events a reaction claims count as sent, the rest as skipped. + + Only the claimed events are decoded: the queue is not selective, so + most of the stream concerns nobody and opening its blob would be + work thrown away." + [cfg rows] + (let [names (claimed-names) + claimed (filterv #(contains? names (:name %)) rows) + batch (mapv consumer/row->reaction-event claimed) + sent (dispatch-reactions cfg batch)] + (l/trc :hint "consumer events chunk" + :consumer "events" + :events sent) + {:sent sent :skipped (- (count rows) sent)})) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; HANDLER +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; + +(defn execute-audit-log-consumer-events + "Events job handler: drain the events queue in chunks, dispatching + each event to the reactions that claim it (heartbeat per iteration: + a burst can span many chunks)." + [cfg] + (if-not (contains? cf/flags :event-consumer) + (l/dbg :hint "audit log consumer (events) skipped, flag off") + (consumer/drain-consumer cfg consumer/consumer-events process-events-chunk))) + +(def schema:audit-log-consumer-events-params + "Params of the events job: none. Cron submits {}; system deps (pool) + live on the job-def cfg closed over by the handler, not in job params. + Closed on purpose, same as the admin-console schema." + [:map {:closed true}]) + +(defmethod ig/assert-key ::events-job-def + [_ cfg] + (assert (db/pool? (::db/pool cfg)) "expected a valid database pool") + (assert (mtx/metrics? (::mtx/metrics cfg)) "expected a valid metrics instance")) + +(defmethod ig/init-key ::events-job-def + [_ cfg] + (l/dbg :hint "audit log consumer job initialized" + :consumer "events") + {::jobs/name :audit-log-consumer-events + ::jobs/schema schema:audit-log-consumer-events-params + ::jobs/handler + (fn [_context _params] + (execute-audit-log-consumer-events cfg)) + ::jobs/decoder (sm/decoder schema:audit-log-consumer-events-params sm/json-transformer) + ::jobs/validator (sm/validator schema:audit-log-consumer-events-params)}) diff --git a/backend/test/backend_tests/loggers_audit_consumer_task_test.clj b/backend/test/backend_tests/loggers_audit_consumer_task_test.clj index 32db1711b0..ef0e5cbf26 100644 --- a/backend/test/backend_tests/loggers_audit_consumer_task_test.clj +++ b/backend/test/backend_tests/loggers_audit_consumer_task_test.clj @@ -22,6 +22,7 @@ [app.metrics :as mtx] [app.metrics.definition :as-alias mdef] [app.setup :as-alias setup] + [app.tasks.event-consumer :as-alias event-consumer] [backend-tests.helpers :as th] [clojure.test :as t] [cuerdas.core :as str] @@ -477,12 +478,12 @@ (invoke-nexus! metrics params) (invoke-admin-console! metrics)) -(defn- invoke-storage-usage! - "Invoke the storage-usage consumer handler in-process with an - isolated metrics registry. It only needs the pool: marking teams - dirty touches no archive destination." +(defn- invoke-events! + "Invoke the events consumer handler in-process with an isolated + metrics registry. It only needs the pool: dispatching the events to + the reactions touches no archive destination." [metrics] - (let [job-def (ig/init-key :app.loggers.audit.consumer-task/storage-usage-job-def + (let [job-def (ig/init-key :app.tasks.event-consumer/events-job-def {::db/pool (:app.db/pool th/*system*) ::mtx/metrics metrics})] ((::jobs/handler job-def) nil {}))) @@ -515,15 +516,15 @@ (let [rows (th/db-exec! ["select * from audit_log"])] (t/is (= 0 (count rows)))))))) -(t/deftest storage-usage-consumer-records-processed-outcome - ;; The storage-usage consumer marks teams dirty, it archives - ;; nothing: its success outcome reads "processed", not "archived". +(t/deftest events-consumer-records-processed-outcome + ;; The events consumer dispatches the stream, it archives nothing: + ;; its success outcome reads "processed", not "archived". (let [prof (th/create-profile* 1 {:is-active true}) metrics (make-consumer-metrics)] - (with-redefs [cf/flags #{:audit-log :storage-usage}] + (with-redefs [cf/flags #{:audit-log :event-consumer :storage-usage}] (insert-audit-row! (:id prof) "upload-file-media-object") - (invoke-storage-usage! metrics) - (t/is (= 1.0 (counter-value metrics :audit-consumer-events ["storage-usage" "processed"]))) + (invoke-events! metrics) + (t/is (= 1.0 (counter-value metrics :audit-consumer-events ["events" "processed"]))) (t/is (= 0 (pending-count)))))) ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; diff --git a/backend/test/backend_tests/tasks_event_consumer_test.clj b/backend/test/backend_tests/tasks_event_consumer_test.clj index 83e030bfc6..c98c8087fb 100644 --- a/backend/test/backend_tests/tasks_event_consumer_test.clj +++ b/backend/test/backend_tests/tasks_event_consumer_test.clj @@ -5,11 +5,14 @@ ;; Copyright (c) KALEIDOS SUBSIDIARY SL (ns backend-tests.tasks-event-consumer-test - "Registro de reacciones del consumidor de eventos del audit log: qué - reacción reclama cada evento y qué hace con el lote que le toca." + "The generalist of the audit-log event consumer: the registry of + reactions, the dispatch and the drain of the queue." (:require + [app.common.time :as ct] [app.common.uuid :as uuid] [app.config :as cf] + [app.db :as db] + [app.loggers.audit :as audit] [app.tasks.event-consumer :as event-consumer] [backend-tests.helpers :as th] [clojure.test :as t])) @@ -18,13 +21,40 @@ (t/use-fixtures :each th/database-reset) (defn- capture-reaction - "Una reacción de mentira que guarda el lote que recibe." + "A fake reaction that keeps the batch it receives." [seen events] {:id :test :flag :test-flag :events events :apply (fn [_cfg batch] (swap! seen conj batch))}) +(defn- append-row + "Append an audit-log row the way the producer does." + ([name props] (append-row name props "backend")) + ([name props source] + (#'audit/append-audit-entry + th/*system* + {:id (uuid/next) + :name name + :type "action" + :source source + :profile-id uuid/zero + :ip-addr "127.0.0.1" + :props props + :context {} + :tracked-at (ct/now) + :created-at (ct/now)}))) + +(defn- events-pendings + "The pendings of the events consumer (id 3)." + [] + (th/db-exec! ["SELECT audit_log_id FROM audit_log_pending WHERE consumer = 3"])) + +(defn- usage-row + [team-id] + (th/db-exec-one! ["SELECT team_id, dirty_at FROM team_storage_usage WHERE team_id = ?" + team-id])) + (t/deftest dispatch-applies-a-reaction-to-the-events-it-declares (let [seen (atom []) reaction (capture-reaction seen #{"delete-file"}) @@ -40,8 +70,8 @@ (t/is (= ["delete-file"] (mapv :name (first @seen)))))))) (t/deftest dispatch-ignores-the-events-no-reaction-claims - ;; La cola del consumidor es no selectiva: los eventos que nadie - ;; reclama se descartan. + ;; The queue of the consumer is not selective: the events that nobody + ;; claims are discarded. (let [seen (atom []) reaction (capture-reaction seen #{"delete-file"})] (with-redefs [cf/flags #{:test-flag} @@ -53,8 +83,8 @@ (t/is (empty? @seen)))))) (t/deftest dispatch-skips-a-reaction-whose-flag-is-off - ;; La flag de la reacción es la que la enciende: sin ella, el evento - ;; no se reclama y la reacción no corre. + ;; The flag of the reaction is the one that turns it on: without it, + ;; the event is not claimed and the reaction does not run. (let [seen (atom []) reaction (capture-reaction seen #{"delete-file"})] (with-redefs [cf/flags #{} @@ -66,17 +96,63 @@ (t/is (empty? @seen)))))) (t/deftest dispatch-gives-each-reaction-only-its-own-events - ;; Dos reacciones que declaran el mismo nombre: cada una recibe su - ;; lote y el evento se cuenta una sola vez. - (let [seen-a (atom []) - seen-b (atom []) + ;; Two reactions that declare the same name: each one receives its + ;; batch and the event is counted once. + (let [seen-a (atom []) + seen-b (atom []) reaction-a (capture-reaction seen-a #{"delete-file"}) reaction-b (capture-reaction seen-b #{"delete-file" "create-file"}) - events [{:id (uuid/next) :name "delete-file" :props {}} - {:id (uuid/next) :name "create-file" :props {}}]] + events [{:id (uuid/next) :name "delete-file" :props {}} + {:id (uuid/next) :name "create-file" :props {}}]] (with-redefs [cf/flags #{:test-flag} event-consumer/reactions [reaction-a reaction-b]] (let [claimed (event-consumer/dispatch-reactions th/*system* events)] (t/is (= 2 claimed)) (t/is (= ["delete-file"] (mapv :name (first @seen-a)))) (t/is (= ["delete-file" "create-file"] (mapv :name (first @seen-b)))))))) + +(t/deftest events-consumer-drains-the-queue-and-marks-the-team-dirty + ;; The whole path: the producer queues the event, the consumer drains + ;; it and the reaction marks the team dirty. + (let [prof (th/create-profile* 41 {:is-active true}) + team (:default-team-id prof)] + (with-redefs [cf/flags #{:audit-log :event-consumer :storage-usage}] + (append-row "upload-file-media-object" {:team-id team}) + (t/is (= 1 (count (events-pendings)))) + (th/run-task! :audit-log-consumer-events {}) + (t/is (= 0 (count (events-pendings)))) + (t/is (some? (:dirty-at (usage-row team))))))) + +(t/deftest events-consumer-discards-the-events-nobody-claims + ;; An event that no reaction claims is drained and dropped: the queue + ;; takes the whole stream, but nobody works on it. + (let [prof (th/create-profile* 42 {:is-active true}) + team (:default-team-id prof)] + (with-redefs [cf/flags #{:audit-log :event-consumer :storage-usage}] + (append-row "navigate" {:route "dashboard"}) + (th/run-task! :audit-log-consumer-events {}) + (t/is (= 0 (count (events-pendings)))) + (t/is (nil? (:dirty-at (usage-row team))))))) + +(t/deftest events-consumer-does-nothing-without-its-flag + ;; Without its flag the consumer does not even queue: the producer + ;; never hands it the stream. + (let [prof (th/create-profile* 43 {:is-active true}) + team (:default-team-id prof)] + (with-redefs [cf/flags #{:audit-log :storage-usage}] + (append-row "upload-file-media-object" {:team-id team}) + (th/run-task! :audit-log-consumer-events {}) + (t/is (= 0 (count (events-pendings)))) + (t/is (nil? (:dirty-at (usage-row team))))))) + +(t/deftest events-consumer-does-nothing-without-the-reaction-flag + ;; The queue takes the event, but the reaction that would work on it + ;; is off: the event is drained and dropped. + (let [prof (th/create-profile* 44 {:is-active true}) + team (:default-team-id prof)] + (with-redefs [cf/flags #{:audit-log :event-consumer}] + (append-row "upload-file-media-object" {:team-id team}) + (t/is (= 1 (count (events-pendings)))) + (th/run-task! :audit-log-consumer-events {}) + (t/is (= 0 (count (events-pendings)))) + (t/is (nil? (:dirty-at (usage-row team))))))) diff --git a/backend/test/backend_tests/tasks_team_storage_test.clj b/backend/test/backend_tests/tasks_team_storage_test.clj index febe3509fb..a88fd00906 100644 --- a/backend/test/backend_tests/tasks_team_storage_test.clj +++ b/backend/test/backend_tests/tasks_team_storage_test.clj @@ -32,16 +32,19 @@ (t/use-fixtures :each th/database-reset) (def ^:private test-flags - "Quote checks run (:quotes), events are stored (:audit-log) and the - storage-usage consumer queues (:storage-usage)." - #{:quotes :audit-log :storage-usage}) + "Quote checks run (:quotes), events are stored (:audit-log), the + events consumer queues them (:event-consumer) and the team-storage + reaction runs (:storage-usage)." + #{:quotes :audit-log :event-consumer :storage-usage}) (def ^:private real-audit-submit "The unstubbed audit submit: helpers.clj replaces it globally per test, so an end-to-end producer test rebinds it back." @#'audit/submit) -(defn- storage-usage-pendings +(defn- events-pendings + "The pendings of the events consumer (id 3): it is not selective, it + takes the whole stream." [] (th/db-exec! ["SELECT audit_log_id, consumer FROM audit_log_pending WHERE consumer = 3"])) @@ -66,34 +69,42 @@ :tracked-at (ct/now) :created-at (ct/now)}))) -(t/deftest storage-usage-queueing-matches-listed-names - (with-redefs [cf/flags #{:audit-log :storage-usage}] +(t/deftest events-queueing-takes-every-event + ;; The events consumer is not selective: the whole stream queues and + ;; each reaction decides what concerns it. + (with-redefs [cf/flags #{:audit-log :event-consumer}] (append-row "upload-file-media-object" {}) (append-row "navigate" {}) (append-row "delete-file" {}) - (t/is (= 2 (count (storage-usage-pendings)))))) + (t/is (= 3 (count (events-pendings)))))) -(t/deftest storage-usage-queueing-ignores-hard-delete-objects +(t/deftest storage-usage-reaction-ignores-hard-delete-objects ;; Physical purges cannot change the live total (they only remove rows - ;; the SUM already excludes), so they queue nothing: this also keeps - ;; purged teams from ever hitting the usage-table foreign key. - (with-redefs [cf/flags #{:audit-log :storage-usage}] - (append-row "hard-delete-objects" {}) - (t/is (= 0 (count (storage-usage-pendings)))))) + ;; the SUM already excludes), so the reaction does not claim them: + ;; this also keeps purged teams from ever hitting the usage-table + ;; foreign key. + (let [prof (th/create-profile* 62 {:is-active true}) + team (:default-team-id prof)] + (with-redefs [cf/flags test-flags] + (append-row "hard-delete-objects" {:team-ids [team]}) + (th/run-task! :audit-log-consumer-events {}) + (t/is (nil? (:dirty-at (usage-row team))))))) -(t/deftest storage-usage-queueing-skips-telemetry-shadows - (with-redefs [cf/flags #{:audit-log :telemetry :storage-usage}] +(t/deftest events-queueing-skips-telemetry-shadows + ;; A telemetry shadow keeps the name of the full row, so queueing it + ;; would dispatch the same event twice. + (with-redefs [cf/flags #{:audit-log :telemetry :event-consumer}] (append-row "upload-file-media-object" {} "backend") (append-row "upload-file-media-object" {} "telemetry:backend") - (t/is (= 1 (count (storage-usage-pendings)))))) + (t/is (= 1 (count (events-pendings)))))) -(t/deftest storage-usage-queueing-covers-every-listed-name - ;; Pins the drain contract for every name in - ;; team-storage-event-names: each one queues a pending and marks the - ;; teams its key map resolves, using the real prop shape. It fails - ;; when a name is added without its case; the producers themselves - ;; are pinned by the end-to-end tests (delete-file, delete-font- - ;; variant, moves, file-gc), which go through real command output. +(t/deftest storage-usage-reaction-covers-every-listed-name + ;; Pins the reaction contract for every name in + ;; team-storage-event-names: each one marks the teams its key map + ;; resolves, using the real prop shape. It fails when a name is + ;; added without its case; the producers themselves are pinned by the + ;; end-to-end tests (delete-file, delete-font-variant, moves, + ;; file-gc), which go through real command output. (let [prof (th/create-profile* 71 {:is-active true}) team (:default-team-id prof) team-b (:id (th/create-team* 71 {:profile-id (:id prof)})) @@ -125,11 +136,10 @@ (with-redefs [cf/flags test-flags] (doseq [[nm props] cases] (append-row nm props)) - ;; Negatives: never queued for the storage-usage consumer. + ;; Negatives: the reaction does not claim them. (append-row "hard-delete-objects" {:team-ids [team]}) (append-row "navigate" {:route "dashboard"}) - (t/is (= (count cases) (count (storage-usage-pendings)))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team)))) (t/is (some? (:dirty-at (usage-row team-b))))))) @@ -144,7 +154,7 @@ real @#'quotes/get-live-media-storage-bytes] (with-redefs [cf/flags test-flags] (append-row "upload-file-media-object" {:team-id team}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (with-redefs [quotes/get-live-media-storage-bytes (fn [conn team-id] (append-row "delete-file" {:team-id team}) @@ -153,9 +163,9 @@ (th/run-pending-jobs)) ;; The stale mark is gone but the mid-run pending survived it. (t/is (nil? (:dirty-at (usage-row team)))) - (t/is (= 1 (count (storage-usage-pendings)))) + (t/is (= 1 (count (events-pendings)))) ;; The drain re-marks from the pending; the next tick clears it. - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team)))) (th/run-task! :schedule-team-storage {}) (th/run-pending-jobs) @@ -169,7 +179,7 @@ team (:default-team-id prof)] (with-redefs [cf/flags test-flags] (append-row "upload-file-media-object" {:team-id team}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (th/db-update! :team-storage-usage {:dirty-at (ct/in-future {:hours 1})} {:team-id team}) @@ -178,10 +188,10 @@ (t/is (== live (:total-bytes (usage-row team)))) (t/is (some? (:dirty-at (usage-row team)))))))) -(t/deftest storage-usage-queueing-needs-flag +(t/deftest events-queueing-needs-flag (with-redefs [cf/flags #{:audit-log}] (append-row "upload-file-media-object" {}) - (t/is (= 0 (count (storage-usage-pendings)))))) + (t/is (= 0 (count (events-pendings)))))) (t/deftest queue-pendings-batches-past-the-threshold ;; Above `single-insert-max-ids` the helper must chunk instead of one @@ -309,10 +319,10 @@ (with-redefs [cf/flags test-flags] (append-row "delete-file" {:team-id nil}) (try - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (finally (remove-watch l/log-record ::capture))) - (t/is (= 0 (count (storage-usage-pendings)))) + (t/is (= 0 (count (events-pendings)))) (t/is (let [deadline (+ (System/currentTimeMillis) 5000)] (loop [] (if (some #(= "storage-usage chunk resolved no team ids" @@ -332,10 +342,10 @@ (with-redefs [cf/flags test-flags] (append-row "restore-profile" {:id (uuid/next) :team-ids []}) (try - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (finally (remove-watch l/log-record ::capture))) - (t/is (= 0 (count (storage-usage-pendings)))) + (t/is (= 0 (count (events-pendings)))) (t/is (not-any? #(= "storage-usage chunk resolved no team ids" (get (::l/props %) :hint)) @records))))) @@ -358,19 +368,19 @@ (db/exec! conn ["BEGIN"]) (db/exec! conn ["SELECT id FROM audit_log WHERE id = ? FOR UPDATE" (:id locked)]) (try - (th/run-task! :audit-log-consumer-storage-usage {}) + (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 (storage-usage-pendings)))) + (t/is (= 1 (count (events-pendings)))) (t/is (nil? (:dirty-at (usage-row team-a)))) (t/is (some? (:dirty-at (usage-row team-b))))))))) -(t/deftest storage-usage-run-summary-names-its-work +(t/deftest events-run-summary-names-its-work ;; The run summary must not say "events archived": this consumer - ;; marks teams dirty and has no destination. - (t/is (= "teams marked dirty" - (#'consumer/consumer-summary consumer/consumer-storage-usage))) + ;; dispatches the stream and has no destination. + (t/is (= "events dispatched" + (#'consumer/consumer-summary consumer/consumer-events))) (t/is (= "events archived" (#'consumer/consumer-summary consumer/consumer-nexus)))) @@ -386,10 +396,10 @@ (with-redefs [cf/flags test-flags] (append-row "delete-file" {}) (try - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (finally (remove-watch l/log-record ::capture))) - (t/is (= 0 (count (storage-usage-pendings)))) + (t/is (= 0 (count (events-pendings)))) (t/is (let [deadline (+ (System/currentTimeMillis) 5000)] (loop [] (if (some #(= "storage-usage chunk resolved no team ids" @@ -425,7 +435,7 @@ (t/is (some? (:team-id (:props event)))) (audit/submit th/*system* event)) ;; Only the command event: the cascade job has not run yet. - (t/is (= 1 (count (storage-usage-pendings))))))) + (t/is (= 1 (count (events-pendings))))))) (t/deftest storage-usage-real-delete-font-variant-queues-pending ;; A producer whose props do NOT carry :team-id (only the request @@ -457,7 +467,7 @@ (t/is (= "delete-font-variant" (:name event))) (t/is (= team (:team-id (:props event)))) (audit/submit th/*system* event)) - (t/is (= 1 (count (storage-usage-pendings))))))) + (t/is (= 1 (count (events-pendings))))))) (t/deftest storage-usage-drain-skips-invalid-team-id ;; One bad row cannot wedge the queue: the event is logged and @@ -467,16 +477,18 @@ (with-redefs [cf/flags test-flags] (append-row "upload-file-media-object" {:team-id "not-a-uuid"}) (append-row "upload-file-media-object" {:team-id team}) - (t/is (= 2 (count (storage-usage-pendings)))) - (th/run-task! :audit-log-consumer-storage-usage {}) - (t/is (= 0 (count (storage-usage-pendings)))) + (t/is (= 2 (count (events-pendings)))) + (th/run-task! :audit-log-consumer-events {}) + (t/is (= 0 (count (events-pendings)))) (t/is (some? (:dirty-at (usage-row team))))))) -(t/deftest storage-usage-append-queues-selectively +(t/deftest events-append-queues-the-whole-stream + ;; The queue of the events consumer is not selective: the selectivity + ;; lives in the reaction, which only claims the names it knows. (with-redefs [cf/flags test-flags] (append-row "upload-file-media-object" {:team-id (uuid/next)}) (append-row "navigate" {:route "dashboard"}) - (t/is (= 1 (count (storage-usage-pendings)))))) + (t/is (= 2 (count (events-pendings)))))) (t/deftest storage-usage-drain-and-recompute (let [prof (th/create-profile* 1) @@ -496,7 +508,7 @@ :content mfile})] (t/is (nil? (:error out)))) (append-row "upload-file-media-object" {:team-id team-id :file-id (:id file)}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (let [row (usage-row team-id)] (t/is (some? row)) (t/is (some? (:dirty-at row)))) @@ -576,7 +588,7 @@ (with-redefs [cf/flags test-flags] (append-row "upload-file-media-object" {:team-id team-b}) (append-row "upload-file-media-object" {:team-id team-a}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (= 2 (th/run-task! :schedule-team-storage {}))) (t/is (= 0 (th/run-task! :schedule-team-storage {}))) (th/run-pending-jobs) @@ -592,7 +604,7 @@ team-storage/scheduler-chunk-size 5] (doseq [team teams] (append-row "upload-file-media-object" {:team-id team})) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (= 12 (th/run-task! :schedule-team-storage {}))) (let [rows (th/db-exec! ["SELECT (round((extract(epoch from (scheduled_at - created_at)) * 1000)::numeric, -3))::bigint AS delay_ms FROM job WHERE name = 'recompute-team-storage'"])] (t/is (= 12 (count rows))) @@ -607,7 +619,7 @@ team-storage/scheduler-chunk-size 5] (doseq [team teams] (append-row "upload-file-media-object" {:team-id team})) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (let [calls (atom 0) orig jobs/submit] (t/is (thrown? clojure.lang.ExceptionInfo @@ -653,7 +665,7 @@ team (:default-team-id prof)] (with-redefs [cf/flags test-flags] (append-row "upload-file-media-object" {:team-id team}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (= 1 (th/run-task! :schedule-team-storage {}))) (let [row (th/db-exec-one! ["SELECT queue FROM job WHERE name = 'recompute-team-storage'"])] (t/is (= "storage" (:queue row))))))) @@ -666,7 +678,7 @@ team (:id (th/create-team* 65 {:profile-id (:id prof)}))] (with-redefs [cf/flags test-flags] (append-row "upload-file-media-object" {:team-id team}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (= 1 (th/run-task! :schedule-team-storage {}))) (with-redefs [sys/system th/*system*] (srepl/delete-team! (str team))) @@ -682,7 +694,7 @@ team-id (:default-team-id prof)] (with-redefs [cf/flags test-flags] (append-row "upload-file-media-object" {:team-id team-id}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (= 1 (th/run-task! :schedule-team-storage {}))) (let [live-fn quotes/get-live-media-storage-bytes] (with-redefs [quotes/get-live-media-storage-bytes @@ -719,10 +731,10 @@ (th/run-pending-jobs) (th/run-task! :objects-gc {:skip-delay true}) (t/is (nil? (th/db-get :team {:id team-doomed} {::db/remove-deleted false}))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) ;; The drain completed instead of aborting the chunk in a loop: ;; no pendings left and the surviving team is marked. - (t/is (= 0 (count (storage-usage-pendings)))) + (t/is (= 0 (count (events-pendings)))) (t/is (some? (:dirty-at (usage-row team-alive))))))) (t/deftest storage-usage-restore-team-moves-totals @@ -745,7 +757,7 @@ :content mfile})] (t/is (nil? (:error out)))) (append-row "upload-file-media-object" {:team-id team :file-id (:id file)}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (th/run-task! :schedule-team-storage {}) (th/run-pending-jobs) (let [live (quotes/get-live-media-storage-bytes th/*system* team)] @@ -753,14 +765,14 @@ (t/is (== live (:total-bytes (usage-row team))))) (with-redefs [sys/system th/*system*] (srepl/delete-team! (str team))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (th/run-task! :schedule-team-storage {}) (th/run-pending-jobs) (t/is (== 0 (:total-bytes (usage-row team)))) (with-redefs [sys/system th/*system*] (srepl/restore-team! (str team))) - (t/is (= 1 (count (storage-usage-pendings)))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (t/is (= 1 (count (events-pendings)))) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team)))) (th/run-task! :schedule-team-storage {}) (th/run-pending-jobs) @@ -779,13 +791,13 @@ :id (:id file)})] (t/is (nil? (:error out)))) (th/run-pending-jobs) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (th/run-task! :schedule-team-storage {}) (th/run-pending-jobs) (with-redefs [sys/system th/*system*] (srepl/restore-file! (str (:id file)))) - (t/is (= 1 (count (storage-usage-pendings)))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (t/is (= 1 (count (events-pendings)))) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team))))))) (t/deftest storage-usage-restore-project-marks-dirty @@ -798,13 +810,13 @@ :id (:id proj)})] (t/is (nil? (:error out)))) (th/run-pending-jobs) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (th/run-task! :schedule-team-storage {}) (th/run-pending-jobs) (with-redefs [sys/system th/*system*] (srepl/restore-project! (str (:id proj)))) - (t/is (= 1 (count (storage-usage-pendings)))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (t/is (= 1 (count (events-pendings)))) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team))))))) (t/deftest storage-usage-restore-profile-marks-dirty @@ -813,13 +825,13 @@ (with-redefs [cf/flags test-flags] (with-redefs [sys/system th/*system*] (srepl/delete-profile! (str (:id prof)))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (th/run-task! :schedule-team-storage {}) (th/run-pending-jobs) (with-redefs [sys/system th/*system*] (srepl/restore-profile! (str (:id prof)))) - (t/is (= 1 (count (storage-usage-pendings)))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (t/is (= 1 (count (events-pendings)))) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team))))))) (t/deftest storage-usage-move-project-moves-totals @@ -863,7 +875,7 @@ (append-row "move-project" (assoc props :team-id team-b :project-id (:id proj-a))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team-a)))) (t/is (some? (:dirty-at (usage-row team-b)))) (th/run-task! :schedule-team-storage {}) @@ -911,7 +923,7 @@ (append-row "move-files" (assoc props :team-id team-b :project-id (:id proj-b))) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team-a)))) (t/is (some? (:dirty-at (usage-row team-b)))) (th/run-task! :schedule-team-storage {}) @@ -934,7 +946,7 @@ :props {:route "dashboard"} :timestamp (ct/now)}]} {:app.http/request (th/make-dummy-request)})) - (t/is (= 0 (count (storage-usage-pendings))))))) + (t/is (= 0 (count (events-pendings))))))) (t/deftest storage-usage-missing-row-fails-open (let [prof (th/create-profile* 1) @@ -1000,11 +1012,11 @@ (t/is (= team-id (-> event :props :team-id))))) (with-redefs [cf/flags test-flags] (append-row "file-gc" {:file-id (:id file) :team-id team-id}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team-id)))) (t/is (= 1 (th/run-task! :schedule-team-storage {}))) (th/run-pending-jobs) (t/is (nil? (:dirty-at (usage-row team-id)))) (append-row "delete-object-with-cascade" {:cascade {:team-ids [team-id]}}) - (th/run-task! :audit-log-consumer-storage-usage {}) + (th/run-task! :audit-log-consumer-events {}) (t/is (some? (:dirty-at (usage-row team-id))))))) diff --git a/common/src/app/common/flags.cljc b/common/src/app/common/flags.cljc index 955b541e57..db407805a4 100644 --- a/common/src/app/common/flags.cljc +++ b/common/src/app/common/flags.cljc @@ -121,6 +121,11 @@ :onboarding :quotes :soft-quotes + ;; Generalist consumer of the audit-log events: it drains the + ;; buffer 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. + :event-consumer ;; Background per-team media storage usage counter fed by the ;; audit-log consumer. Needs :audit-log to receive events. :storage-usage