github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-29 05:50:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Let the audit-log consumer react to any event

Browse Source 
The third audit-log consumer was selective: the producer knew the list
of event names that feed it, and the transport had to know them too.
Every new case meant touching the producer, the transport and the
metric.

The consumer is now the events consumer: it queues the whole stream
(every event but the telemetry shadows, which duplicate the full row)
and each reaction decides what concerns it. The producer no longer
knows any event name, and the transport only owns the queue, the drain
and the metrics.

The consumer has its own flag, :event-consumer, and its own job,
:audit-log-consumer-events, which lives in the generalist
(app.tasks.event-consumer) next to the registry of reactions. The
drain only decodes the events a reaction claims: the queue takes the
whole stream, but most of it concerns nobody.

The team-storage rule stays where it was, as a reaction of the
generalist, so it keeps marking the teams dirty exactly as before.
The metric label changes from storage-usage to events, with the same
"processed" outcome: it dispatches the stream, it archives nothing.

AI-assisted-by: minimax-m3.1-flash
This commit is contained in:
Andrey Antukh 2026-10-23 17:02:53 +00:00
parent 747e79c95a
commit 0f4c549668
9 changed files with 350 additions and 215 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
backend/src/app/loggers/audit.clj
View File

@ -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.

103
backend/src/app/loggers/audit/consumer_task.clj
View File

@ -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)))

12
backend/src/app/main.clj
View File

@ -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

18
backend/src/app/rpc/commands/audit.clj
View File

@ -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

115
backend/src/app/tasks/event_consumer.clj
View File

@ -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)})

23
backend/test/backend_tests/loggers_audit_consumer_task_test.clj
View File

@ -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))))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

102
backend/test/backend_tests/tasks_event_consumer_test.clj
View File

@ -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)))))))

168
backend/test/backend_tests/tasks_team_storage_test.clj
View File

@ -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)))))))

5
common/src/app/common/flags.cljc
View File

@ -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