github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-28 21:40:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

♻️ Make cascade helpers stream rows and enqueue tasks

Browse Source 
The four `process-deleted-*-cascade` helpers of `app.srepl.main`
loaded every soft-deleted row of a table into memory (`db/exec!`)
and ran `:delete-object` inline (`wrk/invoke!`), so they could not
be used against production: `file` alone has 2.1M deleted rows.

They now share a single `submit-deleted-cascade!` that reads the
rows with a server-side cursor (`db/plan`, in a read-only
transaction) and submits one task per row, so the work goes
through the workers like any other task. The row id goes as
`label` with `dedupe` on, and the helpers accept `:limit`, `:ids`
and `:dry-run?` to run bounded batches. Public names and their
zero-arity form are unchanged.

Tests cover the task params, the row's own date (not `now`), the
`limit`/`ids`/`dry-run?` options and that each helper reads its own
table.

AI-assisted-by: deepseek-v4.1-flash
This commit is contained in:
Andrey Antukh 2026-10-12 12:21:10 +00:00
parent f242ff9574
commit 3bbc1c3079
2 changed files with 181 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
backend/src/app/srepl/main.clj
View File

@ -828,45 +828,69 @@
;; CASCADE FIXING
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn- submit-deleted-cascade
"Replay the deletion cascade of every soft-deleted row of `table`, by
submitting one `:delete-object` task per row.
Reads the rows with a server-side cursor (`db/plan`) instead of materialising
the whole table, and enqueues the task instead of running it inline, so it
can be used against production in bounded batches. The task is idempotent (it
only marks the children with the row's own date), `label` is the row id and
`dedupe` drops not-yet-due duplicates.
Options: `:limit` (max rows read), `:ids` (restrict the run to these row ids)
and `:dry-run?` (count without enqueueing). `table` and `object` are internal
constants, never operator input.
Returns `{:found n :submitted m}`.
Note: the submissions are given the outer `system` (with the pool) on
purpose; handing them the read-only transaction cfg would write on it."
[system {:keys [table object limit ids dry-run?]}]
(let [clauses (cond-> ["deleted_at is not null"]
(seq ids) (conj "id = any(?)"))
sql (str "select id, deleted_at from " table
" where " (str/join " and " clauses)
(when limit (str " limit " (long limit))))]
(db/tx-run! (assoc system ::db/read-only true)
(fn [{:keys [::db/conn]}]
(let [query (cond-> [sql]
(seq ids) (conj (db/create-array conn "uuid" ids)))
found (reduce (fn [total {:keys [id deleted-at]}]
(when-not dry-run?
(wrk/submit! (-> system
(assoc ::wrk/task :delete-object)
(assoc ::wrk/params {:object object
:deleted-at deleted-at
:id id})
(assoc ::wrk/label (str id))
(assoc ::wrk/dedupe true))))
(inc total))
0
(db/plan conn query))]
{:found found
:submitted (if dry-run? 0 found)})))))
(defn process-deleted-profiles-cascade
[]
(->> (db/exec! sys/system ["select id, deleted_at from profile where deleted_at is not null"])
(run! (fn [{:keys [id deleted-at]}]
(wrk/invoke! (-> sys/system
(assoc ::wrk/task :delete-object)
(assoc ::wrk/params {:object :profile
:deleted-at deleted-at
:id id})))))))
"Replay the deletion cascade of every soft-deleted profile."
[& {:as opts}]
(submit-deleted-cascade sys/system (merge {:table "profile" :object :profile} opts)))
(defn process-deleted-teams-cascade
[]
(->> (db/exec! sys/system ["select id, deleted_at from team where deleted_at is not null"])
(run! (fn [{:keys [id deleted-at]}]
(wrk/invoke! (-> sys/system
(assoc ::wrk/task :delete-object)
(assoc ::wrk/params {:object :team
:deleted-at deleted-at
:id id})))))))
"Replay the deletion cascade of every soft-deleted team."
[& {:as opts}]
(submit-deleted-cascade sys/system (merge {:table "team" :object :team} opts)))
(defn process-deleted-projects-cascade
[]
(->> (db/exec! sys/system ["select id, deleted_at from project where deleted_at is not null"])
(run! (fn [{:keys [id deleted-at]}]
(wrk/invoke! (-> sys/system
(assoc ::wrk/task :delete-object)
(assoc ::wrk/params {:object :project
:deleted-at deleted-at
:id id})))))))
"Replay the deletion cascade of every soft-deleted project."
[& {:as opts}]
(submit-deleted-cascade sys/system (merge {:table "project" :object :project} opts)))
(defn process-deleted-files-cascade
[]
(->> (db/exec! sys/system ["select id, deleted_at from file where deleted_at is not null"])
(run! (fn [{:keys [id deleted-at]}]
(wrk/invoke! (-> sys/system
(assoc ::wrk/task :delete-object)
(assoc ::wrk/params {:object :file
:deleted-at deleted-at
:id id})))))))
"Replay the deletion cascade of every soft-deleted file."
[& {:as opts}]
(submit-deleted-cascade sys/system (merge {:table "file" :object :file} opts)))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; SSO

126
backend/test/backend_tests/srepl_main_test.clj
View File

