mirror of
https://github.com/penpot/penpot.git
synced 2026-11-12 13:00:37 +00:00
34c702bc80
* ✨ Partition file_change by file_id Partition `file_change` by `HASH (file_id)` in 16 partitions, with the same bounds as `file_data`, so a join between them is resolved partition by partition. The primary key becomes `(file_id, id)`, which a partitioned table needs, and every access filters by `file_id`. The new table carries a `kind` column (1 snapshot, 2 revision) as a reserve: the queries still tell them apart by the label, and `kind` takes over when they are rewritten. The payload of a snapshot stays in `file_data`; the `data` column is kept for reading old `legacy-db` payloads. The four snapshot commands require `file-id`, because it prunes the partition. The objects GC locks and deletes by the pair, in batches. The old table is kept as `file_change_legacy` as the way back. The automatic migration refuses to run above 10 GiB or 2M rows and points to the runbook; big instances partition online with that runbook, which copies by id, follows the `updated_at` watermark in delta passes and checks that no row of the source is missing from the target. Closes #12416 AI-assisted-by: deepseek-v4.1-flash * 🐛 Address the review of the file_change partition The production runbook promised a rollback it did not have: the section was a sketch, and its statements could write to the table about to be dropped. The rollback is removed, and so is every comment that described it. `file_change_legacy` stays as a frozen copy of the previous layout. The cutover check now aborts the transaction when a row of the old table is missing in the new one, instead of leaving the decision to the operator. The five secondary indexes of the new table are built after the last delta pass: the bulk and the delta passes only need the primary key, so the batches no longer maintain them. The copy of 0158 is now replayed in a test against the pre-0158 shape with a labelled and an unlabelled row, plus an empty instance that must leave no legacy table behind. The updated_at assertion of the snapshot mutations is bounded by the test clock, and the repair revision stops writing the `features` of the file, like the normal save. AI-assisted-by: deepseek-v4.1-flash * 📎 Fix linter issues * 🐛 Fix stale helper call in rpc_file_snapshot_test The suite does not load: `th/run-pending-tasks!` does not exist in backend-tests.helpers. The pending jobs runner is `run-pending-jobs` (no arguments), the name every other call site of the suite uses. AI-assisted-by: inkling-2-small-free * 📚 Align the partition change with the 2.20.0 milestone 2.19.0 is already shipped; this work targets 2.20.0. The operations file of the partition procedure carries the 2.20.0 prefix, the references that point at it (the migration, the runbook table and the memory) follow, and the RPC doc/changes tags of the snapshot commands record 2.20 as the release that makes `file-id` required. AI-assisted-by: inkling-2-small-free * 🐛 Fix review findings on the partition rebase The 0157 legacy test was missed by the rebase and inserted file_change rows without the new mandatory `kind` column, so the backend suite fails against the migrated schema. The inserts now carry `:kind` (snapshot-kind for the labelled rows, revision-kind for the rest). The post-cutover verification block of the runbook terminates the first query with a semicolon so the block pastes into psql without a syntax error. The guard message of the migration names the step (0158-partition-file-change) instead of "0158", ambiguous since two migrations share that number. Closes #12416 AI-assisted-by: inkling-2-small-free * 🐛 Fix the stale label in the snapshot restore test The rename block of snapshots-crud runs before the restore block, so the restored version is already named "label1-updated" when the restore publishes its message. The command names the version by the label the row has at that moment, not by the one it had when it was created, so the test was asserting a stale value. Assert the current label, and say why in the comment. AI-assisted-by: deepseek-v4.1-flash * ♻️ Split the version event tests into one test per command The workspace version events and the plugin proxy send the file-id of the open file on every snapshot command, because the command needs it to prune the partition of file_change. Both tests checked that with a doseq and t/testing blocks, so a failure named the whole group and not the command that broke. Give each command its own test. The workspace ones share a helper that runs the event and returns the command and params it sends; the plugin ones keep their own mock of rp/cmd!. AI-assisted-by: deepseek-v4.1-flash
499 lines
16 KiB
Clojure
499 lines
16 KiB
Clojure
;; 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.features.file-snapshots
|
|
(:require
|
|
[app.binfile.common :as bfc]
|
|
[app.common.data :as d]
|
|
[app.common.exceptions :as ex]
|
|
[app.common.features :as-alias cfeat]
|
|
[app.common.files.migrations :as fmg]
|
|
[app.common.logging :as l]
|
|
[app.common.schema :as sm]
|
|
[app.common.time :as ct]
|
|
[app.common.uuid :as uuid]
|
|
[app.config :as cf]
|
|
[app.db :as db]
|
|
[app.db.sql :as-alias sql]
|
|
[app.features.fdata :as fdata]
|
|
[app.features.file-changes :as fchanges]
|
|
[app.jobs :as jobs]
|
|
[app.storage :as sto]
|
|
[app.util.blob :as blob]
|
|
[cuerdas.core :as str]))
|
|
|
|
(def sql:snapshots
|
|
"SELECT c.id,
|
|
c.label,
|
|
c.created_at,
|
|
c.updated_at AS modified_at,
|
|
c.deleted_at,
|
|
c.profile_id,
|
|
c.created_by,
|
|
c.locked_by,
|
|
c.revn,
|
|
c.features,
|
|
c.migrations,
|
|
c.version,
|
|
c.file_id,
|
|
c.data AS legacy_data,
|
|
fd.data AS data,
|
|
coalesce(fd.backend, 'legacy-db') AS backend,
|
|
fd.metadata AS metadata
|
|
FROM file_change AS c
|
|
LEFT JOIN file_data AS fd ON (fd.file_id = c.file_id
|
|
AND fd.id = c.id
|
|
AND fd.type = 'snapshot')
|
|
WHERE c.label IS NOT NULL")
|
|
|
|
;; `:migrations` is an ordered set everywhere (see `common.types.file`),
|
|
;; and `need-migration?`/`migrate` use set semantics on it: the pgarray
|
|
;; must never surface as a vector, where `contains?`/`set/difference`
|
|
;; would compare indexes instead of names.
|
|
(defn- decode-migrations
|
|
[value]
|
|
(into (d/ordered-set) (db/decode-pgarray value [])))
|
|
|
|
(defn- decode-snapshot
|
|
[snapshot]
|
|
(some-> snapshot
|
|
(-> (d/update-when :metadata fdata/decode-metadata)
|
|
(d/update-when :migrations decode-migrations)
|
|
(d/update-when :features db/decode-pgarray #{}))))
|
|
|
|
(def ^:private sql:get-minimal-file
|
|
"SELECT f.id,
|
|
f.revn,
|
|
f.modified_at,
|
|
f.deleted_at,
|
|
fd.backend AS backend,
|
|
fd.metadata AS metadata
|
|
FROM file AS f
|
|
LEFT JOIN file_data AS fd ON (fd.file_id = f.id AND fd.id = f.id)
|
|
WHERE f.id = ?")
|
|
|
|
(def ^:private sql:snapshot-columns
|
|
"c.id,
|
|
c.label,
|
|
c.revn,
|
|
c.created_at,
|
|
c.modified_at,
|
|
c.deleted_at,
|
|
c.profile_id,
|
|
c.created_by,
|
|
c.locked_by,
|
|
c.features,
|
|
c.metadata,
|
|
c.migrations,
|
|
c.version,
|
|
c.file_id")
|
|
|
|
(def ^:private sql:snapshot-visible
|
|
"AND CASE WHEN c.created_by = 'user'
|
|
THEN c.deleted_at IS NULL
|
|
WHEN c.created_by = 'system'
|
|
THEN c.deleted_at IS NULL OR c.deleted_at >= ?::timestamptz
|
|
END")
|
|
|
|
(def ^:private sql:get-snapshot-without-data
|
|
(str "WITH snapshots AS (" sql:snapshots ")"
|
|
"SELECT " sql:snapshot-columns
|
|
" FROM snapshots AS c
|
|
WHERE c.file_id = ?
|
|
AND c.id = ? " sql:snapshot-visible))
|
|
|
|
(defn get-minimal-snapshot
|
|
"Get the minimal snapshot info (without the data blob). The query filters
|
|
by `file_id`, which lets the planner prune the partitions of `file_change`
|
|
when it chooses an index scan. Raises `:snapshot-not-found` when there is
|
|
no matching snapshot."
|
|
[cfg file-id snapshot-id]
|
|
(let [now (ct/now)]
|
|
(or (-> (db/get-with-sql cfg [sql:get-snapshot-without-data file-id snapshot-id now]
|
|
{::db/remove-deleted false
|
|
::db/throw-if-not-exists false})
|
|
(decode-snapshot))
|
|
(ex/raise :type :not-found
|
|
:code :snapshot-not-found
|
|
:hint "unable to find snapshot with the provided id"
|
|
:snapshot-id snapshot-id
|
|
:file-id file-id))))
|
|
|
|
(def ^:private sql:get-snapshot
|
|
(str sql:snapshots
|
|
" AND c.file_id = ?
|
|
AND c.id = ? "
|
|
sql:snapshot-visible))
|
|
|
|
(defn get-snapshot
|
|
"Get a fully decoded snapshot for read-only preview or restoration.
|
|
Returns the snapshot map with decoded :data field, or nil when the file has
|
|
no snapshot with that id."
|
|
[cfg file-id snapshot-id]
|
|
(let [now (ct/now)]
|
|
(some->> (db/get-with-sql cfg [sql:get-snapshot file-id snapshot-id now]
|
|
{::db/remove-deleted false
|
|
::db/throw-if-not-exists false})
|
|
(decode-snapshot)
|
|
(fdata/resolve-file-data cfg)
|
|
(fdata/decode-file-data cfg))))
|
|
|
|
(def ^:private sql:get-visible-snapshots
|
|
(str "WITH "
|
|
"snapshots1 AS ( " sql:snapshots "),"
|
|
"snapshots2 AS (
|
|
SELECT c.id,
|
|
c.label,
|
|
c.revn,
|
|
c.version,
|
|
c.created_at,
|
|
c.modified_at,
|
|
c.created_by,
|
|
c.locked_by,
|
|
c.profile_id,
|
|
c.deleted_at
|
|
FROM snapshots1 AS c
|
|
WHERE c.file_id = ?
|
|
ORDER BY c.created_at DESC
|
|
), snapshots3 AS (
|
|
(SELECT * FROM snapshots2
|
|
WHERE created_by = 'system'
|
|
AND (deleted_at IS NULL OR
|
|
deleted_at >= ?::timestamptz)
|
|
LIMIT 500)
|
|
UNION ALL
|
|
(SELECT * FROM snapshots2
|
|
WHERE created_by = 'user'
|
|
AND deleted_at IS NULL
|
|
LIMIT 500)
|
|
)
|
|
SELECT * FROM snapshots3;"))
|
|
|
|
(defn get-visible-snapshots
|
|
"Return a list of snapshots fecheable from the API, it has a limited
|
|
set of fields and applies big but safe limits over all available
|
|
snapshots. It return a ordered vector by the snapshot date of
|
|
creation."
|
|
[cfg file-id]
|
|
(let [now (ct/now)]
|
|
(->> (db/exec! cfg [sql:get-visible-snapshots file-id now])
|
|
(mapv decode-snapshot))))
|
|
|
|
(def ^:private schema:decoded-file
|
|
[:map {:title "DecodedFile"}
|
|
[:id ::sm/uuid]
|
|
[:revn :int]
|
|
[:vern :int]
|
|
[:data :map]
|
|
[:version :int]
|
|
[:features ::cfeat/features]
|
|
[:migrations [::sm/set :string]]])
|
|
|
|
(def ^:private schema:snapshot
|
|
[:map {:title "Snapshot"}
|
|
[:id ::sm/uuid]
|
|
[:revn [::sm/int {:min 0}]]
|
|
[:version [::sm/int {:min 0}]]
|
|
[:features ::cfeat/features]
|
|
[:migrations [::sm/set ::sm/text]]
|
|
[:profile-id {:optional true} ::sm/uuid]
|
|
[:label ::sm/text]
|
|
[:file-id ::sm/uuid]
|
|
[:created-by [:enum "system" "user" "admin"]]
|
|
[:deleted-at {:optional true} ::ct/inst]
|
|
[:modified-at ::ct/inst]
|
|
[:created-at ::ct/inst]])
|
|
|
|
(def ^:private check-snapshot
|
|
(sm/check-fn schema:snapshot))
|
|
|
|
(def ^:private check-decoded-file
|
|
(sm/check-fn schema:decoded-file))
|
|
|
|
(defn- generate-snapshot-label
|
|
[]
|
|
(let [ts (-> (ct/now)
|
|
(ct/format-inst)
|
|
(str/replace #"[T:\.]" "-")
|
|
(str/rtrim "Z"))]
|
|
(str "snapshot-" ts)))
|
|
|
|
(def ^:private schema:create-params
|
|
[:map {:title "SnapshotCreateParams"}
|
|
[:profile-id ::sm/uuid]
|
|
[:created-by {:optional true} [:enum "user" "system"]]
|
|
[:label {:optional true} ::sm/text]
|
|
[:session-id {:optional true} ::sm/uuid]
|
|
[:modified-at {:optional true} ::ct/inst]
|
|
[:deleted-at {:optional true} ::ct/inst]])
|
|
|
|
(def ^:private check-create-params
|
|
(sm/check-fn schema:create-params))
|
|
|
|
(defn create!
|
|
"Create a file snapshot; expects a non-encoded file"
|
|
[cfg file & {:as params}]
|
|
(let [{:keys [label created-by deleted-at profile-id session-id]}
|
|
(check-create-params params)
|
|
|
|
file
|
|
(check-decoded-file file)
|
|
|
|
created-by
|
|
(or created-by "system")
|
|
|
|
snapshot-id
|
|
(uuid/next)
|
|
|
|
created-at
|
|
(ct/now)
|
|
|
|
deleted-at
|
|
(or deleted-at
|
|
(if (= created-by "system")
|
|
(ct/in-future (cf/get-deletion-delay))
|
|
nil))
|
|
|
|
label
|
|
(or label (generate-snapshot-label))
|
|
|
|
snapshot
|
|
(cond-> {:id snapshot-id
|
|
:revn (:revn file)
|
|
:version (:version file)
|
|
:file-id (:id file)
|
|
:features (:features file)
|
|
:migrations (:migrations file)
|
|
:label label
|
|
:created-at created-at
|
|
:modified-at created-at
|
|
:created-by created-by}
|
|
|
|
deleted-at
|
|
(assoc :deleted-at deleted-at)
|
|
|
|
:always
|
|
(check-snapshot))]
|
|
|
|
(db/insert! cfg :file-change
|
|
(-> snapshot
|
|
(update :features into-array)
|
|
(update :migrations into-array)
|
|
(assoc :kind fchanges/snapshot-kind)
|
|
(assoc :updated-at created-at)
|
|
(assoc :profile-id profile-id)
|
|
(assoc :session-id session-id)
|
|
(dissoc :modified-at))
|
|
{::db/return-keys false})
|
|
|
|
(fdata/upsert! cfg
|
|
{:id snapshot-id
|
|
:file-id (:id file)
|
|
:type "snapshot"
|
|
:data (blob/encode (:data file))
|
|
:created-at created-at
|
|
:deleted-at deleted-at})
|
|
|
|
snapshot))
|
|
|
|
(def ^:private schema:update-params
|
|
[:map {:title "SnapshotUpdateParams"}
|
|
[:id ::sm/uuid]
|
|
[:file-id ::sm/uuid]
|
|
[:label ::sm/text]
|
|
[:modified-at {:optional true} ::ct/inst]])
|
|
|
|
(def ^:private check-update-params
|
|
(sm/check-fn schema:update-params))
|
|
|
|
(defn update!
|
|
[cfg params]
|
|
|
|
(let [{:keys [id file-id label modified-at]}
|
|
(check-update-params params)
|
|
|
|
modified-at
|
|
(or modified-at (ct/now))]
|
|
|
|
(db/update! cfg :file-data
|
|
{:deleted-at nil
|
|
:modified-at modified-at}
|
|
{:file-id file-id
|
|
:id id
|
|
:type "snapshot"}
|
|
{::db/return-keys false})
|
|
|
|
(-> (db/update! cfg :file-change
|
|
{:label label
|
|
:created-by "user"
|
|
;; the partition runbook uses updated_at as the
|
|
;; watermark of its delta passes: a rename has to move
|
|
;; it, unlike file_data.modified_at above
|
|
:updated-at (ct/now)
|
|
:deleted-at nil}
|
|
{:file-id file-id
|
|
:id id}
|
|
{::db/return-keys false})
|
|
(db/get-update-count)
|
|
(pos?))))
|
|
|
|
(defn restore!
|
|
[{:keys [::db/conn] :as cfg} file-id snapshot-id]
|
|
(let [lock-sql (str sql:get-minimal-file " FOR UPDATE OF f SKIP LOCKED")
|
|
row (db/exec-one! conn [lock-sql file-id])]
|
|
|
|
(when-not row
|
|
(ex/raise :type :conflict
|
|
:code :file-locked
|
|
:hint "the file is currently locked by another operation, retry later"))
|
|
|
|
(let [file (d/update-when row :metadata fdata/decode-metadata)
|
|
vern (rand-int Integer/MAX_VALUE)
|
|
|
|
;; We reuse the main connection here for storage operations
|
|
;; becaue the main operations are touching and we need them
|
|
;; to be atomic with the current transaction
|
|
storage
|
|
(sto/resolve cfg ::db/reuse-conn true)
|
|
|
|
snapshot
|
|
(get-snapshot cfg file-id snapshot-id)]
|
|
|
|
(when-not snapshot
|
|
(ex/raise :type :not-found
|
|
:code :snapshot-not-found
|
|
:hint "unable to find snapshot with the provided id"
|
|
:snapshot-id snapshot-id
|
|
:file-id file-id))
|
|
|
|
(when-not (:data snapshot)
|
|
(ex/raise :type :internal
|
|
:code :snapshot-without-data
|
|
:hint "snapshot has no data"
|
|
:label (:label snapshot)
|
|
:file-id file-id))
|
|
|
|
(let [;; If the snapshot has applied migrations stored, we reuse
|
|
;; them, if not, we take a safest set of migrations as
|
|
;; starting point. This is because, at the time of
|
|
;; implementing snapshots, migrations were not taken into
|
|
;; account so we need to make this backward compatible in
|
|
;; some way.
|
|
migrations
|
|
(or (:migrations snapshot)
|
|
(fmg/generate-migrations-from-version 67))
|
|
|
|
file
|
|
(-> file
|
|
(update :revn inc)
|
|
(assoc :migrations migrations)
|
|
(assoc :data (:data snapshot))
|
|
(assoc :vern vern)
|
|
(assoc :version (:version snapshot))
|
|
(assoc :has-media-trimmed false)
|
|
(assoc :modified-at (:modified-at snapshot))
|
|
(assoc :features (:features snapshot)))]
|
|
|
|
(l/dbg :hint "restoring snapshot"
|
|
:file-id (str file-id)
|
|
:label (:label snapshot)
|
|
:snapshot-id (str (:id snapshot)))
|
|
|
|
;; In the same way, on reseting the file data, we need to restore
|
|
;; the applied migrations on the moment of taking the snapshot.
|
|
;; The snapshot data may hold shared pointer-mapped pages; the
|
|
;; loader stays bound while persisting so any page the encode
|
|
;; path does load keeps working (`enable-objects-map` skips the
|
|
;; unloaded ones).
|
|
(fdata/with-pointers cfg file-id
|
|
(bfc/update-file! cfg file ::bfc/reset-migrations? true))
|
|
|
|
;; FIXME: this should be separated functions, we should not have
|
|
;; inline sql here.
|
|
|
|
;; clean object thumbnails
|
|
(let [sql (str "update file_tagged_object_thumbnail "
|
|
" set deleted_at = now() "
|
|
" where file_id=? returning media_id")
|
|
res (db/exec! conn [sql file-id])]
|
|
(doseq [media-id (into #{} (keep :media-id) res)]
|
|
(sto/touch-object! storage media-id)))
|
|
|
|
;; clean file thumbnails
|
|
(let [sql (str "update file_thumbnail "
|
|
" set deleted_at = now() "
|
|
" where file_id=? returning media_id")
|
|
res (db/exec! conn [sql file-id])]
|
|
(doseq [media-id (into #{} (keep :media-id) res)]
|
|
(sto/touch-object! storage media-id)))
|
|
|
|
vern))))
|
|
|
|
(defn delete!
|
|
[cfg & {:keys [id file-id deleted-at]}]
|
|
(assert (uuid? id) "missing id")
|
|
(assert (uuid? file-id) "missing file-id")
|
|
(assert (ct/inst? deleted-at) "missing deleted-at")
|
|
|
|
(jobs/submit cfg
|
|
{::jobs/name :delete-object
|
|
::jobs/params {:object :snapshot
|
|
:deleted-at deleted-at
|
|
:file-id file-id
|
|
:id id}})
|
|
(db/update! cfg :file-change
|
|
{:deleted-at deleted-at
|
|
:updated-at (ct/now)}
|
|
{:id id :file-id file-id}
|
|
{::db/return-keys false})
|
|
true)
|
|
|
|
(def ^:private sql:get-snapshots
|
|
(str sql:snapshots " AND c.file_id = ?"))
|
|
|
|
(defn lock-by!
|
|
[conn file-id id profile-id]
|
|
(assert (uuid? file-id) "missing file-id")
|
|
(assert (uuid? id) "missing id")
|
|
(-> (db/update! conn :file-change
|
|
{:locked-by profile-id
|
|
:updated-at (ct/now)}
|
|
{:id id :file-id file-id}
|
|
{::db/return-keys false})
|
|
(db/get-update-count)
|
|
(pos?)))
|
|
|
|
(defn unlock!
|
|
[conn file-id id]
|
|
(assert (uuid? file-id) "missing file-id")
|
|
(assert (uuid? id) "missing id")
|
|
(-> (db/update! conn :file-change
|
|
{:locked-by nil
|
|
:updated-at (ct/now)}
|
|
{:id id :file-id file-id}
|
|
{::db/return-keys false})
|
|
(db/get-update-count)
|
|
(pos?)))
|
|
|
|
(defn reduce-snapshots
|
|
"Process the file snapshots using efficient reduction; the file
|
|
reduction comes with all snapshots, including maked as deleted.
|
|
|
|
Snapshot data may hold pointer-mapped pages, so the fragment loader
|
|
is bound while reducing: consumers that walk page content (media
|
|
refs, stats) transparently load fragments, consumers that only read
|
|
pointer ids never trigger a load."
|
|
[cfg file-id xform f init]
|
|
(let [conn (db/get-connection cfg)
|
|
xform (comp
|
|
(map (partial fdata/resolve-file-data cfg))
|
|
(map (partial fdata/decode-file-data cfg))
|
|
xform)]
|
|
|
|
(fdata/with-pointers cfg file-id
|
|
(->> (db/plan conn [sql:get-snapshots file-id] {:fetch-size 1})
|
|
(transduce xform f init)))))
|