github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-29 22:10:34 +00:00
Code Issues Packages Projects Releases Wiki Activity
penpot/backend/test/backend_tests/storage_test.clj
Andrey Antukh 9abcfebf60
✨ Run binfile import and export as durable jobs (#12056) 
* ✨ Run binfile import and export as durable jobs

Import and export of `.penpot` files become durable jobs: two commands
create them and answer at once with a job id the caller follows, while
a worker of the `binfile` queue does the work. Each job keeps its own
ledger (owner, params frozen at creation, result, progress history and
an expiry), and the legacy commands and their SSE keep working.

Why: both operations ran inside the request. A large import held it
open for minutes, progress only existed while the caller kept the
stream open, and a client that went away left work nobody recorded,
retried or could cancel. The unified `job` substrate already dispatched,
leased, retried and logged events, but it had no user-facing ledger.

How:
- The substrate gains the pieces a user-facing job needs: optional
  job-def metadata (family, resource-role), an expiry on submit, the
  owner in the runner context, and job resources in their own bucket,
  written with the profile of the job and never deduplicated.
- One generic command per family: the caller names the job and sends its
  business params, the registry checks the family, the job-def decodes
  and validates, and permission and quota are checked first.
- Progress taps of the core become milestones of the job: a shared stage
  vocabulary with self-contained counters, throttled heartbeats, and an
  interrupt that stops the run inside its own transaction on cancel.
- Cancel reaches pending and running jobs from both modals, and closing
  a modal mid-run cancels the job it was following.
- The export re-checks the read permission at run time and answers with
  the package descriptor; the import re-checks the edition permission,
  runs the core in a transaction so a cancellation rolls it back, and
  releases the consumed package and its temporary copy.

AI-assisted-by: deepseek-v4.1-flash

* ♻️ Handle externally-cancelled binfile jobs end to end

The `cancelled` outcome of a job nobody followed no longer hangs its
screens: the data layers emit an explicit terminal message for it.

- The export emits `{:cancelled true}` per file and the modal paints a
  neutral cancelled row instead of success; a message without an
  artifact never reaches the download branch.
- The import closes every entry with a dedicated cancelled message
  through the existing error rendering, so the wizard completes.
- Both flows are pinned by data-layer tests, including the red state
  they replace.

Also in this change: the progress-contract comment of the management
API, early business-params validation in create-import-job (before
assembling the upload), a `release-resource` helper without fabricated
job ids, a note on the heartbeat write cost, and a memory cleanup.

AI-assisted-by: muse-spark-1.3-contributor

* ♻️ Pre-translate the cancelled import message

The cancelled entry of the import wizard carried its message as a key
for the UI to translate, which needed a rehash marker comment. Like
its neighboring errors, it now translates eagerly in the data layer:
the literal call keeps rehash clean with no marker, and the tests
assert through the same call instead of the raw key.

AI-assisted-by: muse-spark-1.3-contributor

* ♻️ Move cancelled strings under generic labels keys

The dashboard-specific cancelled key never shipped: before use it moves
under `labels.*` next to the generic strings both flows already use,
with Spanish translations following the established wording.

AI-assisted-by: muse-spark-1.3-contributor

* ✨ Add minor improvements

* ♻️ Rename export job command to create-export-binfile-job

Export now has one command per job type, the shape the :export-assets
command needs: create-export-binfile-job freezes exactly the
:export-binfile job, so :name no longer travels in the body and only
:params does. Import keeps the generic envelope.

submit-job reads the queue from the job-def metadata ::jobs/queue-name
(falling back to :binfile), read by the command and never by the
substrate, so a job-def can route its work to another queue without a
change in the command.

The rename reaches the frontend and the tests: the binfile export is
asked for under its own name, and the exporter service keeps
create-export-job.

AI-assisted-by: deepseek-v4.1-flash

* ✨ Polish the import and export status rows

The status row of a file already carries what the file is doing, so the
generic "Uploading file…"/"Downloading file…" lines are gone: the
import wizard drops its own (the export one went with the jobs
refactor), along with the now-unused style.

In the export, every file is queued up front, even the ones whose job
is created only after the previous one is over, and a file stays queued
until its first milestone: a worker picking the job up no longer ends
the queue. The queued text is a step smaller than the milestone.

AI-assisted-by: deepseek-v4.1-flash

* 🐛 Show the cancelled export and fix the import error

The cancelled export was unreachable: mark-file-cancelled sets :loading
false and the label lived inside the loading branch, so a cancelled
file only painted a neutral notification. The terminal state now
carries the label, and the export flags lose the `?` suffix
(predicates keep it; a data prop is not a question).

The import wizard rendered (:error entry) through tr, but every
producer writes user-facing text (a hint or a translated label), never
a key: it renders the text as-is now, and the dynamic-key ignores and
the stale comment are gone.

The total-text-limit comment no longer references the 200MiB it
replaced; the value is 500MiB in develop already.

AI-assisted-by: deepseek-v4.1-flash

* ♻️ Reflect the real job params in the creation commands

The creation commands declared their params as an open empty map, so the
RPC contract and the generated docs said nothing about them. Each command
now imports the schema from the job namespace
(`export-binfile/schema:params`, `import-binfile/schema:create-params`),
and the RPC validates the business params before the command runs. The
job-def still owns the params and re-validates the assembled ones at run
time. The commands no longer decode the params again: the RPC did it.

Import joins export as one command per job type: `create-import-job`
becomes `create-import-binfile-job`, its body no longer carries `:name`
and the job-def is resolved by name like export's. Its creation schema
keeps the version optional, because the command reads it from the package
header when the caller does not say it, and fills the manifest metadata
kept for audit.

A version out of range is now refused by the schema (`:params-validation`)
instead of the command's own `:unsupported-version` check, which is gone.

AI-assisted-by: deepseek-v4.1-flash

* 🐛 Stop the test system from running job runners

The test system is main/system-config plus main/worker-config with a
few components dissoc'ed. The runners of default, webhook and cron were
listed by hand; the binfile runner added with the durable binfile jobs
was not, so a real penpot/job-runner/binfile/0 thread ran during the
whole suite.

That runner races the tests that drive the loop by hand (run-batch plus
run-worker-loop): it takes the job the test just dispatched and
completes it, and the test then reads a state it did not produce. That
is the reported flake of binfile-jobs-integration-test, whose middle
assertion waits for a runner of another queue to not take the job. It
only reproduced in CI: in the devenv PENPOT_TENANT makes the tenant
baked into main/worker-config differ from the one the test builds, so
the two used different queue keys locally.

Drop every [<profile> :app.worker/runner] by key shape instead of
listing the known queues, so a new queue cannot bring the race back.

AI-assisted-by: deepseek-v4.1-flash

* ✨ Upload the package one chunk at a time and as a percentage

The upload milestone counts chunks, not units of work, so the row wrote
"Uploading the package 3/9" over a file the user had uploaded once: it
read as if nine packages were on their way. Write the stage as a
percentage of the file instead ("Uploading file… (33%)"), and make the
English text of the stage say file, not package.

The chunks also go one at a time now (concat-all instead of merge-all
with two in flight), so the percentage moves in a straight line instead
of jumping. upload-blob-chunked is the generic uploader, so media and
font uploads become sequential too.

AI-assisted-by: deepseek-v4.1-flash

* 🌐 Complete the Spanish catalogue

es.po was 25 keys short of en.po: the 15 that the durable binfile jobs
introduced (14 jobs.progress.stage.* and labels.queued) plus 10 gaps
that predate them (errors.connection-error, errors.save-retrying, the
find shortcuts, workspace.header.retrying,
workspace.tokens.stroke-width and the mcp session labels).

sync -l es then sorts the entries, syncs the #: comments from en and
drops the 7 dashboard.import.progress.* keys en no longer has. The
msgid sets of en and es are now identical (0 missing, 0 extra).

AI-assisted-by: deepseek-v4.1-flash

* ♻️ Move the job status keys into the jobs namespace

The three status words the durable import and export rows render were
filed under labels.*, next to the generic labels bucket, while their
progress stages live under jobs.progress.stage.*: the strings of one
feature sat in two places.

Rename them so the whole job vocabulary is one namespace:

  labels.queued           -> jobs.queued
  labels.export-cancelled -> jobs.export-cancelled
  labels.import-cancelled -> jobs.import-cancelled

No other locale has these keys yet, so the rename costs no translation.
The convention (a key names the feature that renders it; labels.* holds
the generic reusable words) is written down in the frontend/translations
memory.

AI-assisted-by: deepseek-v4.1-flash

* 🔥 Remove the translation keys nothing references

Six keys of en.po (and of es.po, which now mirrors it) have no (tr ...)
call site left in frontend/src or common/src, and no literal reference
either:

  labels.uploading-file    a duplicate of jobs.progress.stage.upload
                           since the upload stage says "Uploading file…"
  labels.downloading-file  dead since the durable jobs replaced the old
                           import and export progress rows
  inspect.attributes.stroke.alignment.center
  inspect.attributes.stroke.alignment.inner
  inspect.attributes.stroke.alignment.outer
  workspace.toolbar.mcp-connect-here

The first two are leftovers of the durable binfile jobs, the other four
predate them. Both catalogues keep an identical msgid set afterwards.

AI-assisted-by: deepseek-v4.1-flash
2026-10-07 11:52:48 +02:00

1371 lines
62 KiB
Clojure
Raw Permalink Blame History

;; 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.storage-test
(:require
[app.common.exceptions :as ex]
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.metrics :as mtx]
[app.metrics.definition :as-alias mdef]
[app.rpc :as-alias rpc]
[app.storage :as sto]
[app.storage.fs :as-alias sto.fs]
[app.storage.gc-deleted :as gcd]
[app.storage.impl :as impl]
[app.storage.s3 :as-alias sto.s3]
[app.storage.schema :as stsch]
[backend-tests.helpers :as th]
[clojure.test :as t]
[cuerdas.core :as str]
[datoteka.fs :as fs]
[datoteka.io :as io]
[mockery.core :refer [with-mocks]]
[promesa.core :as p])
(:import
(io.prometheus.client
Counter
Counter$Child)
(org.postgresql.util
PGobject)
(software.amazon.awssdk.services.s3
S3AsyncClient)
(software.amazon.awssdk.services.s3.model
NoSuchKeyException)
(software.amazon.awssdk.services.s3.presigner
S3Presigner)))
(t/use-fixtures :once th/state-init)
(t/use-fixtures :each (th/serial
th/database-reset
th/clean-storage))
(defn configure-storage-backend
"Given storage map, returns a storage configured with the appropriate
backend for assets."
[storage]
(assoc storage ::sto/backend :fs))
(defn- poison-counter-value
"Read the process-wide gc-poison counter from the test system metrics."
[]
(let [metrics (:app.metrics/metrics th/*system*)
collector (mtx/get-collector metrics :storage-gc-poison)
instance (::mdef/instance collector)]
(.get ^Counter$Child (.labels ^Counter instance (into-array String [])))))
(t/deftest put-and-retrieve-object
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"})]
(t/is (sto/object? object))
(t/is (fs/path? (sto/get-object-path storage object)))
(t/is (nil? (:expired-at object)))
(t/is (= :fs (:backend object)))
(t/is (= "file-media-object" (:bucket (meta object))))
(t/is (= "text/plain" (:content-type (meta object))))
(t/is (= "content" (slurp (sto/get-object-data storage object))))
(t/is (= "content" (slurp (sto/get-object-path storage object))))))
(t/deftest put-and-retrieve-with-json-metadata-flag
;; End-to-end through the jsonb column: with the flag on, the row is
;; stored as plain JSON and still decodes to native types.
(binding [cf/config (assoc cf/config :storage-metadata-as-json true)]
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
profile (uuid/random)
object (sto/put-object! storage {::sto/content (sto/content "content")
:bucket "tempfile"
:content-type "application/zip"
:profile-id profile})
row (th/db-exec-one! ["select metadata::text as metadata from storage_object where id = ?"
(:id object)])]
(t/is (not (str/includes? (:metadata row) "\"~:")))
(t/is (str/includes? (:metadata row) "\"bucket\""))
(let [loaded (sto/get-object storage (:id object))]
(t/is (= "tempfile" (:bucket (meta loaded))))
(t/is (= "application/zip" (:content-type (meta loaded))))
(t/is (= profile (:profile-id (meta loaded))))))))
(t/deftest tempfile-objects-are-not-deduplicated
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "content")
(sto/wrap-with-hash "same-hash"))
object1 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
::sto/touched-at (ct/in-future {:minutes 10})
:bucket "tempfile"
:content-type "text/plain"})
object2 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
::sto/touched-at (ct/in-future {:minutes 10})
:bucket "tempfile"
:content-type "text/plain"})]
(t/is (not= (:id object1) (:id object2)))))
(t/deftest job-resource-objects-require-an-owner
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")]
(t/testing "an object of this bucket cannot be stored without its owner"
(t/is (thrown? Throwable
(sto/put-object! storage {::sto/content content
:bucket sto/job-resource-bucket
:content-type "text/plain"}))))
(t/testing "and storing it with an owner still works"
(t/is (some? (sto/put-object! storage {::sto/content content
:bucket sto/job-resource-bucket
:profile-id (uuid/next)
:content-type "text/plain"}))))))
(t/deftest job-resource-objects-are-not-deduplicated
;; An object in this bucket belongs to one profile: sharing a blob
;; between two owners would serve one user's artifact to another.
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "content")
(sto/wrap-with-hash "same-hash"))
object1 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
::sto/touched-at (ct/now)
:bucket sto/job-resource-bucket
:profile-id (uuid/next)
:content-type "text/plain"})
object2 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
::sto/touched-at (ct/now)
:bucket sto/job-resource-bucket
:profile-id (uuid/next)
:content-type "text/plain"})]
(t/is (not= (:id object1) (:id object2)))))
(t/deftest put-and-retrieve-expired-object
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
::sto/expired-at (ct/in-future {:hours 1})
:content-type "text/plain"})]
(t/is (sto/object? object))
(t/is (ct/inst? (:expired-at object)))
(t/is (ct/is-after? (:expired-at object) (ct/now)))
(t/is (nil? (sto/get-object storage (:id object))))))
(t/deftest put-and-delete-object
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"})]
(t/is (sto/object? object))
(t/is (true? (sto/del-object! storage (:id object))))
;; retrieving the same object should be not nil because the
;; deletion is not immediate
(t/is (some? (sto/get-object-data storage object)))
(t/is (some? (sto/get-object-url storage object)))
(t/is (some? (sto/get-object-path storage object)))
;; But you can't retrieve the object again because in database is
;; marked as deleted/expired.
(t/is (nil? (sto/get-object storage (:id object))))))
(t/deftest delete-expired-object-returns-false
;; An object stored with `::sto/expired-at` is born with `deleted_at`
;; set, so deleting it is a no-op and must report `false`.
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"
::sto/expired-at (ct/in-future {:hours 1})})]
(t/is (some? (:expired-at object)))
(t/is (false? (sto/del-object! storage (:id object))))))
(t/deftest deleted-gc-task
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content1 (sto/content "content1")
content2 (sto/content "content2")
content3 (sto/content "content3")
object1 (sto/put-object! storage {::sto/content content1
::sto/expired-at (ct/now)
:content-type "text/plain"})
object2 (sto/put-object! storage {::sto/content content2
::sto/expired-at (ct/in-future {:hours 2})
:content-type "text/plain"})
object3 (sto/put-object! storage {::sto/content content3
::sto/expired-at (ct/in-future {:hours 1})
:content-type "text/plain"})]
(binding [ct/*clock* (ct/fixed-clock (ct/in-future {:minutes 0}))]
(let [res (th/run-task! :storage-gc-deleted {})]
(t/is (= 1 (:deleted res)))))
(let [res (th/db-exec-one! ["select count(*) from storage_object;"])]
(t/is (= 2 (:count res))))
(binding [ct/*clock* (ct/fixed-clock (ct/in-future {:minutes 61}))]
(let [res (th/run-task! :storage-gc-deleted {})]
(t/is (= 1 (:deleted res)))))
(let [res (th/db-exec-one! ["select count(*) from storage_object;"])]
(t/is (= 1 (:count res))))))
(t/deftest touched-gc-task-1
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
prof (th/create-profile* 1)
proj (th/create-project* 1 {:profile-id (:id prof)
:team-id (:default-team-id prof)})
file (th/create-file* 1 {:profile-id (:id prof)
:project-id (:default-project-id prof)
:is-shared false})
mfile {:filename "sample.jpg"
:path (th/tempfile "backend_tests/test_files/sample.jpg")
:mtype "image/jpeg"
:size 312043}
params {::th/type :upload-file-media-object
::rpc/profile-id (:id prof)
:file-id (:id file)
:is-local true
:name "testfile"
:content mfile}
out1 (th/command! params)
out2 (th/command! params)]
(t/is (nil? (:error out1)))
(t/is (nil? (:error out2)))
(let [result-1 (:result out1)
result-2 (:result out2)]
(t/is (uuid? (:id result-1)))
(t/is (uuid? (:id result-2)))
(t/is (uuid? (:media-id result-1)))
(t/is (uuid? (:media-id result-2)))
(t/is (= (:media-id result-1) (:media-id result-2)))
(th/db-update! :file-media-object
{:deleted-at (ct/now)}
{:id (:id result-1)})
;; run the objects gc task for permanent deletion
(let [res (th/run-task! :objects-gc {})]
(t/is (= 1 (:processed res))))
;; check that we still have all the storage objects
(let [res (th/db-exec-one! ["select count(*) from storage_object"])]
(t/is (= 2 (:count res))))
;; now check if the storage objects are touched
(let [res (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"])]
(t/is (= 2 (:count res))))
;; run the touched gc task
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 2 (:freeze res)))
(t/is (= 0 (:delete res))))
;; now check that there are no touched objects
(let [res (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"])]
(t/is (= 0 (:count res))))
;; now check that all objects are marked to be deleted
(let [res (th/db-exec-one! ["select count(*) from storage_object where deleted_at is not null"])]
(t/is (= 0 (:count res)))))))
(defn- upload-font-chunked!
"Splits `font-bytes` into a single chunk, creates an upload session,
uploads the chunk, and returns the session-id UUID."
[prof ^bytes font-bytes mtype]
(let [tmp (fs/create-tempfile :dir "/tmp/penpot" :prefix "test-font-chunk-")
_ (io/write* tmp font-bytes)
mfile {:filename "chunk" :path tmp :mtype mtype :size (alength font-bytes)}
session-id (-> (th/command! {::th/type :create-upload-session
::rpc/profile-id (:id prof)
:total-chunks 1})
:result :session-id)
out (th/command! {::th/type :upload-chunk
::rpc/profile-id (:id prof)
:session-id session-id
:index 0
:content mfile})]
(assert (nil? (:error out)))
session-id))
(t/deftest touched-gc-task-2
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
prof (th/create-profile* 1 {:is-active true})
team-id (:default-team-id prof)
proj-id (:default-project-id prof)
font-id (uuid/custom 10 1)
proj (th/create-project* 1 {:profile-id (:id prof)
:team-id team-id})
file (th/create-file* 1 {:profile-id (:id prof)
:project-id proj-id
:is-shared false})
ttfdata (-> (io/resource "backend_tests/test_files/font-1.ttf")
(io/read*))
mfile {:filename "sample.jpg"
:path (th/tempfile "backend_tests/test_files/sample.jpg")
:mtype "image/jpeg"
:size 312043}
params1 {::th/type :upload-file-media-object
::rpc/profile-id (:id prof)
:file-id (:id file)
:is-local true
:name "testfile"
:content mfile}
session-id (upload-font-chunked! prof ttfdata "font/ttf")
params2 {::th/type :create-font-variant
::rpc/profile-id (:id prof)
:team-id team-id
:font-id font-id
:font-family "somefont"
:font-weight 400
:font-style "normal"
:uploads {"font/ttf" session-id}}
out1 (th/command! params1)
out2 (th/command! params2)]
;; (th/print-result! out)
(t/is (nil? (:error out1)))
(t/is (nil? (:error out2)))
;; run the touched gc task
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 5 (:freeze res)))
(t/is (= 1 (:delete res)))
(let [result-1 (:result out1)
result-2 (:result out2)]
(th/db-update! :team-font-variant
{:deleted-at (ct/now)}
{:id (:id result-2)})
;; run the objects gc task for permanent deletion
;; (processed = 2: the consumed upload session plus the font variant)
(let [res (th/run-task! :objects-gc {})]
(t/is (= 2 (:processed res))))
;; revert touched state to all storage objects
(th/db-exec-one! ["update storage_object set touched_at=?" (ct/now)])
;; Run the task again
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 2 (:freeze res)))
(t/is (= 4 (:delete res))))
;; now check that there are no touched objects
(let [res (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"])]
(t/is (= 0 (:count res))))
;; now check that all objects are marked to be deleted
(let [res (th/db-exec-one! ["select count(*) from storage_object where deleted_at is not null"])]
(t/is (= 4 (:count res))))))))
(t/deftest touched-gc-task-3
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
prof (th/create-profile* 1)
proj (th/create-project* 1 {:profile-id (:id prof)
:team-id (:default-team-id prof)})
file (th/create-file* 1 {:profile-id (:id prof)
:project-id (:default-project-id prof)
:is-shared false})
mfile {:filename "sample.jpg"
:path (th/tempfile "backend_tests/test_files/sample.jpg")
:mtype "image/jpeg"
:size 312043}
params {::th/type :upload-file-media-object
::rpc/profile-id (:id prof)
:file-id (:id file)
:is-local true
:name "testfile"
:content mfile}
out1 (th/command! params)
out2 (th/command! params)]
(t/is (nil? (:error out1)))
(t/is (nil? (:error out2)))
(let [result-1 (:result out1)
result-2 (:result out2)]
;; now we proceed to manually mark all storage objects touched
(th/db-exec! ["update storage_object set touched_at=?" (ct/now)])
;; run the touched gc task
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 2 (:freeze res)))
(t/is (= 0 (:delete res))))
;; check that we have all object in the db
(let [rows (th/db-exec! ["select * from storage_object"])]
(t/is (= 2 (count rows)))))
;; now we proceed to manually delete all file_media_object
(th/db-exec! ["update file_media_object set deleted_at = ?" (ct/now)])
(let [res (th/run-task! :objects-gc {})]
(t/is (= 2 (:processed res))))
;; run the touched gc task
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 0 (:freeze res)))
(t/is (= 2 (:delete res))))
;; check that we have all no objects
(let [rows (th/db-exec! ["select * from storage_object where deleted_at is null"])]
(t/is (= 0 (count rows))))))
(t/deftest tempfile-bucket-test
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content1 (sto/content "content1")
now (ct/now)
object1 (sto/put-object! storage {::sto/content content1
::sto/touched-at (ct/plus now {:hours 1})
:bucket "tempfile"
:content-type "text/plain"})]
;; not eligible while the touched-at is in the future
(binding [ct/*clock* (ct/fixed-clock now)]
(let [res (th/run-task! :storage-gc-touched {})]
(t/is (= 0 (:freeze res)))
(t/is (= 0 (:delete res)))))
;; still not eligible: touched-at (now+1h) is beyond the threshold
(binding [ct/*clock* (ct/fixed-clock (ct/plus now {:hours 2}))]
(let [res (th/run-task! :storage-gc-touched {})]
(t/is (= 0 (:freeze res)))
(t/is (= 0 (:delete res)))))
;; eligible: marked for deletion immediately, without any extra delay
(let [clock (ct/plus now {:hours 3})]
(binding [ct/*clock* (ct/fixed-clock clock)]
(let [res (th/run-task! :storage-gc-touched {})]
(t/is (= 0 (:freeze res)))
(t/is (= 1 (:delete res)))))
(let [row (th/db-exec-one! ["select deleted_at from storage_object where id = ?" (:id object1)])]
(t/is (ct/is-before-or-equal? (:deleted-at row) (ct/plus clock {:seconds 1})))))
;; removed on the next deleted gc run
(binding [ct/*clock* (ct/fixed-clock (ct/plus now {:hours 4}))]
(let [res (th/run-task! :storage-gc-deleted {})]
(t/is (= 1 (:deleted res)))))))
(t/deftest touched-gc-task-skip-delay
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content1")
now (ct/now)
object1 (sto/put-object! storage {::sto/content content
::sto/touched-at now
:bucket "tempfile"
:content-type "text/plain"})]
;; too recent: not processed without skip-delay
(binding [ct/*clock* (ct/fixed-clock now)]
(let [res (th/run-task! :storage-gc-touched {})]
(t/is (= 0 (:freeze res)))
(t/is (= 0 (:delete res)))))
;; processed immediately with skip-delay, bypassing min-age
(binding [ct/*clock* (ct/fixed-clock now)]
(let [res (th/run-task! :storage-gc-touched {:skip-delay true})]
(t/is (= 0 (:freeze res)))
(t/is (= 1 (:delete res)))))
;; and marked for deletion without any additional delay
(let [row (th/db-exec-one! ["select deleted_at from storage_object where id = ?" (:id object1)])]
(t/is (ct/is-before-or-equal? (:deleted-at row) (ct/plus now {:seconds 1}))))))
(t/deftest storage-gc-touched-null-metadata
;; A NULL metadata column (predates any normalization) flows through
;; the lookup fallback with a warning instead of breaking the GC loop.
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
::sto/touch true
:content-type "text/plain"})]
(th/db-exec! ["update storage_object set metadata = null where id = ?" (:id object)])
(let [res (th/run-task! :storage-gc-touched {:skip-delay true})]
(t/is (= 0 (:freeze res)))
(t/is (= 1 (:delete res))))))
(t/deftest storage-gc-touched-defers-corrupt-metadata
;; A non-map metadata row neither blocks the chunk nor gets
;; collected: it is logged and deferred exactly one day, keeping its
;; metadata intact for a later repair.
(let [now (ct/now)
before (poison-counter-value)
storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
healthy (sto/put-object! storage {::sto/content (sto/content "healthy")
::sto/touched-at now
:content-type "text/plain"})
poison (uuid/random)]
(th/db-exec! ["insert into storage_object (id, backend, metadata, touched_at) values (?, 'fs', '[]'::jsonb, ?)"
poison now])
(binding [ct/*clock* (ct/fixed-clock now)]
(let [res (th/run-task! :storage-gc-touched {:skip-delay true})]
(t/is (= 0 (:freeze res)))
(t/is (= 1 (:delete res)))))
(let [row (th/db-exec-one! ["select metadata::text as metadata, touched_at from storage_object where id = ?" poison])]
(t/is (= "[]" (:metadata row)))
;; inst-ms: timestamptz keeps micros, the frozen clock has nanos.
(t/is (= (inst-ms (ct/plus now {:days 1}))
(inst-ms (:touched-at row)))))
;; One poison row observed, one increment.
(t/is (= (inc before) (poison-counter-value)))))
(t/deftest storage-gc-touched-poison-only
;; A chunk made only of poison rows still terminates: the rows are
;; deferred, the next scan returns nothing, and no delete is counted.
(let [now (ct/now)
poison (uuid/random)]
(th/db-exec! ["insert into storage_object (id, backend, metadata, touched_at) values (?, 'fs', '[]'::jsonb, ?)"
poison now])
(binding [ct/*clock* (ct/fixed-clock now)]
(let [res (th/run-task! :storage-gc-touched {:skip-delay true})]
(t/is (= 0 (:freeze res)))
(t/is (= 0 (:delete res)))))
(let [row (th/db-exec-one! ["select touched_at from storage_object where id = ?" poison])]
(t/is (= (inst-ms (ct/plus now {:days 1}))
(inst-ms (:touched-at row)))))))
(t/deftest storage-gc-touched-drains-past-a-poison-only-batch
;; A full batch of poison rows must not stop the run: they are deferred
;; and the loop keeps draining, so a healthy row queued behind the
;; LIMIT (later touched_at) still gets collected.
(let [now (ct/now)
storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
earlier (ct/minus now {:seconds 1})
healthy (sto/put-object! storage {::sto/content (sto/content "healthy")
::sto/touched-at now
:content-type "text/plain"})
poisons (repeatedly 10 uuid/random)]
(doseq [id poisons]
(th/db-exec! ["insert into storage_object (id, backend, metadata, touched_at) values (?, 'fs', '[]'::jsonb, ?)"
id earlier]))
(binding [ct/*clock* (ct/fixed-clock now)]
(let [res (th/run-task! :storage-gc-touched {:skip-delay true})]
(t/is (= 0 (:freeze res)))
(t/is (= 1 (:delete res)))))
(doseq [id poisons]
(let [row (th/db-exec-one! ["select touched_at from storage_object where id = ?" id])]
(t/is (= (inst-ms (ct/plus now {:days 1}))
(inst-ms (:touched-at row))))))))
(def ^:private migration-0155-fixtures
;; [id transit-metadata]: production-shaped legacy rows (nil payload
;; means a NULL column).
[["11111111-1111-1111-1111-111111111111"
"{\"~:reference\":\"~:file-media-object\",\"~:content-type\":\"image/png\",\"~:hash\":\"blake2b:aaa\"}"]
["22222222-2222-2222-2222-222222222222"
"{\"~:content-type\":\"image/svg+xml\"}"]
["33333333-3333-3333-3333-333333333333"
"{\"~:bucket\":\"tempfile\",\"~:reference\":\"~:tempfile\",\"~:content-type\":\"application/zip\"}"]
["44444444-4444-4444-4444-444444444444"
"{\"~:bucket\":\"tempfile\",\"~:content-type\":\"application/zip\",\"~:upload-id\":\"~u86907e95-1cb8-8122-8008-4eb7ba07d89d\",\"~:chunk-index\":3}"]
["55555555-5555-5555-5555-555555555555"
nil]])
(defn- run-migration-0155!
;; Re-runs the 0155 statements (not migratus: it already applied at
;; bootstrap) over the fixture rows above.
[]
(let [sql (-> (io/resource "app/migrations/sql/0155-normalize-storage-object-metadata.sql")
(slurp))
no-comments (->> (.split ^String sql "\n")
(remove #(.startsWith ^String (str/trim %) "--"))
(str/join "\n"))]
(doseq [stmt (->> (.split ^String no-comments ";")
(map str/trim)
(remove str/blank?))]
(th/db-exec! [stmt]))))
(defn- get-metadata-by-id
[id]
(:metadata (th/db-exec-one! ["select metadata from storage_object where id = ?"
(parse-uuid id)])))
(t/deftest storage-migration-0155-normalizes-legacy-rows
(doseq [[id mdata] migration-0155-fixtures]
(th/db-exec! ["insert into storage_object (id, backend, metadata) values (?, 'fs', ?::jsonb)"
(parse-uuid id) mdata]))
(run-migration-0155!)
(let [mdata (fn [id] (stsch/decode-metadata (get-metadata-by-id id)))]
(t/is (= {:bucket "file-media-object"
:content-type "image/png"
:hash "blake2b:aaa"}
(mdata "11111111-1111-1111-1111-111111111111")))
(t/is (= {:bucket "file-media-object"
:content-type "image/svg+xml"}
(mdata "22222222-2222-2222-2222-222222222222")))
(t/is (= {:bucket "tempfile"
:content-type "application/zip"}
(mdata "33333333-3333-3333-3333-333333333333")))
(t/is (= {:bucket "tempfile"
:content-type "application/zip"}
(mdata "44444444-4444-4444-4444-444444444444")))
(t/is (= {:bucket "file-media-object"}
(mdata "55555555-5555-5555-5555-555555555555"))))
;; second run changes nothing (idempotent)
(let [raw (fn [] (mapv #(.getValue ^PGobject (get-metadata-by-id %))
(map first migration-0155-fixtures)))
before (raw)]
(run-migration-0155!)
(t/is (= before (raw)))))
(t/deftest storage-gc-deleted-immediate
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content1")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"})]
;; mark as deleted right now
(th/db-exec! ["update storage_object set deleted_at = ?" (ct/now)])
;; the deleted gc removes it on the next run
(let [res (th/run-task! :storage-gc-deleted {})]
(t/is (= 1 (:deleted res))))))
(t/deftest storage-gc-deleted-commits-each-chunk-separately
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
;; more than one chunk (chunk-size is 25); distinct contents so
;; the storage layer does not deduplicate them into one row
ids (doall (for [i (range 26)]
(:id (sto/put-object! storage
{::sto/content (sto/content (str "content" i))
:content-type "text/plain"}))))]
(t/is (= 26 (count ids)))
(th/db-exec! ["update storage_object set deleted_at = ?" (ct/now)])
(let [orig @#'gcd/process-chunk
calls (atom 0)]
(alter-var-root #'gcd/process-chunk
(constantly (fn [& args]
(when (= 2 (swap! calls inc))
(throw (ex-info "boom" {})))
(apply orig args))))
(try
(t/is (thrown? Exception (th/run-task! :storage-gc-deleted {})))
(t/testing "first chunk committed before the fault"
(t/is (= 1 (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM storage_object WHERE deleted_at IS NOT NULL"])))))
(finally
(alter-var-root #'gcd/process-chunk (constantly orig)))))))
(t/deftest objects-gc-task-skip-delay
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
prof (th/create-profile* 1)
proj (th/create-project* 1 {:profile-id (:id prof)
:team-id (:default-team-id prof)})
file (th/create-file* 1 {:profile-id (:id prof)
:project-id (:default-project-id prof)
:is-shared false})
mfile {:filename "sample.jpg"
:path (th/tempfile "backend_tests/test_files/sample.jpg")
:mtype "image/jpeg"
:size 312043}
params {::th/type :upload-file-media-object
::rpc/profile-id (:id prof)
:file-id (:id file)
:is-local true
:name "testfile"
:content mfile}
out1 (th/command! params)
out2 (th/command! params)]
(t/is (nil? (:error out1)))
(t/is (nil? (:error out2)))
(let [result-1 (:result out1)
result-2 (:result out2)]
;; mark as deleted but in the future (not yet eligible)
(th/db-update! :file-media-object
{:deleted-at (ct/in-future {:days 1})}
{:id (:id result-1)})
;; without skip-delay the future deleted row is not processed
(let [res (th/run-task! :objects-gc {})]
(t/is (= 0 (:processed res))))
;; with skip-delay the future deleted row IS processed immediately
(let [res (th/run-task! :objects-gc {:skip-delay true})]
(t/is (= 1 (:processed res)))
(t/is (nil? (th/db-get :file-media-object {:id (:id result-1)})))))))
(t/deftest put-object-write-failure-leaves-pending-row
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
;; Point the fs backend at a path that is actually a file so the
;; blob write fails.
blocked (fs/path "/tmp/penpot" (str "blocked-" (uuid/next)))
_ (spit (str blocked) "x")]
(try
(let [broken (assoc-in storage [::sto/backends :fs ::sto.fs/directory] (str blocked))
content (sto/content "content")
ex (try
(sto/put-object! broken {::sto/content content
:content-type "text/plain"})
nil
(catch Throwable cause cause))]
(t/is (some? ex))
;; the pending row stays behind and is reclaimed asynchronously
;; by the :storage-pending-gc task
(let [rows (th/db-query :storage-object {:status "pending"})]
(t/is (= 1 (count rows)))
(th/db-update! :storage-object
{:created-at (ct/in-past {:days 2})}
{:id (:id (first rows))})
(let [res (th/run-task! :storage-pending-gc {})]
(t/is (= 1 (:processed res))))
(let [row (th/db-exec-one! ["select count(*) from storage_object"])]
(t/is (= 0 (:count row))))))
(finally
(fs/delete blocked)))))
(t/deftest pending-gc-reclaims-unpromoted-object
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"})
path (sto/get-object-path storage object)]
;; valid objects are never reclaimed
(let [res (th/run-task! :storage-pending-gc {})]
(t/is (= 0 (:processed res))))
;; simulate a crash: the object was created but never promoted
(th/db-update! :storage-object {:status "pending"
:created-at (ct/in-past {:days 2})}
{:id (:id object)})
(t/is (fs/exists? path))
(let [res (th/run-task! :storage-pending-gc {})]
(t/is (= 1 (:processed res))))
;; both the row and the orphaned blob are removed
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" (:id object)])]
(t/is (= 0 (:count row))))
(t/is (not (fs/exists? path)))))
(t/deftest pending-objects-excluded-from-gc-touched
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
::sto/touched-at (ct/now)
:content-type "text/plain"})]
;; mark it pending and touched in the past
(th/db-update! :storage-object {:status "pending"
:touched-at (ct/in-past {:days 1})}
{:id (:id object)})
(binding [ct/*clock* (ct/fixed-clock (ct/now))]
(let [res (th/run-task! :storage-gc-touched {})]
(t/is (= 0 (:freeze res)))
(t/is (= 0 (:delete res)))))
;; still present and not marked as deleted
(let [row (th/db-exec-one! ["select * from storage_object where id = ?" (:id object)])]
(t/is (some? row))
(t/is (nil? (:deleted-at row))))))
(t/deftest pending-objects-excluded-from-get
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"})]
(t/is (some? (sto/get-object storage (:id object))))
(th/db-update! :storage-object {:status "pending"} {:id (:id object)})
(t/is (nil? (sto/get-object storage (:id object))))))
(t/deftest pending-objects-excluded-from-dedup
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "content")
(sto/wrap-with-hash "same-hash"))
object1 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"})]
;; mark the only matching row as pending
(th/db-update! :storage-object {:status "pending"} {:id (:id object1)})
(let [object2 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"})]
(t/is (not= (:id object1) (:id object2))))))
(t/deftest dedup-reuses-existing-blob
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "content")
(sto/wrap-with-hash "same-hash"))
object1 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"})
object2 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"})]
(t/is (= (:id object1) (:id object2)))
(let [row (th/db-exec-one! ["select count(*) from storage_object"])]
(t/is (= 1 (:count row))))))
(t/deftest dedup-reuses-json-encoded-blob
;; With the JSON flag on, the dedup lookup must find the row it just
;; wrote; a "~:"-only lookup used to miss it and duplicate the blob.
(binding [cf/config (assoc cf/config :storage-metadata-as-json true)]
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "json-content")
(sto/wrap-with-hash "json-hash"))
params {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"}
object1 (sto/put-object! storage params)
object2 (sto/put-object! storage params)]
(t/is (= (:id object1) (:id object2)))
(let [row (th/db-exec-one! ["select count(*) from storage_object"])]
(t/is (= 1 (:count row)))))))
(t/deftest dedup-json-put-finds-transit-row
;; Both encodings coexist during the transition: a JSON write must
;; reuse a blob already stored as Transit.
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "mixed-content")
(sto/wrap-with-hash "mixed-hash"))
params {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"}
object1 (binding [cf/config (assoc cf/config :storage-metadata-as-json nil)]
(sto/put-object! storage params))
object2 (binding [cf/config (assoc cf/config :storage-metadata-as-json true)]
(sto/put-object! storage params))]
(t/is (= (:id object1) (:id object2)))
(let [row (th/db-exec-one! ["select count(*) from storage_object"])]
(t/is (= 1 (:count row))))))
(t/deftest dedup-repairs-stale-object
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "content")
(sto/wrap-with-hash "same-hash"))
object1 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"})]
;; remove the physical blob to simulate a stale/broken object
(let [path (sto/get-object-path storage object1)]
(fs/delete path))
;; re-uploading identical content repairs the same reference in place
(let [object2 (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"})]
(t/is (= (:id object1) (:id object2)))
;; the row stays live: no tombstone and no extra row
(let [row (th/db-exec-one! ["select status, deleted_at from storage_object where id = ?" (:id object1)])]
(t/is (= "valid" (:status row)))
(t/is (nil? (:deleted-at row))))
(let [row (th/db-exec-one! ["select count(*) from storage_object"])]
(t/is (= 1 (:count row))))
;; the repaired blob is readable again under the original id
(t/is (= "content" (slurp (sto/get-object-data storage object2)))))))
(t/deftest gc-deleted-removes-broken-object
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"})]
;; mark as deleted and remove the physical blob
(th/db-update! :storage-object {:deleted-at (ct/in-past {:minutes 1})}
{:id (:id object)})
(let [path (sto/get-object-path storage object)]
(fs/delete path))
;; the deleted gc removes the row without error even though the blob is
;; missing (the physical deletion is best-effort)
(let [res (th/run-task! :storage-gc-deleted {})]
(t/is (= 1 (:deleted res))))
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" (:id object)])]
(t/is (= 0 (:count row))))))
(t/deftest pending-objects-excluded-from-gc-deleted
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"})]
;; mark as pending + deleted in the past
(th/db-update! :storage-object {:status "pending"
:deleted-at (ct/in-past {:minutes 1})}
{:id (:id object)})
;; gc-deleted skips it because status != 'valid'
(let [res (th/run-task! :storage-gc-deleted {})]
(t/is (= 0 (:deleted res))))
;; row still exists (with deleted_at set — we set it above)
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?"
(:id object)])]
(t/is (= 1 (:count row))))))
(t/deftest gc-deleted-gives-up-after-max-attempts
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (sto/content "content")
object (sto/put-object! storage {::sto/content content
:content-type "text/plain"})]
(th/db-update! :storage-object {:deleted-at (ct/in-past {:minutes 1})
:deletion_attempts 6}
{:id (:id object)})
(with-mocks [_mock {:target 'app.storage.impl/del-objects-in-bulk
:return (fn [_ ids] (set ids))}]
(let [res (th/run-task! :storage-gc-deleted {})]
(t/is (= 0 (:deleted res)))))
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" (:id object)])]
(t/is (= 0 (:count row))))))
(t/deftest dedup-reuses-existing-blob-with-touch
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "content")
(sto/wrap-with-hash "same-hash"))
t0 (ct/now)
params {::sto/deduplicate? true
::sto/touch true
:bucket "file-media-object"
:content-type "text/plain"}
object1 (binding [ct/*clock* (ct/fixed-clock t0)]
(sto/put-object! storage (assoc params ::sto/content content)))]
;; a touched hit reuses the object and updates its touched_at
(let [object2 (binding [ct/*clock* (ct/fixed-clock (ct/plus t0 {:hours 1}))]
(sto/put-object! storage (assoc params ::sto/content content)))]
(t/is (= (:id object1) (:id object2)))
(let [row (th/db-exec-one! ["select touched_at from storage_object where id = ?" (:id object1)])]
(t/is (ct/is-after? (:touched-at row) t0))))
;; with the blob removed, the touched hit repairs the stale row in
;; place: the same id is kept, the row is not deleted and touched_at
;; is left untouched (the touch flag only applies to healthy hits)
(let [path (sto/get-object-path storage object1)]
(fs/delete path))
(let [object3 (binding [ct/*clock* (ct/fixed-clock (ct/plus t0 {:hours 2}))]
(sto/put-object! storage (assoc params ::sto/content content)))]
(t/is (= (:id object1) (:id object3)))
(let [row (th/db-exec-one! ["select deleted_at, touched_at from storage_object where id = ?" (:id object1)])]
(t/is (nil? (:deleted-at row)))
;; the touch flag does not apply to repairs: touched_at was last
;; set by the healthy hit and is not bumped by the repair
(t/is (ct/is-before? (:touched-at row) (ct/plus t0 {:hours 2})))))))
(t/deftest put-object-repair-failure-leaves-row-intact
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
content (-> (sto/content "content")
(sto/wrap-with-hash "same-hash"))
object (sto/put-object! storage {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"})
path (sto/get-object-path storage object)
;; Point the fs backend at a path that is actually a file so the
;; blob write fails.
blocked (fs/path "/tmp/penpot" (str "blocked-" (uuid/next)))
_ (spit (str blocked) "x")]
(try
;; remove the physical blob to simulate a stale/broken object
(fs/delete path)
(let [broken (assoc-in storage [::sto/backends :fs ::sto.fs/directory] (str blocked))
ex (try
(sto/put-object! broken {::sto/content content
::sto/deduplicate? true
:bucket "file-media-object"
:content-type "text/plain"})
nil
(catch Throwable cause cause))]
(t/is (some? ex))
;; the failed repair leaves the original row exactly as it was:
;; live and valid, so a later upload can retry the healing
(let [row (th/db-exec-one! ["select status, deleted_at from storage_object where id = ?" (:id object)])]
(t/is (= "valid" (:status row)))
(t/is (nil? (:deleted-at row))))
(let [row (th/db-exec-one! ["select count(*) from storage_object"])]
(t/is (= 1 (:count row)))))
(finally
(fs/delete blocked)))))
(t/deftest upload-chunks-exclude-pending
(let [prof (th/create-profile* 1)
_ (th/create-project* 1 {:profile-id (:id prof)
:team-id (:default-team-id prof)})
file (th/create-file* 1 {:profile-id (:id prof)
:project-id (:default-project-id prof)
:is-shared false})
mfile {:filename "chunk"
:path (th/tempfile "backend_tests/test_files/sample.jpg")
:mtype "image/jpeg"
:size 312043}
session-id (-> (th/command! {::th/type :create-upload-session
::rpc/profile-id (:id prof)
:total-chunks 1})
:result :session-id)
out (th/command! {::th/type :upload-chunk
::rpc/profile-id (:id prof)
:session-id session-id
:index 0
:content mfile})]
(t/is (nil? (:error out)))
;; mark all the chunks of this session as pending (simulates rows that
;; were never promoted)
(th/db-exec! ["update storage_object set status = 'pending' where id in (select object_id from upload_session_chunk where session_id = ?)"
session-id])
;; assembling fails because no chunk is visible anymore
(let [assemble-out (th/command! {::th/type :assemble-file-media-object
::rpc/profile-id (:id prof)
:session-id session-id
:file-id (:id file)
:is-local true
:name "assembled-image"
:mtype "image/jpeg"})]
(t/is (some? (:error assemble-out))))))
(t/deftest upload-session-stalled-purge-lifecycle
;; Full lifecycle of a stalled session: objects-gc purges the session and
;; its mappings while touching the objects, touched-gc marks them deleted
;; and deleted-gc removes rows and blobs.
(let [prof (th/create-profile* 1)
_ (th/create-project* 1 {:profile-id (:id prof)
:team-id (:default-team-id prof)})
_ (th/create-file* 1 {:profile-id (:id prof)
:project-id (:default-project-id prof)
:is-shared false})
mfile {:filename "chunk"
:path (th/tempfile "backend_tests/test_files/sample.jpg")
:mtype "image/jpeg"
:size 312043}
session-id (-> (th/command! {::th/type :create-upload-session
::rpc/profile-id (:id prof)
:total-chunks 1})
:result :session-id)
out (th/command! {::th/type :upload-chunk
::rpc/profile-id (:id prof)
:session-id session-id
:index 0
:content mfile})]
(t/is (nil? (:error out)))
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
session-id]))))
;; backdate the session so it counts as stalled
(th/db-exec! ["update upload_session set created_at = now() - interval '2 hours' where id = ?"
session-id])
;; objects-gc purges session and mappings, touching the objects
(let [res (th/run-task! :objects-gc {})]
(t/is (= 1 (:processed res))))
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
session-id]))))
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
session-id]))))
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"]))))
;; touched-gc marks the orphaned object as deleted
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 0 (:freeze res)))
(t/is (= 1 (:delete res))))
;; deleted-gc removes the row and the blob (clock past the mark time)
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 4}))]
(th/run-task! :storage-gc-deleted {}))]
(t/is (= 1 (:deleted res))))
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from storage_object"]))))))
(t/deftest upload-session-consumed-purge
;; An assembled session is marked as consumed and objects-gc purges it
;; right away, without waiting for the stalled threshold.
(let [prof (th/create-profile* 1)
_ (th/create-project* 1 {:profile-id (:id prof)
:team-id (:default-team-id prof)})
file (th/create-file* 1 {:profile-id (:id prof)
:project-id (:default-project-id prof)
:is-shared false})
mfile {:filename "chunk"
:path (th/tempfile "backend_tests/test_files/sample.jpg")
:mtype "image/jpeg"
:size 312043}
session-id (-> (th/command! {::th/type :create-upload-session
::rpc/profile-id (:id prof)
:total-chunks 1})
:result :session-id)
out (th/command! {::th/type :upload-chunk
::rpc/profile-id (:id prof)
:session-id session-id
:index 0
:content mfile})]
(t/is (nil? (:error out)))
(let [assemble-out (th/command! {::th/type :assemble-file-media-object
::rpc/profile-id (:id prof)
:session-id session-id
:file-id (:id file)
:is-local true
:name "assembled-image"
:mtype "image/jpeg"})]
(t/is (nil? (:error assemble-out))))
;; mappings stay, session row stays marked as consumed; objects-gc
;; purges both
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
session-id]))))
(t/is (some? (:deleted-at (th/db-exec-one! ["select deleted_at from upload_session where id = ?"
session-id]))))
;; objects-gc purges the consumed session immediately
(let [res (th/run-task! :objects-gc {})]
(t/is (= 1 (:processed res))))
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
session-id]))))
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
session-id]))))))
(t/deftest upload-session-profile-purge
;; Sessions owned by a profile pending purge are drained first, so the
;; profile delete (which cascades to its sessions) never hits the chunk
;; RESTRICT foreign keys.
(let [prof (th/create-profile* 1)
mfile {:filename "chunk"
:path (th/tempfile "backend_tests/test_files/sample.jpg")
:mtype "image/jpeg"
:size 312043}
session-id (-> (th/command! {::th/type :create-upload-session
::rpc/profile-id (:id prof)
:total-chunks 1})
:result :session-id)
out (th/command! {::th/type :upload-chunk
::rpc/profile-id (:id prof)
:session-id session-id
:index 0
:content mfile})]
(t/is (nil? (:error out)))
;; soft-delete the profile; the live session is neither consumed nor stalled
(th/db-update! :profile {:deleted-at (ct/now)} {:id (:id prof)})
(th/run-task! :objects-gc {})
;; session and mappings are gone, profile row deletes cleanly
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
session-id]))))
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
session-id]))))
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from profile where id = ?"
(:id prof)]))))
;; and the chunk object was touched for the storage GC
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"]))))))
(defn- fake-s3-backend
[]
{::sto/type :s3
::sto.s3/client (reify S3AsyncClient)
::sto.s3/presigner (reify S3Presigner)})
(t/deftest s3-exists-object-returns-true-on-found
(with-mocks [mock {:target 'app.storage.s3/head-object
:return (p/resolved {})}]
(t/is (true? (impl/exists-object? (fake-s3-backend) {:id (uuid/next)})))
(t/is (= 1 (:call-count @mock)))))
(t/deftest s3-exists-object-returns-false-on-missing-key
(with-mocks [mock {:target 'app.storage.s3/head-object
:return (p/rejected (-> (NoSuchKeyException/builder)
(.message "no key")
(.build)))}]
(t/is (false? (impl/exists-object? (fake-s3-backend) {:id (uuid/next)})))
;; a missing key is a definitive answer: no retries
(t/is (= 1 (:call-count @mock)))))
(t/deftest s3-exists-object-retries-transient-errors
(let [calls (atom 0)]
(with-mocks [_mock {:target 'app.storage.s3/head-object
:return (fn [& _]
(swap! calls inc)
(if (< @calls 3)
(p/rejected (RuntimeException. "boom"))
(p/resolved {})))}]
(t/is (true? (impl/exists-object? (fake-s3-backend) {:id (uuid/next)})))
(t/is (= 3 @calls)))))
(t/deftest s3-exists-object-throws-after-retries-exhausted
(with-mocks [mock {:target 'app.storage.s3/head-object
:return (p/rejected (RuntimeException. "boom"))}]
;; p/await returns the rejection wrapped in an ExecutionException
(let [ex (try
(impl/exists-object? (fake-s3-backend) {:id (uuid/next)})
nil
(catch Throwable cause cause))]
(t/is (some? ex))
(t/is (= "boom" (ex-message (ex-cause ex)))))
;; one initial attempt plus max-retries
(t/is (= 4 (:call-count @mock)))))
(t/deftest touched-gc-job-resource-bucket
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))
job-id (uuid/next)
object (sto/put-object! storage {::sto/content (sto/content "content")
:content-type "text/plain"
:profile-id (uuid/next)
:bucket sto/job-resource-bucket})]
;; the object is created with the job-resource bucket in metadata
(t/is (= sto/job-resource-bucket (:bucket (meta object))))
;; a live job row referencing the object keeps it frozen
(th/db-update! :storage-object {:touched-at (ct/now)} {:id (:id object)})
(th/db-insert! :job {:id job-id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "running"
:resource-id (:id object)
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 1 (:freeze res)))
(t/is (= 0 (:delete res)))
(t/is (nil? (:touched-at (th/db-get :storage-object {:id (:id object)}
:id :touched-at))))
(t/is (nil? (:deleted-at (th/db-get :storage-object {:id (:id object)}
:id :deleted-at)))))
;; once the referencing row is gone, the touched object becomes
;; eligible for deletion (the job table is deletion-protected)
(th/db-force-delete :job {:id job-id})
(th/db-update! :storage-object {:touched-at (ct/now)} {:id (:id object)})
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 0 (:freeze res)))
(t/is (= 1 (:delete res)))
(t/is (nil? (:touched-at (th/db-get :storage-object {:id (:id object)}
:id :touched-at))))
(t/is (some? (:deleted-at (th/db-get :storage-object
{:id (:id object)}
{::db/remove-deleted false})))))
;; a touched object with no referencing job row at all is deleted too
(let [object (sto/put-object! storage {::sto/content (sto/content "content")
:content-type "text/plain"
:profile-id (uuid/next)
:bucket sto/job-resource-bucket})]
(th/db-update! :storage-object {:touched-at (ct/now)} {:id (:id object)})
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
(th/run-task! :storage-gc-touched {}))]
(t/is (= 0 (:freeze res)))
(t/is (>= (:delete res) 1))
(t/is (some? (:deleted-at (th/db-get :storage-object
{:id (:id object)}
{::db/remove-deleted false}))))))))
Reference in New Issue View Git Blame Copy Permalink