@ -2,13 +2,45 @@
;; 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
;; Copyright (c) KALEIDOS INC Sucursal en España SL
(ns backend-tests.srepl-main-test
(:require
[app.common.time :as ct]
[app.srepl.main :as srepl]
[app.system :as sys]
[app.worker :as wrk]
[backend-tests.helpers :as th]
[clojure.test :as t]))
(t/use-fixtures :once th/state-init)
(t/use-fixtures :each th/database-reset)
;; --- HELPERS
(defn- capture-submissions!
"Runs `f` with `wrk/submit!` replaced by a collector, and returns the option
maps it was called with, in order."
[f]
(let [submitted (atom [])]
(with-redefs [wrk/submit! (fn [opts]
(swap! submitted conj opts)
(get-in opts [::wrk/params :id]))]
(f))
@submitted))
(defn- params-of
[submitted]
(mapv ::wrk/params submitted))
(defn- create-deleted-profile!
[i deleted-at]
(let [profile (th/create-profile* i)]
(th/db-update! :profile {:deleted-at deleted-at} {:id (:id profile)})
profile))
;; --- TESTS
(t/deftest parse-emails
(t/is (= ["some@example.com"]
(srepl/parse-emails "some@example.com")))
@ -26,3 +58,95 @@
(t/is (thrown? clojure.lang.ExceptionInfo
(srepl/parse-emails 42))))
(t/deftest cascade-enqueues-one-delete-object-task-per-deleted-row
(let [deleted (create-deleted-profile! 991 (ct/in-past {:days 30}))
_ (th/create-profile* 992)
submitted (capture-submissions!
#(#'srepl/submit-deleted-cascade
th/*system* {:table "profile" :object :profile}))]
(t/testing "only the soft-deleted row is processed"
(t/is (= 1 (count submitted))))
(t/testing "the enqueued task is a delete-object for that row"
(let [opts (first submitted)]
(t/is (= :delete-object (::wrk/task opts)))
(t/is (= {:object :profile :id (:id deleted)}
(select-keys (::wrk/params opts) [:object :id])))))
(t/testing "the date is the row's own, not now"
(t/is (ct/is-before? (:deleted-at (::wrk/params (first submitted)))
(ct/in-past {:days 29}))))
(t/testing "label is the row id and dedupe is on"
(t/is (= (str (:id deleted)) (::wrk/label (first submitted))))
(t/is (true? (::wrk/dedupe (first submitted)))))))
(t/deftest cascade-honours-limit-and-ids
(let [p1 (create-deleted-profile! 993 (ct/in-past {:days 10}))
_ (create-deleted-profile! 994 (ct/in-past {:days 9}))
_ (create-deleted-profile! 995 (ct/in-past {:days 8}))]
(t/testing "limit bounds how many rows are read"
(let [submitted (capture-submissions!
#(#'srepl/submit-deleted-cascade
th/*system* {:table "profile" :object :profile :limit 2}))]
(t/is (= 2 (count submitted)))))
(t/testing "ids restricts the run to the given rows"
(let [submitted (capture-submissions!
#(#'srepl/submit-deleted-cascade
th/*system* {:table "profile" :object :profile :ids [(:id p1)]}))]
(t/is (= [{:object :profile :id (:id p1)}]
(mapv #(select-keys % [:object :id]) (params-of submitted))))))))
(t/deftest cascade-dry-run-counts-without-enqueueing
(let [_ (create-deleted-profile! 996 (ct/in-past {:days 10}))
submitted (atom [])
result (with-redefs [wrk/submit! (fn [opts] (swap! submitted conj opts))]
(#'srepl/submit-deleted-cascade
th/*system* {:table "profile" :object :profile :dry-run? true}))]
(t/is (= 1 (:found result)))
(t/is (= 0 (:submitted result)))
(t/is (empty? @submitted))))
(t/deftest public-helpers-read-their-own-table
(let [profile (th/create-profile* 997)
team (th/create-team* 997 {:profile-id (:id profile)})
project (th/create-project* 997 {:profile-id (:id profile)
:team-id (:id team)})
file (th/create-file* 997 {:profile-id (:id profile)
:project-id (:id project)})
old (ct/in-past {:days 10})]
(th/db-update! :profile {:deleted-at old} {:id (:id profile)})
(th/db-update! :team {:deleted-at old} {:id (:id team)})
(th/db-update! :project {:deleted-at old} {:id (:id project)})
(th/db-update! :file {:deleted-at old} {:id (:id file)})
(with-redefs [sys/system th/*system*]
(let [ids (fn [submitted] (into #{} (map :id (params-of submitted))))]
(t/testing "profiles helper"
(let [submitted (capture-submissions! #(srepl/process-deleted-profiles-cascade))]
(t/is (contains? (ids submitted) (:id profile)))))
(t/testing "teams helper"
(let [submitted (capture-submissions! #(srepl/process-deleted-teams-cascade))]
(t/is (contains? (ids submitted) (:id team)))))
(t/testing "projects helper"
(let [submitted (capture-submissions! #(srepl/process-deleted-projects-cascade))]
(t/is (contains? (ids submitted) (:id project)))))
(t/testing "files helper"
(let [submitted (capture-submissions! #(srepl/process-deleted-files-cascade))]
(t/is (contains? (ids submitted) (:id file)))))
(t/testing "the helpers accept options"
(let [submitted (capture-submissions!
#(srepl/process-deleted-files-cascade :ids [(:id file)]))]
(t/is (= [{:object :file :id (:id file)}]
(mapv #(select-keys % [:object :id]) (params-of submitted))))))))))