mirror of
https://github.com/penpot/penpot.git
synced 2026-10-28 21:40:34 +00:00
🐛 Keep objects-gc running past a blocked chunk
objects-gc deleted every kind of object inside one transaction per proc, looping until empty, so a single row that a foreign key still references aborted the whole run: the task died on its first file every night, after the earlier steps had already spent around twenty minutes. The per-file savepoint only saved the files proc and left the same shape everywhere else. Now every proc streams its pending keys with db/plan inside a read-only transaction of its own and deletes them chunk by chunk, each chunk in a transaction of its own with FOR UPDATE SKIP LOCKED. A chunk that fails is reported with l/err and skipped, so the run reaches the end and the next run retries the skipped rows. AI-assisted-by: muse-spark-1.3-contributor
This commit is contained in:
Andrey Antukh
parent
115b9b9006
commit
0947e666c0
@ -27,7 +27,7 @@
|
||||
- `del-object!` sets `deleted_at`. It does not remove backend content.
|
||||
- `storage-gc-deleted` removes the database row and backend content after the deletion delay, streaming the pending rows with `db/plan` inside a read-only transaction and deleting them in chunks of 25, each chunk in its own transaction: a chunk that fails (for example, an object a foreign key still references) is logged and skipped instead of aborting the run.
|
||||
- `storage-gc-touched` finds references before it sets `deleted_at`.
|
||||
- `objects-gc` removes deleted domain rows and touches their storage object IDs.
|
||||
- `objects-gc` removes deleted domain rows and touches their storage object IDs. Each proc streams its pending keys with `db/plan` inside a read-only transaction and deletes them in chunks, each chunk in its own transaction with `FOR UPDATE SKIP LOCKED`: a chunk that fails is logged and skipped instead of aborting the run.
|
||||
- Use `::db/reuse-conn true` with `sto/resolve` inside a database transaction.
|
||||
|
||||
## Deduplication
|
||||
|
||||
@ -16,305 +16,563 @@
|
||||
[app.tasks.delete-object :as dobj]
|
||||
[integrant.core :as ig]))
|
||||
|
||||
;; Each proc below streams its pending keys with `db/plan`, a server-side
|
||||
;; cursor, inside a read-only transaction of its own, and deletes them chunk
|
||||
;; by chunk, each chunk in a transaction of its own, unrelated to the plan
|
||||
;; one. A chunk that fails is rolled back on its own, reported with `l/err`
|
||||
;; and skipped, so it takes down its own work and nothing else: the run keeps
|
||||
;; going and the next run retries the skipped rows.
|
||||
;;
|
||||
;; A plan row is a view over the open result set, valid only while the
|
||||
;; reduction is running, so the key is read inside the reducing function:
|
||||
;; buffering the row itself yields nil columns.
|
||||
|
||||
(def ^:private sql:get-profiles
|
||||
"SELECT id, photo_id FROM profile
|
||||
"SELECT id FROM profile
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-profiles
|
||||
"SELECT id, photo_id, deleted_at FROM profile
|
||||
WHERE id = ANY(?::uuid[])
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn- delete-profiles!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size ::sto/storage] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-profiles timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [id photo-id]}]
|
||||
(l/trc :obj "profile" :id (str id))
|
||||
(defn- delete-profiles-chunk
|
||||
[cfg ids]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp ::sto/storage] :as cfg}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(->> (db/exec! conn [sql:lock-profiles ids timestamp])
|
||||
(reduce (fn [total {:keys [id photo-id deleted-at]}]
|
||||
(l/trc :obj "profile" :id (str id))
|
||||
|
||||
;; Mark as deleted the storage object
|
||||
(some->> photo-id (sto/touch-object! storage))
|
||||
;; Mark as deleted the storage object
|
||||
(some->> photo-id (sto/touch-object! storage))
|
||||
|
||||
;; Cascade soft-delete to owned teams, projects, files, etc.
|
||||
(dobj/delete-object cfg {:object :profile
|
||||
:id id
|
||||
:deleted-at timestamp})
|
||||
;; Cascade soft-delete to owned teams, projects, files, etc.
|
||||
;; with the deletion date of the profile, not the
|
||||
;; timestamp of the run.
|
||||
(dobj/delete-object cfg {:object :profile
|
||||
:id id
|
||||
:deleted-at deleted-at})
|
||||
|
||||
(let [affected (-> (db/delete! conn :profile {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :profile {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "profile"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk ids
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-profiles
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-profiles-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map :id) (partition-all chunk-size)) rf 0
|
||||
(db/plan conn [sql:get-profiles timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-teams
|
||||
"SELECT deleted_at, id, photo_id FROM team
|
||||
"SELECT id FROM team
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-teams
|
||||
"SELECT deleted_at, id, photo_id FROM team
|
||||
WHERE id = ANY(?::uuid[])
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn- delete-teams!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size ::sto/storage] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-teams timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [id photo-id deleted-at]}]
|
||||
(l/trc :obj "team"
|
||||
:id (str id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(defn- delete-teams-chunk
|
||||
[cfg ids]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp ::sto/storage]}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(->> (db/exec! conn [sql:lock-teams ids timestamp])
|
||||
(reduce (fn [total {:keys [id photo-id deleted-at]}]
|
||||
(l/trc :obj "team"
|
||||
:id (str id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
;; Mark as deleted the storage object
|
||||
(some->> photo-id (sto/touch-object! storage))
|
||||
;; Mark as deleted the storage object
|
||||
(some->> photo-id (sto/touch-object! storage))
|
||||
|
||||
;; And finally, permanently delete the team.
|
||||
(let [affected (-> (db/delete! conn :team {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
;; And finally, permanently delete the team.
|
||||
(let [affected (-> (db/delete! conn :team {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "team"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk ids
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-teams
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-teams-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map :id) (partition-all chunk-size)) rf 0
|
||||
(db/plan conn [sql:get-teams timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-fonts
|
||||
"SELECT id FROM team_font_variant
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-fonts
|
||||
"SELECT id, team_id, deleted_at, woff1_file_id, woff2_file_id, otf_file_id, ttf_file_id
|
||||
FROM team_font_variant
|
||||
WHERE deleted_at IS NOT NULL
|
||||
WHERE id = ANY(?::uuid[])
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn- delete-fonts!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size ::sto/storage] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-fonts timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [id team-id deleted-at] :as font}]
|
||||
(l/trc :obj "font-variant"
|
||||
:id (str id)
|
||||
:team-id (str team-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(defn- delete-fonts-chunk
|
||||
[cfg ids]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp ::sto/storage]}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(->> (db/exec! conn [sql:lock-fonts ids timestamp])
|
||||
(reduce (fn [total {:keys [id team-id deleted-at] :as font}]
|
||||
(l/trc :obj "font-variant"
|
||||
:id (str id)
|
||||
:team-id (str team-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
;; Mark as deleted the all related storage objects
|
||||
(some->> (:woff1-file-id font) (sto/touch-object! storage))
|
||||
(some->> (:woff2-file-id font) (sto/touch-object! storage))
|
||||
(some->> (:otf-file-id font) (sto/touch-object! storage))
|
||||
(some->> (:ttf-file-id font) (sto/touch-object! storage))
|
||||
;; Mark as deleted the all related storage objects
|
||||
(some->> (:woff1-file-id font) (sto/touch-object! storage))
|
||||
(some->> (:woff2-file-id font) (sto/touch-object! storage))
|
||||
(some->> (:otf-file-id font) (sto/touch-object! storage))
|
||||
(some->> (:ttf-file-id font) (sto/touch-object! storage))
|
||||
|
||||
(let [affected (-> (db/delete! conn :team-font-variant {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :team-font-variant {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "font-variant"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk ids
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-fonts
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-fonts-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map :id) (partition-all chunk-size)) rf 0
|
||||
(db/plan conn [sql:get-fonts timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-projects
|
||||
"SELECT id, deleted_at, team_id
|
||||
FROM project
|
||||
"SELECT id FROM project
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-projects
|
||||
"SELECT id, deleted_at, team_id FROM project
|
||||
WHERE id = ANY(?::uuid[])
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn- delete-projects!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-projects timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [id team-id deleted-at]}]
|
||||
(l/trc :obj "project"
|
||||
:id (str id)
|
||||
:team-id (str team-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(defn- delete-projects-chunk
|
||||
[cfg ids]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp]}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(->> (db/exec! conn [sql:lock-projects ids timestamp])
|
||||
(reduce (fn [total {:keys [id team-id deleted-at]}]
|
||||
(l/trc :obj "project"
|
||||
:id (str id)
|
||||
:team-id (str team-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
(let [affected (-> (db/delete! conn :project {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :project {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "project"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk ids
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-projects
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-projects-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map :id) (partition-all chunk-size)) rf 0
|
||||
(db/plan conn [sql:get-projects timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-files
|
||||
"SELECT f.id,
|
||||
f.deleted_at,
|
||||
f.project_id
|
||||
FROM file AS f
|
||||
WHERE f.deleted_at IS NOT NULL
|
||||
AND f.deleted_at <= ?
|
||||
ORDER BY f.deleted_at ASC
|
||||
LIMIT ?
|
||||
"SELECT id FROM file
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-files
|
||||
"SELECT id, deleted_at, project_id FROM file
|
||||
WHERE id = ANY(?::uuid[])
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn- delete-files!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-files timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [id deleted-at project-id] :as file}]
|
||||
(l/trc :obj "file"
|
||||
:id (str id)
|
||||
:project-id (str project-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(defn- delete-files-chunk
|
||||
[cfg ids]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp]}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(->> (db/exec! conn [sql:lock-files ids timestamp])
|
||||
(reduce (fn [total {:keys [id deleted-at project-id]}]
|
||||
(l/trc :obj "file"
|
||||
:id (str id)
|
||||
:project-id (str project-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
(let [affected (-> (db/delete! conn :file {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :file {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "file"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk ids
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-files
|
||||
"Deletes the pending files chunk by chunk.
|
||||
|
||||
A file that cannot be deleted (for example, because a child row was left
|
||||
unmarked and its `NO ACTION` foreign key still points at it) aborts the
|
||||
transaction of the whole chunk: the chunk is reported and skipped, and the
|
||||
next run retries its files."
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-files-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map :id) (partition-all chunk-size)) rf 0
|
||||
(db/plan conn [sql:get-files timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-file-thumbnails
|
||||
"SELECT file_id, revn FROM file_thumbnail
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-file-thumbnails
|
||||
"SELECT file_id, revn, media_id, deleted_at
|
||||
FROM file_thumbnail
|
||||
WHERE deleted_at IS NOT NULL
|
||||
WHERE (file_id, revn) IN (SELECT file_id, revn FROM unnest(?::uuid[], ?::bigint[]) AS u(file_id, revn))
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn delete-file-thumbnails!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size ::sto/storage] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-file-thumbnails timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [file-id revn media-id deleted-at]}]
|
||||
(l/trc :obj "file-thumbnail"
|
||||
:file-id (str file-id)
|
||||
:revn revn
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(defn- delete-file-thumbnails-chunk
|
||||
[cfg rows]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp ::sto/storage]}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [file-ids (db/create-array conn "uuid" (map :file-id rows))
|
||||
revns (db/create-array conn "int8" (map :revn rows))]
|
||||
(->> (db/exec! conn [sql:lock-file-thumbnails file-ids revns timestamp])
|
||||
(reduce (fn [total {:keys [file-id revn media-id deleted-at]}]
|
||||
(l/trc :obj "file-thumbnail"
|
||||
:file-id (str file-id)
|
||||
:revn revn
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
;; Mark as deleted the storage object
|
||||
(some->> media-id (sto/touch-object! storage))
|
||||
;; Mark as deleted the storage object
|
||||
(some->> media-id (sto/touch-object! storage))
|
||||
|
||||
(let [affected (-> (db/delete! conn :file-thumbnail {:file-id file-id :revn revn})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :file-thumbnail {:file-id file-id :revn revn})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "file-thumbnail"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk rows
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-file-thumbnails
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-file-thumbnails-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map (fn [row] {:file-id (:file-id row) :revn (:revn row)}))
|
||||
(partition-all chunk-size))
|
||||
rf 0
|
||||
(db/plan conn [sql:get-file-thumbnails timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-file-object-thumbnails
|
||||
"SELECT file_id, object_id FROM file_tagged_object_thumbnail
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-file-object-thumbnails
|
||||
"SELECT file_id, object_id, media_id, deleted_at
|
||||
FROM file_tagged_object_thumbnail
|
||||
WHERE deleted_at IS NOT NULL
|
||||
WHERE (file_id, object_id) IN (SELECT file_id, object_id FROM unnest(?::uuid[], ?::text[]) AS u(file_id, object_id))
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn delete-file-object-thumbnails!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size ::sto/storage] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-file-object-thumbnails timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [file-id object-id media-id deleted-at]}]
|
||||
(l/trc :obj "file-object-thumbnail"
|
||||
:file-id (str file-id)
|
||||
:object-id object-id
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(defn- delete-file-object-thumbnails-chunk
|
||||
[cfg rows]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp ::sto/storage]}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [file-ids (db/create-array conn "uuid" (map :file-id rows))
|
||||
object-ids (db/create-array conn "text" (map :object-id rows))]
|
||||
(->> (db/exec! conn [sql:lock-file-object-thumbnails file-ids object-ids timestamp])
|
||||
(reduce (fn [total {:keys [file-id object-id media-id deleted-at]}]
|
||||
(l/trc :obj "file-object-thumbnail"
|
||||
:file-id (str file-id)
|
||||
:object-id object-id
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
;; Mark as deleted the storage object
|
||||
(some->> media-id (sto/touch-object! storage))
|
||||
;; Mark as deleted the storage object
|
||||
(some->> media-id (sto/touch-object! storage))
|
||||
|
||||
(let [affected (-> (db/delete! conn :file-tagged-object-thumbnail
|
||||
{:file-id file-id :object-id object-id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :file-tagged-object-thumbnail
|
||||
{:file-id file-id :object-id object-id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "file-object-thumbnail"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk rows
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-file-object-thumbnails
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-file-object-thumbnails-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map (fn [row] {:file-id (:file-id row) :object-id (:object-id row)}))
|
||||
(partition-all chunk-size))
|
||||
rf 0
|
||||
(db/plan conn [sql:get-file-object-thumbnails timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-file-media-objects
|
||||
"SELECT id FROM file_media_object
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-file-media-objects
|
||||
"SELECT id, file_id, media_id, thumbnail_id, deleted_at
|
||||
FROM file_media_object
|
||||
WHERE deleted_at IS NOT NULL
|
||||
WHERE id = ANY(?::uuid[])
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn- delete-file-media-objects!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size ::sto/storage] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-file-media-objects timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [id file-id deleted-at] :as fmo}]
|
||||
(l/trc :obj "file-media-object"
|
||||
:id (str id)
|
||||
:file-id (str file-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(defn- delete-file-media-objects-chunk
|
||||
[cfg ids]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp ::sto/storage]}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(->> (db/exec! conn [sql:lock-file-media-objects ids timestamp])
|
||||
(reduce (fn [total {:keys [id file-id deleted-at] :as fmo}]
|
||||
(l/trc :obj "file-media-object"
|
||||
:id (str id)
|
||||
:file-id (str file-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
;; Mark as deleted the all related storage objects
|
||||
(some->> (:media-id fmo) (sto/touch-object! storage))
|
||||
(some->> (:thumbnail-id fmo) (sto/touch-object! storage))
|
||||
;; Mark as deleted the all related storage objects
|
||||
(some->> (:media-id fmo) (sto/touch-object! storage))
|
||||
(some->> (:thumbnail-id fmo) (sto/touch-object! storage))
|
||||
|
||||
(let [affected (-> (db/delete! conn :file-media-object {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :file-media-object {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "file-media-object"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk ids
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-file-media-objects
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-file-media-objects-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map :id) (partition-all chunk-size)) rf 0
|
||||
(db/plan conn [sql:get-file-media-objects timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-file-data
|
||||
"SELECT file_id, id FROM file_data
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-file-data
|
||||
"SELECT file_id, id, type, deleted_at, metadata, backend
|
||||
FROM file_data
|
||||
WHERE deleted_at IS NOT NULL
|
||||
WHERE (file_id, id) IN (SELECT file_id, id FROM unnest(?::uuid[], ?::uuid[]) AS u(file_id, id))
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn- delete-file-data!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-file-data timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [file-id id type deleted-at metadata backend]}]
|
||||
(defn- delete-file-data-chunk
|
||||
[cfg rows]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp] :as cfg}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [file-ids (db/create-array conn "uuid" (map :file-id rows))
|
||||
ids (db/create-array conn "uuid" (map :id rows))]
|
||||
(->> (db/exec! conn [sql:lock-file-data file-ids ids timestamp])
|
||||
(reduce (fn [total {:keys [file-id id type deleted-at metadata backend]}]
|
||||
|
||||
(some->> metadata
|
||||
(fdata/decode-metadata)
|
||||
(fdata/process-metadata cfg))
|
||||
(some->> metadata
|
||||
(fdata/decode-metadata)
|
||||
(fdata/process-metadata cfg))
|
||||
|
||||
(l/trc :obj "file-data"
|
||||
:id (str id)
|
||||
:file-id (str file-id)
|
||||
:type type
|
||||
:backend backend
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(l/trc :obj "file-data"
|
||||
:id (str id)
|
||||
:file-id (str file-id)
|
||||
:type type
|
||||
:backend backend
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
(let [affected (-> (db/delete! conn :file-data
|
||||
{:file-id file-id
|
||||
:id id
|
||||
:type type})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :file-data
|
||||
{:file-id file-id
|
||||
:id id
|
||||
:type type})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "file-data"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk rows
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(defn- delete-file-data
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-file-data-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map (fn [row] {:file-id (:file-id row) :id (:id row)}))
|
||||
(partition-all chunk-size))
|
||||
rf 0
|
||||
(db/plan conn [sql:get-file-data timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(def ^:private sql:get-file-change
|
||||
"SELECT id, file_id, deleted_at
|
||||
FROM file_change
|
||||
"SELECT id FROM file_change
|
||||
WHERE deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
ORDER BY deleted_at ASC
|
||||
LIMIT ?
|
||||
ORDER BY deleted_at ASC")
|
||||
|
||||
(def ^:private sql:lock-file-change
|
||||
"SELECT id, file_id, deleted_at FROM file_change
|
||||
WHERE id = ANY(?::uuid[])
|
||||
AND deleted_at IS NOT NULL
|
||||
AND deleted_at <= ?
|
||||
FOR UPDATE
|
||||
SKIP LOCKED")
|
||||
|
||||
(defn- delete-file-changes!
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size] :as cfg}]
|
||||
(->> (db/plan conn [sql:get-file-change timestamp chunk-size] {:fetch-size 5})
|
||||
(reduce (fn [total {:keys [id file-id deleted-at] :as xlog}]
|
||||
(l/trc :obj "file-change"
|
||||
:id (str id)
|
||||
:file-id (str file-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
(defn- delete-file-changes-chunk
|
||||
[cfg ids]
|
||||
(try
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::timestamp]}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(->> (db/exec! conn [sql:lock-file-change ids timestamp])
|
||||
(reduce (fn [total {:keys [id file-id deleted-at]}]
|
||||
(l/trc :obj "file-change"
|
||||
:id (str id)
|
||||
:file-id (str file-id)
|
||||
:deleted-at (ct/format-inst deleted-at))
|
||||
|
||||
(let [affected (-> (db/delete! conn :file-change {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))
|
||||
(let [affected (-> (db/delete! conn :file-change {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0)))))
|
||||
(catch Throwable cause
|
||||
(l/err :obj "file-change"
|
||||
:hint "error on chunk deletion, skipping it"
|
||||
:chunk ids
|
||||
:cause cause)
|
||||
0)))
|
||||
|
||||
(def ^:private deletion-proc-vars
|
||||
[#'delete-profiles!
|
||||
#'delete-file-media-objects!
|
||||
#'delete-file-object-thumbnails!
|
||||
#'delete-file-thumbnails!
|
||||
#'delete-file-data!
|
||||
#'delete-file-changes!
|
||||
#'delete-files!
|
||||
#'delete-projects!
|
||||
#'delete-fonts!
|
||||
#'delete-teams!])
|
||||
|
||||
(defn- execute-proc!
|
||||
"A generic function that executes the specified proc iterativelly
|
||||
until 0 results is returned"
|
||||
[cfg proc-fn]
|
||||
(loop [total 0]
|
||||
(let [result (db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn] :as cfg}]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(proc-fn cfg)))]
|
||||
(if (pos? result)
|
||||
(recur (long (+ total result)))
|
||||
total))))
|
||||
(defn- delete-file-changes
|
||||
[{:keys [::timestamp ::chunk-size] :as cfg}]
|
||||
(let [rf (fn ([total] total)
|
||||
([total chunk] (+ total (delete-file-changes-chunk cfg chunk))))]
|
||||
(db/tx-run! (assoc cfg ::db/read-only true)
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(transduce (comp (map :id) (partition-all chunk-size)) rf 0
|
||||
(db/plan conn [sql:get-file-change timestamp]
|
||||
{:fetch-size 5}))))))
|
||||
|
||||
(defmethod ig/assert-key ::handler
|
||||
[_ params]
|
||||
@ -327,12 +585,18 @@
|
||||
|
||||
(defmethod ig/init-key ::handler
|
||||
[_ cfg]
|
||||
(fn [_]
|
||||
(let [cfg (assoc cfg ::timestamp (ct/now))]
|
||||
(loop [procs (map deref deletion-proc-vars)
|
||||
total 0]
|
||||
(if-let [proc-fn (first procs)]
|
||||
(let [result (execute-proc! cfg proc-fn)]
|
||||
(recur (rest procs)
|
||||
(long (+ total result))))
|
||||
{:processed total})))))
|
||||
(fn [{:keys [props]}]
|
||||
(let [cfg (assoc cfg ::timestamp (ct/now))
|
||||
cfg (assoc cfg ::chunk-size (or (:chunk-size props)
|
||||
(::chunk-size cfg)))
|
||||
total (+ (delete-profiles cfg)
|
||||
(delete-file-media-objects cfg)
|
||||
(delete-file-object-thumbnails cfg)
|
||||
(delete-file-thumbnails cfg)
|
||||
(delete-file-data cfg)
|
||||
(delete-file-changes cfg)
|
||||
(delete-files cfg)
|
||||
(delete-projects cfg)
|
||||
(delete-fonts cfg)
|
||||
(delete-teams cfg))]
|
||||
{:processed total})))
|
||||
|
||||
125
backend/test/backend_tests/objects_gc_test.clj
Normal file
125
backend/test/backend_tests/objects_gc_test.clj
Normal file
@ -0,0 +1,125 @@
|
||||
;; 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 INC Sucursal en España SL
|
||||
|
||||
(ns backend-tests.objects-gc-test
|
||||
(:require
|
||||
[app.common.logging :as l]
|
||||
[app.common.time :as ct]
|
||||
[app.db :as db]
|
||||
[app.storage :as sto]
|
||||
[app.tasks.objects-gc :as ogc]
|
||||
[backend-tests.helpers :as th]
|
||||
[clojure.test :as t]))
|
||||
|
||||
(t/use-fixtures :once th/state-init)
|
||||
(t/use-fixtures :each th/database-reset)
|
||||
|
||||
(defn- create-deleted-file!
|
||||
"Creates a file and applies the deletion cascade on it (the same task the RPC
|
||||
uses), so its children are marked as deleted for the given date. Returns the
|
||||
file."
|
||||
[profile i deleted-at]
|
||||
(let [file (th/create-file* i {:profile-id (:id profile)
|
||||
:project-id (:default-project-id profile)})]
|
||||
(th/run-task! :delete-object {:object :file
|
||||
:deleted-at deleted-at
|
||||
:id (:id file)})
|
||||
file))
|
||||
|
||||
(defn- block-file!
|
||||
"Leaves an unmarked `file_media_object` pointing at the file, which is what
|
||||
happens in production: a child that the deletion cascade did not mark. Its
|
||||
foreign key towards `file` is `NO ACTION`, so it blocks the file deletion."
|
||||
[file]
|
||||
(let [media (sto/put-object! (::sto/storage th/*system*)
|
||||
{::sto/content (sto/content "media")
|
||||
:content-type "text/plain"
|
||||
:bucket "file-media-object"})]
|
||||
(th/create-file-media-object* {:file-id (:id file)
|
||||
:media-id (:id media)})))
|
||||
|
||||
(defn- find-file
|
||||
[id]
|
||||
(th/db-get :file {:id id} {::db/remove-deleted false}))
|
||||
|
||||
(defn- chunk-error
|
||||
"The first captured log record that reports a skipped chunk for `obj`,
|
||||
waiting for it because logging is asynchronous."
|
||||
[records obj]
|
||||
(let [deadline (+ (System/currentTimeMillis) 5000)]
|
||||
(loop []
|
||||
(let [record (first (filter (fn [record]
|
||||
(let [props (::l/props record)]
|
||||
(and (= obj (get props :obj))
|
||||
(= "error on chunk deletion, skipping it"
|
||||
(get props :hint)))))
|
||||
@records))]
|
||||
(cond
|
||||
record record
|
||||
(> (System/currentTimeMillis) deadline) nil
|
||||
:else (do (Thread/sleep 10) (recur)))))))
|
||||
|
||||
(t/deftest deletes-the-pending-files
|
||||
(let [profile (th/create-profile* 1)
|
||||
file (create-deleted-file! profile 1 (ct/in-past "10m"))
|
||||
result (th/run-task! :objects-gc {})]
|
||||
|
||||
(t/is (pos? (:processed result)))
|
||||
(t/is (nil? (find-file (:id file))))))
|
||||
|
||||
(t/deftest profile-cascade-keeps-profile-deletion-date
|
||||
;; The cascade marks the owned rows with the deletion date of the profile,
|
||||
;; like the RPC delete-profile flow does, not with the timestamp of the run.
|
||||
;; The team survives because only the profiles chunk runs here, so its row
|
||||
;; still shows that date.
|
||||
(let [profile (th/create-profile* 1)
|
||||
_ (th/db-update! :profile {:deleted-at (ct/in-past "30m")} {:id (:id profile)})
|
||||
|
||||
deleted-at (:deleted-at (th/db-get :profile {:id (:id profile)}
|
||||
{::db/remove-deleted false}))
|
||||
|
||||
cfg {::db/pool th/*pool*
|
||||
::sto/storage (::sto/storage th/*system*)
|
||||
::ogc/timestamp deleted-at}
|
||||
total (#'ogc/delete-profiles-chunk cfg [(:id profile)])]
|
||||
|
||||
(t/testing "the profile is gone and the chunk reports it"
|
||||
(t/is (= 1 total))
|
||||
(t/is (nil? (th/db-get :profile {:id (:id profile)} {::db/remove-deleted false}))))
|
||||
|
||||
(t/testing "the owned team keeps the deletion date of the profile"
|
||||
(t/is (= deleted-at (:deleted-at (th/db-get :team {:id (:default-team-id profile)}
|
||||
{::db/remove-deleted false})))))))
|
||||
|
||||
(t/deftest blocked-chunk-does-not-stop-the-run
|
||||
;; Chunks of one file: the blocked file takes down its own chunk and nothing
|
||||
;; else, so the clean file of the next chunk is still deleted and the run
|
||||
;; finishes. The next run retries the blocked file.
|
||||
(let [profile (th/create-profile* 1)
|
||||
blocked (create-deleted-file! profile 1 (ct/in-past "30m"))
|
||||
clean (create-deleted-file! profile 2 (ct/in-past "20m"))
|
||||
|
||||
_ (block-file! blocked)
|
||||
|
||||
records (atom [])
|
||||
_ (add-watch l/log-record ::capture
|
||||
(fn [_ _ _ record] (swap! records conj record)))
|
||||
result (try
|
||||
(th/run-task! :objects-gc {:chunk-size 1})
|
||||
(finally
|
||||
(remove-watch l/log-record ::capture)))]
|
||||
|
||||
(t/testing "the clean file of the next chunk is deleted"
|
||||
(t/is (nil? (find-file (:id clean)))))
|
||||
|
||||
(t/testing "the blocked file is still pending"
|
||||
(t/is (some? (find-file (:id blocked)))))
|
||||
|
||||
(t/testing "the failed chunk is reported"
|
||||
(t/is (some? (chunk-error records "file"))))
|
||||
|
||||
(t/testing "the run finishes"
|
||||
(t/is (pos? (:processed result))))))
|
||||
Loading…
x
Reference in New Issue
Block a user