mirror of
https://github.com/penpot/penpot.git
synced 2026-10-31 23:10:31 +00:00
♻️ Make the team-storage rule a reaction of the event consumer
The third audit-log consumer was written for a single case: the list of event names that feed it lived in the producer, and the rule itself (which props carry the team id, how a team is marked dirty) lived in the transport that drains the queues. Every new case meant touching the producer, the transport and the metric. The transport keeps only what is generic (the queues, the drain, the housekeeping, the metrics) and the rules move to a registry of reactions, one per domain. The team-storage rule moves to the domain that owns its table (app.tasks.team-storage) and becomes a reaction: the event names it cares about, the flag that turns it on and the function that marks the teams dirty. The registry and the dispatch live in the new app.tasks.event-consumer. A reaction receives the whole batch of events it declares, so marking the teams dirty is one write per batch and not one write per event. No behavior change: the same events queue, the same teams get marked and the same metrics are recorded. The view of a row that a reaction sees carries the row id, so a reaction can correlate what it did with the row it came from; the id does not travel to the archives. AI-assisted-by: minimax-m3.1-flash
This commit is contained in:
Andrey Antukh
parent
924f2988dc
commit
593576c1ea
@ -26,6 +26,7 @@
|
||||
[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]
|
||||
@ -264,9 +265,11 @@
|
||||
|
||||
;; 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).
|
||||
;; 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? consumer/storage-usage-event-names (:name params))
|
||||
(contains? team-storage/team-storage-event-names (:name params))
|
||||
(not (telemetry-shadow-source? (:source params))))
|
||||
(consumer/queue-pendings cfg [(:id row)] consumer/consumer-storage-usage))
|
||||
|
||||
|
||||
@ -21,6 +21,7 @@
|
||||
[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]))
|
||||
|
||||
@ -102,57 +103,6 @@
|
||||
(when (contains? cf/flags :nexus)
|
||||
(queue-pendings cfg ids consumer-nexus)))
|
||||
|
||||
(def ^:private default-team-keys
|
||||
"Key map of the events that carry a single :team-id in props."
|
||||
{:paths [[:team-id]]
|
||||
:keys [:team-id]})
|
||||
|
||||
(def ^:private team-keys-by-name
|
||||
"Single source of the storage-usage contract: event name -> where its
|
||||
team ids live. `:paths` are the prop paths `team-ids-of-event` reads;
|
||||
`:keys` are the top-level props `carries-team-keys?` checks with
|
||||
`contains?`, so an explicit empty collection counts as a fact and not
|
||||
as a producer that lost its keys. Adding a name here also adds it to
|
||||
`storage-usage-event-names`, so the two cannot drift."
|
||||
{"upload-file-media-object" default-team-keys
|
||||
"create-file-media-object-from-url" default-team-keys
|
||||
"clone-file-media-object" default-team-keys
|
||||
"assemble-file-media-object" default-team-keys
|
||||
;; Files created in bulk (import, clone, duplicate) may carry media,
|
||||
;; so they can change the total even though create-file itself adds
|
||||
;; no bytes: do not drop it, and do not copy it to cheaper events.
|
||||
"create-file" default-team-keys
|
||||
"create-font-variant" default-team-keys
|
||||
"delete-font" default-team-keys
|
||||
"delete-font-variant" default-team-keys
|
||||
"delete-file" default-team-keys
|
||||
"delete-object-with-cascade" {:paths [[:team-ids] [:cascade :team-ids]]
|
||||
:keys [:team-ids :cascade]}
|
||||
"move-files" {:paths [[:source-team-id] [:dest-team-id]]
|
||||
:keys [:source-team-id :dest-team-id]}
|
||||
"move-project" {:paths [[:source-team-id] [:dest-team-id]]
|
||||
:keys [:source-team-id :dest-team-id]}
|
||||
"permanently-delete-team-files" default-team-keys
|
||||
"restore-deleted-team-files" default-team-keys
|
||||
"restore-file" default-team-keys
|
||||
"restore-profile" {:paths [[:team-ids] [:cascade :team-ids]]
|
||||
:keys [:team-ids :cascade]}
|
||||
"restore-project" default-team-keys
|
||||
"restore-team" default-team-keys
|
||||
"file-gc" default-team-keys})
|
||||
|
||||
(def storage-usage-event-names
|
||||
"Audit event names that can change per-team media storage. Only these
|
||||
queue a pending for the storage-usage consumer; everything else skips
|
||||
it, so unrelated traffic pays no extra row and the drain stays lean.
|
||||
Derived from `team-keys-by-name` so the allowlist and the key map
|
||||
cannot disagree. Purges (`hard-delete-objects`) are excluded on
|
||||
purpose: the physical delete only removes rows the live SUM already
|
||||
ignores (everything with `deleted_at` set), so queueing them would
|
||||
recompute the same total — and purged teams would violate the usage
|
||||
table foreign key."
|
||||
(set (keys team-keys-by-name)))
|
||||
|
||||
(def ^:private consumers-by-id
|
||||
(into {} (map (juxt :id identity)) consumers))
|
||||
|
||||
@ -199,6 +149,14 @@
|
||||
[row]
|
||||
(select-keys row event-keys))
|
||||
|
||||
(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
|
||||
payload they receive is the event alone."
|
||||
[row]
|
||||
(assoc (row->event (decode-row row)) :id (:id row)))
|
||||
|
||||
(defn- record-count
|
||||
[metrics id labels amount]
|
||||
(mtx/run! metrics :id id :labels labels :inc amount))
|
||||
@ -535,80 +493,6 @@
|
||||
(complete-batch cfg consumer-admin-console rows)
|
||||
admin-console-res)))
|
||||
|
||||
(def ^:private sql:mark-storage-usage-dirty
|
||||
"INSERT INTO team_storage_usage AS tsu (team_id, dirty_at)
|
||||
SELECT id, now() FROM unnest(?::uuid[]) AS u(id)
|
||||
WHERE EXISTS (SELECT 1 FROM team AS t WHERE t.id = u.id)
|
||||
ON CONFLICT (team_id) DO UPDATE
|
||||
SET dirty_at = EXCLUDED.dirty_at
|
||||
WHERE tsu.dirty_at IS NULL")
|
||||
|
||||
(defn- team-ids-of-event
|
||||
"Team ids of one storage-usage event, read through its key map (see
|
||||
team-keys-by-name). Throws on a present non-UUID: a producer bug,
|
||||
never coerced; the caller logs the row and skips that event."
|
||||
[{:keys [name props]}]
|
||||
(let [{:keys [paths]} (get team-keys-by-name name default-team-keys)
|
||||
;; A path holds either one id (:team-id) or a collection
|
||||
;; (:team-ids): normalize both to a seq.
|
||||
ids (mapcat (fn [path]
|
||||
(let [v (get-in props path)]
|
||||
(cond (nil? v) nil
|
||||
(coll? v) v
|
||||
:else [v])))
|
||||
paths)]
|
||||
(doseq [id ids :when (some? id)]
|
||||
(when-not (uuid? id)
|
||||
(ex/raise :type :validation
|
||||
:code :non-uuid-team-id
|
||||
:hint "non-uuid team id in storage-usage event"
|
||||
:event name)))
|
||||
ids))
|
||||
|
||||
(defn- carries-team-keys?
|
||||
"True when the event carries any of the team keys its key map names,
|
||||
even an empty collection: an explicit empty list (a profile that owns
|
||||
no team) is a fact, not a producer bug."
|
||||
[{:keys [name props]}]
|
||||
(let [{:keys [keys]} (get team-keys-by-name name default-team-keys)]
|
||||
;; `some?` on the value, not `contains?`: an explicit empty
|
||||
;; collection counts (a fact), a nil value does not (the producer
|
||||
;; failed to resolve its team).
|
||||
(boolean (some #(some? (get props %)) keys))))
|
||||
|
||||
(defn- collect-storage-usage-team-ids
|
||||
"Team ids whose media storage may have changed in this chunk, read
|
||||
straight from props: producers key their events by team so the drain
|
||||
never joins production tables. Each event name maps to its keys
|
||||
explicitly; anything else falls back to the single :team-id in
|
||||
props. A present non-UUID is a producer bug (see get-team-ids, never
|
||||
coerced): that event is logged with its row id and skipped, so one
|
||||
bad row cannot wedge the queue. The key map lives in
|
||||
team-keys-by-name; storage-usage-queueing-covers-every-listed-name
|
||||
pins the real prop shape of every listed name."
|
||||
[entries]
|
||||
(into #{}
|
||||
(comp (map (fn [[id event]]
|
||||
(try
|
||||
(team-ids-of-event event)
|
||||
(catch Throwable cause
|
||||
(l/warn :hint "skipping storage-usage event with invalid team ids"
|
||||
:audit-log-id (str id)
|
||||
:event (:name event)
|
||||
:cause cause)
|
||||
nil))))
|
||||
(mapcat (fn [ids] (or ids [])))
|
||||
(filter some?))
|
||||
entries))
|
||||
|
||||
(defn- mark-storage-usage-dirty
|
||||
[conn team-ids]
|
||||
(when (seq team-ids)
|
||||
;; Keeps the oldest dirty mark: a team marked twice in one window
|
||||
;; is recomputed once.
|
||||
(db/exec-one! conn [sql:mark-storage-usage-dirty
|
||||
(db/create-array conn "uuid" team-ids)])))
|
||||
|
||||
(defn- consumer-storage-usage-chunk
|
||||
"Mark dirty the teams changed in one pending chunk and complete it.
|
||||
Returns {:sent :skipped} like the other chunk fns: relevant events
|
||||
@ -618,30 +502,13 @@
|
||||
[cfg]
|
||||
(db/exec! cfg ["SET LOCAL lock_timeout = '30s'"])
|
||||
(when-let [rows (get-pending-rows cfg consumer-storage-usage)]
|
||||
(let [pairs (mapv (fn [row]
|
||||
[(:id row) ((comp row->event decode-row) row)])
|
||||
rows)
|
||||
relevant (into []
|
||||
(filter (fn [[_ event]]
|
||||
(contains? storage-usage-event-names (:name event))))
|
||||
pairs)
|
||||
skipped (- (count pairs) (count relevant))
|
||||
teams (collect-storage-usage-team-ids relevant)
|
||||
_ (mark-storage-usage-dirty (::db/conn cfg) teams)
|
||||
;; No relevant event carries any team key: a producer
|
||||
;; stopped keying its events. An explicit empty collection
|
||||
;; does not count (see carries-team-keys?).
|
||||
_ (when (and (seq relevant)
|
||||
(not-any? (comp carries-team-keys? second) relevant))
|
||||
(l/warn :hint "storage-usage chunk resolved no team ids"
|
||||
:events (count relevant)
|
||||
:ids (mapv (comp str first) relevant)))]
|
||||
(let [batch (mapv row->reaction-event rows)
|
||||
sent (events/dispatch-reactions cfg batch)]
|
||||
(l/trc :hint "consumer events chunk"
|
||||
:consumer "storage-usage"
|
||||
:events (count relevant)
|
||||
:teams (count teams))
|
||||
:events sent)
|
||||
(complete-batch cfg consumer-storage-usage rows)
|
||||
{:sent (count relevant) :skipped skipped})))
|
||||
{:sent sent :skipped (- (count batch) sent)})))
|
||||
|
||||
(defn- drain-consumer
|
||||
"Drain one consumer queue chunk by chunk, one transaction per chunk.
|
||||
|
||||
63
backend/src/app/tasks/event_consumer.clj
Normal file
63
backend/src/app/tasks/event_consumer.clj
Normal file
@ -0,0 +1,63 @@
|
||||
;; This Source Code Form is subject to the terms of the Mozilla Public
|
||||
;; License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
;;
|
||||
;; Copyright (c) KALEIDOS SUBSIDIARY SL
|
||||
|
||||
(ns app.tasks.event-consumer
|
||||
"Registro de reacciones del consumidor de eventos del audit log.
|
||||
|
||||
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:
|
||||
|
||||
{: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
|
||||
|
||||
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.
|
||||
|
||||
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)."
|
||||
(:require
|
||||
[app.config :as cf]
|
||||
[app.tasks.team-storage :as team-storage]))
|
||||
|
||||
(def reactions
|
||||
"Las reacciones registradas, en el orden en que se aplican."
|
||||
[team-storage/team-storage-reaction])
|
||||
|
||||
(defn enabled-reactions
|
||||
"Las reacciones cuya flag está encendida."
|
||||
[]
|
||||
(filterv #(contains? cf/flags (:flag %)) reactions))
|
||||
|
||||
(defn- claimed-by
|
||||
"Las reacciones activas que declaran el nombre de este evento."
|
||||
[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.
|
||||
|
||||
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."
|
||||
[cfg events]
|
||||
(let [reactions (enabled-reactions)
|
||||
batches (keep (fn [reaction]
|
||||
(let [mine (filterv #(contains? (:events reaction) (:name %)) events)]
|
||||
(when (seq mine)
|
||||
[reaction mine])))
|
||||
reactions)]
|
||||
(doseq [[reaction mine] batches]
|
||||
((:apply reaction) cfg mine))
|
||||
(count (filter (fn [event]
|
||||
(seq (claimed-by reactions event)))
|
||||
events))))
|
||||
@ -9,14 +9,15 @@
|
||||
(`team_storage_usage`) that the media-storage-bytes-per-team quota
|
||||
reads instead of scanning `storage_object` on every upload.
|
||||
|
||||
The audit-log storage-usage consumer marks dirty the teams whose
|
||||
media may have changed. Every few minutes the `:schedule-team-storage`
|
||||
scheduler streams all dirty teams oldest-first and enqueues one
|
||||
`:recompute-team-storage` job per team (skipped while one is already live),
|
||||
so a failure is contained in its own job and the next tick
|
||||
reschedules it. Teams without a row are born on first quota check
|
||||
(fail open + dirty), never swept."
|
||||
The team-storage reaction of the audit-log event consumer marks dirty
|
||||
the teams whose media may have changed. Every few minutes the
|
||||
`:schedule-team-storage` scheduler streams all dirty teams oldest-first
|
||||
and enqueues one `:recompute-team-storage` job per team (skipped while
|
||||
one is already live), so a failure is contained in its own job and the
|
||||
next tick reschedules it. Teams without a row are born on first quota
|
||||
check (fail open + dirty), never swept."
|
||||
(:require
|
||||
[app.common.exceptions :as ex]
|
||||
[app.common.logging :as l]
|
||||
[app.common.schema :as sm]
|
||||
[app.db :as db]
|
||||
@ -46,6 +47,162 @@
|
||||
dirty_at = CASE WHEN tsu.dirty_at > ?::timestamptz THEN tsu.dirty_at ELSE NULL END,
|
||||
updated_at = now()")
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; REACTION
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
(def ^:private default-team-keys
|
||||
"Key map of the events that carry a single :team-id in props."
|
||||
{:paths [[:team-id]]
|
||||
:keys [:team-id]})
|
||||
|
||||
(def ^:private team-keys-by-name
|
||||
"Single source of the storage-usage contract: event name -> where its
|
||||
team ids live. `:paths` are the prop paths `team-ids-of-event` reads;
|
||||
`:keys` are the top-level props `carries-team-keys?` checks with
|
||||
`contains?`, so an explicit empty collection counts as a fact and not
|
||||
as a producer that lost its keys. Adding a name here also adds it to
|
||||
`team-storage-event-names`, so the two cannot drift."
|
||||
{"upload-file-media-object" default-team-keys
|
||||
"create-file-media-object-from-url" default-team-keys
|
||||
"clone-file-media-object" default-team-keys
|
||||
"assemble-file-media-object" default-team-keys
|
||||
;; Files created in bulk (import, clone, duplicate) may carry media,
|
||||
;; so they can change the total even though create-file itself adds
|
||||
;; no bytes: do not drop it, and do not copy it to cheaper events.
|
||||
"create-file" default-team-keys
|
||||
"create-font-variant" default-team-keys
|
||||
"delete-font" default-team-keys
|
||||
"delete-font-variant" default-team-keys
|
||||
"delete-file" default-team-keys
|
||||
"delete-object-with-cascade" {:paths [[:team-ids] [:cascade :team-ids]]
|
||||
:keys [:team-ids :cascade]}
|
||||
"move-files" {:paths [[:source-team-id] [:dest-team-id]]
|
||||
:keys [:source-team-id :dest-team-id]}
|
||||
"move-project" {:paths [[:source-team-id] [:dest-team-id]]
|
||||
:keys [:source-team-id :dest-team-id]}
|
||||
"permanently-delete-team-files" default-team-keys
|
||||
"restore-deleted-team-files" default-team-keys
|
||||
"restore-file" default-team-keys
|
||||
"restore-profile" {:paths [[:team-ids] [:cascade :team-ids]]
|
||||
:keys [:team-ids :cascade]}
|
||||
"restore-project" default-team-keys
|
||||
"restore-team" default-team-keys
|
||||
"file-gc" default-team-keys})
|
||||
|
||||
(def team-storage-event-names
|
||||
"Audit event names that can change per-team media storage. Only these
|
||||
mark the team dirty. Derived from `team-keys-by-name` so the list and
|
||||
the key map cannot disagree. Purges (`hard-delete-objects`) are
|
||||
excluded on purpose: the physical delete only removes rows the live
|
||||
SUM already ignores (everything with `deleted_at` set), so marking
|
||||
them would recompute the same total — and purged teams would violate
|
||||
the usage table foreign key."
|
||||
(set (keys team-keys-by-name)))
|
||||
|
||||
(defn- team-ids-of-event
|
||||
"Team ids of one storage-usage event, read through its key map (see
|
||||
team-keys-by-name). Throws on a present non-UUID: a producer bug,
|
||||
never coerced; the caller logs the row and skips that event."
|
||||
[{:keys [name props]}]
|
||||
(let [{:keys [paths]} (get team-keys-by-name name default-team-keys)
|
||||
;; A path holds either one id (:team-id) or a collection
|
||||
;; (:team-ids): normalize both to a seq.
|
||||
ids (mapcat (fn [path]
|
||||
(let [v (get-in props path)]
|
||||
(cond (nil? v) nil
|
||||
(coll? v) v
|
||||
:else [v])))
|
||||
paths)]
|
||||
(doseq [id ids :when (some? id)]
|
||||
(when-not (uuid? id)
|
||||
(ex/raise :type :validation
|
||||
:code :non-uuid-team-id
|
||||
:hint "non-uuid team id in storage-usage event"
|
||||
:event name)))
|
||||
ids))
|
||||
|
||||
(defn- carries-team-keys?
|
||||
"True when the event carries any of the team keys its key map names,
|
||||
even an empty collection: an explicit empty list (a profile that owns
|
||||
no team) is a fact, not a producer bug."
|
||||
[{:keys [name props]}]
|
||||
(let [{:keys [keys]} (get team-keys-by-name name default-team-keys)]
|
||||
;; `some?` on the value, not `contains?`: an explicit empty
|
||||
;; collection counts (a fact), a nil value does not (the producer
|
||||
;; failed to resolve its team).
|
||||
(boolean (some #(some? (get props %)) keys))))
|
||||
|
||||
(defn- collect-team-ids
|
||||
"Team ids whose media storage may have changed in these events, read
|
||||
straight from props: producers key their events by team so the drain
|
||||
never joins production tables. Each event name maps to its keys
|
||||
explicitly; anything else falls back to the single :team-id in props.
|
||||
A present non-UUID is a producer bug (see team-ids-of-event, never
|
||||
coerced): that event is logged with its row id and skipped, so one bad
|
||||
row cannot wedge the queue."
|
||||
[events]
|
||||
(into #{}
|
||||
(comp (map (fn [{:keys [id] :as event}]
|
||||
(try
|
||||
(team-ids-of-event event)
|
||||
(catch Throwable cause
|
||||
(l/warn :hint "skipping storage-usage event with invalid team ids"
|
||||
:audit-log-id (str id)
|
||||
:event (:name event)
|
||||
:cause cause)
|
||||
nil))))
|
||||
(mapcat (fn [ids] (or ids [])))
|
||||
(filter some?))
|
||||
events))
|
||||
|
||||
(def ^:private sql:mark-team-storage-dirty
|
||||
"INSERT INTO team_storage_usage AS tsu (team_id, dirty_at)
|
||||
SELECT id, now() FROM unnest(?::uuid[]) AS u(id)
|
||||
WHERE EXISTS (SELECT 1 FROM team AS t WHERE t.id = u.id)
|
||||
ON CONFLICT (team_id) DO UPDATE
|
||||
SET dirty_at = EXCLUDED.dirty_at
|
||||
WHERE tsu.dirty_at IS NULL")
|
||||
|
||||
(defn- mark-team-storage-dirty
|
||||
"Mark the teams dirty, keeping the oldest mark: a team marked twice in
|
||||
one window is recomputed once."
|
||||
[conn team-ids]
|
||||
(when (seq team-ids)
|
||||
(db/exec-one! conn [sql:mark-team-storage-dirty
|
||||
(db/create-array conn "uuid" team-ids)])))
|
||||
|
||||
(def team-storage-reaction
|
||||
"Reaction that marks dirty the teams whose media storage may have
|
||||
changed. It is one case of the audit-log event consumer: the event
|
||||
arrives through the common queue and this reaction decides whether it
|
||||
concerns it.
|
||||
|
||||
It receives the whole batch of events it declares, not one event at a
|
||||
time: marking the teams dirty is a single write for the batch."
|
||||
{:id :team-storage-usage
|
||||
:flag :storage-usage
|
||||
:events team-storage-event-names
|
||||
:apply (fn [cfg events]
|
||||
(let [teams (collect-team-ids events)]
|
||||
(mark-team-storage-dirty (::db/conn cfg) teams)
|
||||
(l/trc :hint "storage-usage teams marked dirty"
|
||||
:events (count events)
|
||||
:teams (count teams))
|
||||
;; No relevant event carries any team key: a producer
|
||||
;; stopped keying its events. An explicit empty collection
|
||||
;; does not count (see carries-team-keys?).
|
||||
(when (and (seq events)
|
||||
(not-any? carries-team-keys? events))
|
||||
(l/warn :hint "storage-usage chunk resolved no team ids"
|
||||
:events (count events)
|
||||
:ids (mapv (comp str :id) events)))
|
||||
teams))})
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; RECOMPUTE
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
(defn- recompute-team-storage
|
||||
"Recompute one team: slow read, fast write. A mark written by a
|
||||
transaction that started after this recompute began survives (its
|
||||
|
||||
82
backend/test/backend_tests/tasks_event_consumer_test.clj
Normal file
82
backend/test/backend_tests/tasks_event_consumer_test.clj
Normal file
@ -0,0 +1,82 @@
|
||||
;; This Source Code Form is subject to the terms of the Mozilla Public
|
||||
;; License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
;;
|
||||
;; 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."
|
||||
(:require
|
||||
[app.common.uuid :as uuid]
|
||||
[app.config :as cf]
|
||||
[app.tasks.event-consumer :as event-consumer]
|
||||
[backend-tests.helpers :as th]
|
||||
[clojure.test :as t]))
|
||||
|
||||
(t/use-fixtures :once th/state-init)
|
||||
(t/use-fixtures :each th/database-reset)
|
||||
|
||||
(defn- capture-reaction
|
||||
"Una reacción de mentira que guarda el lote que recibe."
|
||||
[seen events]
|
||||
{:id :test
|
||||
:flag :test-flag
|
||||
:events events
|
||||
:apply (fn [_cfg batch] (swap! seen conj batch))})
|
||||
|
||||
(t/deftest dispatch-applies-a-reaction-to-the-events-it-declares
|
||||
(let [seen (atom [])
|
||||
reaction (capture-reaction seen #{"delete-file"})
|
||||
event-a {:id (uuid/next) :name "delete-file" :props {}}
|
||||
event-b {:id (uuid/next) :name "create-file" :props {}}]
|
||||
(with-redefs [cf/flags #{:test-flag}
|
||||
event-consumer/reactions [reaction]]
|
||||
(let [claimed (event-consumer/dispatch-reactions
|
||||
th/*system*
|
||||
[event-a event-b])]
|
||||
(t/is (= 1 claimed))
|
||||
(t/is (= 1 (count @seen)))
|
||||
(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.
|
||||
(let [seen (atom [])
|
||||
reaction (capture-reaction seen #{"delete-file"})]
|
||||
(with-redefs [cf/flags #{:test-flag}
|
||||
event-consumer/reactions [reaction]]
|
||||
(let [claimed (event-consumer/dispatch-reactions
|
||||
th/*system*
|
||||
[{:id (uuid/next) :name "create-file" :props {}}])]
|
||||
(t/is (= 0 claimed))
|
||||
(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.
|
||||
(let [seen (atom [])
|
||||
reaction (capture-reaction seen #{"delete-file"})]
|
||||
(with-redefs [cf/flags #{}
|
||||
event-consumer/reactions [reaction]]
|
||||
(let [claimed (event-consumer/dispatch-reactions
|
||||
th/*system*
|
||||
[{:id (uuid/next) :name "delete-file" :props {}}])]
|
||||
(t/is (= 0 claimed))
|
||||
(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 [])
|
||||
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 {}}]]
|
||||
(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))))))))
|
||||
@ -89,7 +89,7 @@
|
||||
|
||||
(t/deftest storage-usage-queueing-covers-every-listed-name
|
||||
;; Pins the drain contract for every name in
|
||||
;; storage-usage-event-names: each one queues a pending and marks the
|
||||
;; 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-
|
||||
@ -121,7 +121,7 @@
|
||||
["restore-project" {:team-id team-b}]
|
||||
["restore-team" {:team-id team}]
|
||||
["file-gc" {:team-id team :file-id (:id file)}]]]
|
||||
(t/is (= consumer/storage-usage-event-names (set (map first cases))))
|
||||
(t/is (= team-storage/team-storage-event-names (set (map first cases))))
|
||||
(with-redefs [cf/flags test-flags]
|
||||
(doseq [[nm props] cases]
|
||||
(append-row nm props))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user