mirror of
https://github.com/penpot/penpot.git
synced 2026-10-30 06:20:32 +00:00
🐛 Keep objects whose storage target is no longer configured
An object whose :storage-target id is not in the S3 routes configuration can no longer be located. Reads and serving now fail with :invalid-storage-target instead of silently using the default bucket, and garbage collection refuses to delete the row: it logs an error and parks the row (deleted_at +1 day, no deletion attempts, no give-up) until the target is configured again. Legacy rows (no target) and the default target keep working and deleting normally. Also addresses review findings: per-target dedup isolation and stale-repair coverage, legacy-row GC coverage through the real task paths, a single declared-target schema, closing already-built S3 clients when target init fails, and loader edge pins. Docs and memory updated. Closes #11630 AI-assisted-by: deepseek-flash
This commit is contained in:
parent
0d855b3436
commit
6eece49f69
@ -24,9 +24,13 @@
|
||||
- The reserved `:default` target is implicit and built from `PENPOT_OBJECTS_STORAGE_S3_*`; declared targets inherit missing region/endpoint/prefix from it.
|
||||
- Without a routes file, `::sto/bucket->target` is nil and every object uses `:default` (unchanged behavior).
|
||||
- The chosen target id is stored in object metadata as `:storage-target` (plain string) by `put-object!`.
|
||||
- `app.storage.s3/resolve-target` reads the object metadata; missing/unknown ids fall back to `:default`.
|
||||
- One S3 client/presigner is built per distinct `[region endpoint]` and shared by targets.
|
||||
- Target ids are logged in metadata, so they must stay stable; removing one makes its old rows unreadable/unGC-able.
|
||||
- `app.storage.s3/resolve-target` reads the object metadata; `nil` (legacy) and `"default"` use the default target, an unknown non-nil id raises `:invalid-storage-target` (no fallback) on reads/serving/deletes.
|
||||
- `impl/target-resolvable?` (wrapped as `sto/target-resolvable?`) reports whether a target id is configured; `:fs` is always true.
|
||||
- GC-deleted and pending-gc refuse to delete rows whose target is not resolvable: they log `:err`, park the row (`deleted_at = now()+1d`, no attempts, no give-up) and never remove it until the target is configured again.
|
||||
- `deleted_at` doubles as the pending-gc park marker; the pending selection skips rows whose `deleted_at` is in the future.
|
||||
- One S3 client/presigner is built per distinct `[region endpoint]` and shared by targets; a failed init closes the already-built pairs.
|
||||
- Target ids are stored in metadata, so they must stay stable; removing one makes its old rows unreadable and unGC-able by design.
|
||||
- `:storage-target` metadata is load-bearing: `pending-gc` passes it via `with-meta` so `del-object` resolves the right target.
|
||||
|
||||
## Object Lifecycle
|
||||
|
||||
|
||||
@ -329,6 +329,16 @@
|
||||
(ct/is-after? (:expired-at object) (ct/now))))
|
||||
(-> (impl/get-object-url backend object nil) file-url->path))))
|
||||
|
||||
(defn target-resolvable?
|
||||
"Returns true when the backend referenced by `backend-id` can resolve
|
||||
`target` to a real destination. GC callers must refuse to delete objects
|
||||
whose target is not resolvable, so a misconfigured target never removes
|
||||
the database row while orphaning the blob."
|
||||
[storage backend-id target]
|
||||
(assert (valid-storage? storage))
|
||||
(-> (impl/resolve-backend storage backend-id)
|
||||
(impl/target-resolvable? target)))
|
||||
|
||||
(defn del-object!
|
||||
[storage object-or-id]
|
||||
(assert (valid-storage? storage))
|
||||
|
||||
@ -30,15 +30,18 @@
|
||||
[app.common.uri :as u]
|
||||
[app.config :as cf]
|
||||
[app.storage :as sto]
|
||||
[app.storage.s3 :as sto.s3]
|
||||
[clojure.edn :as edn]
|
||||
[clojure.java.io :as io]))
|
||||
|
||||
(def ^:private schema:target
|
||||
[:map {:title "s3-target"}
|
||||
[:bucket :string]
|
||||
[:region {:optional true} :keyword]
|
||||
[:endpoint {:optional true} [:or :string ::sm/uri]]
|
||||
[:prefix {:optional true} :string]])
|
||||
"The EDN declares targets with the same shape as the S3 backend, except
|
||||
the endpoint is accepted as a plain string and normalized to a URI by
|
||||
`normalize-target`."
|
||||
[:merge
|
||||
sto.s3/schema:target
|
||||
[:map
|
||||
[:endpoint {:optional true} [:or :string ::sm/uri]]]])
|
||||
|
||||
(def ^:private schema:file
|
||||
[:map {:title "storage-routes"}
|
||||
|
||||
@ -153,3 +153,7 @@
|
||||
(conj fail-ids id)))))
|
||||
#{} ids)))
|
||||
|
||||
(defmethod impl/target-resolvable? :fs
|
||||
[_backend _target]
|
||||
true)
|
||||
|
||||
|
||||
@ -78,6 +78,29 @@
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(db/exec-one! conn [sql:delete-give-up ids max-attempts])))
|
||||
|
||||
(defn- log-refusal!
|
||||
"Indirection over the error log so tests can capture the refusal payload."
|
||||
[backend-id target ids]
|
||||
(l/err :hint "storage target is not configured, deletion refused"
|
||||
:backend (name backend-id)
|
||||
:target target
|
||||
:ids (mapv str ids)))
|
||||
|
||||
(def ^:private sql:defer-unresolvable
|
||||
"UPDATE storage_object
|
||||
SET deleted_at = NOW() + INTERVAL '1 day'
|
||||
WHERE id = ANY(?::uuid[])")
|
||||
|
||||
(defn- park-unresolvable!
|
||||
"Refuses to delete rows whose target id is not configured: logs the
|
||||
misconfiguration and pushes `deleted_at` forward so the rows leave the
|
||||
selection window without being deleted and without counting a deletion
|
||||
attempt (they are therefore never subject to the give-up window)."
|
||||
[conn backend-id target ids]
|
||||
(log-refusal! backend-id target ids)
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(db/exec-one! conn [sql:defer-unresolvable ids])))
|
||||
|
||||
(defn- process-chunk
|
||||
"Attempt to delete a chunk of storage objects from a specific backend.
|
||||
|
||||
@ -142,19 +165,27 @@
|
||||
|
||||
(defn- clean-deleted!
|
||||
[cfg]
|
||||
(loop [total 0]
|
||||
(let [deleted (db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::sto/storage]}]
|
||||
(let [chunk (get-deleted-chunk conn chunk-size)]
|
||||
(when (seq chunk)
|
||||
(let [by-route (group-by-route chunk)]
|
||||
(reduce-kv (fn [acc [backend-id target] ids]
|
||||
(+ acc (process-chunk conn storage backend-id target ids)))
|
||||
0
|
||||
by-route))))))]
|
||||
(if deleted
|
||||
(recur (+ total deleted))
|
||||
total))))
|
||||
(loop [deleted 0
|
||||
parked 0]
|
||||
(let [result (db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn ::sto/storage]}]
|
||||
(let [chunk (get-deleted-chunk conn chunk-size)]
|
||||
(when (seq chunk)
|
||||
(let [by-route (group-by-route chunk)]
|
||||
(reduce-kv
|
||||
(fn [acc [backend-id target] ids]
|
||||
(if (sto/target-resolvable? storage backend-id target)
|
||||
(update acc :deleted + (process-chunk conn storage backend-id target ids))
|
||||
(do
|
||||
(park-unresolvable! conn backend-id target ids)
|
||||
(update acc :parked + (count ids)))))
|
||||
{:deleted 0 :parked 0}
|
||||
by-route))))))]
|
||||
(if result
|
||||
(recur (+ deleted (:deleted result))
|
||||
(+ parked (:parked result)))
|
||||
{:deleted deleted
|
||||
:parked parked}))))
|
||||
|
||||
(defmethod ig/assert-key ::handler
|
||||
[_ params]
|
||||
@ -164,6 +195,7 @@
|
||||
(defmethod ig/init-key ::handler
|
||||
[_ cfg]
|
||||
(fn [_]
|
||||
(let [total (clean-deleted! cfg)]
|
||||
(l/inf :hint "task finished" :total total)
|
||||
{:deleted total})))
|
||||
(let [{:keys [deleted parked]} (clean-deleted! cfg)]
|
||||
(l/inf :hint "task finished" :total deleted :parked parked)
|
||||
{:deleted deleted
|
||||
:parked parked})))
|
||||
|
||||
@ -92,6 +92,17 @@
|
||||
:code :invalid-storage-backend
|
||||
:context cfg))
|
||||
|
||||
(defmulti target-resolvable?
|
||||
"Returns true when the backend can resolve `target` to a real destination.
|
||||
GC callers must refuse deletion (and keep the row) when this is false."
|
||||
(fn [cfg _] (::sto/type cfg)))
|
||||
|
||||
(defmethod target-resolvable? :default
|
||||
[cfg _]
|
||||
(ex/raise :type :internal
|
||||
:code :invalid-storage-backend
|
||||
:context cfg))
|
||||
|
||||
;; --- HELPERS
|
||||
|
||||
(defn uuid->hex
|
||||
|
||||
@ -25,6 +25,7 @@
|
||||
FROM storage_object
|
||||
WHERE status = 'pending'
|
||||
AND created_at <= now() - interval '24 hours'
|
||||
AND (deleted_at IS NULL OR deleted_at <= now())
|
||||
ORDER BY created_at ASC
|
||||
LIMIT ?
|
||||
FOR UPDATE
|
||||
@ -37,25 +38,71 @@
|
||||
(def ^:private sql:delete-pending-sobject
|
||||
"DELETE FROM storage_object WHERE id = ? AND status = 'pending'")
|
||||
|
||||
(defn- log-refusal!
|
||||
"Indirection over the error log so tests can capture the refusal payload."
|
||||
[backend-id target ids]
|
||||
(l/err :hint "storage target is not configured, deletion refused"
|
||||
:backend (name backend-id)
|
||||
:target target
|
||||
:ids (mapv str ids)))
|
||||
|
||||
(def ^:private sql:park-unresolvable
|
||||
"UPDATE storage_object
|
||||
SET deleted_at = NOW() + INTERVAL '1 day'
|
||||
WHERE id = ANY(?::uuid[])
|
||||
AND status = 'pending'")
|
||||
|
||||
(defn- park-unresolvable!
|
||||
"Refuses to delete rows whose target id is not configured: logs the
|
||||
misconfiguration and pushes `deleted_at` forward so the rows are excluded
|
||||
from the next selection without being deleted."
|
||||
[conn backend-id target ids]
|
||||
(log-refusal! backend-id target ids)
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(db/exec-one! conn [sql:park-unresolvable ids])))
|
||||
|
||||
(def ^:private chunk-size
|
||||
100)
|
||||
|
||||
(defn- group-by-route
|
||||
[rows]
|
||||
(group-by (fn [{:keys [backend target]}]
|
||||
[(keyword backend) target])
|
||||
rows))
|
||||
|
||||
(defn- delete-pending-rows!
|
||||
"Select, lock and delete a chunk of pending rows in a single transaction.
|
||||
Returns the deleted rows or nil when there is nothing left to reclaim."
|
||||
Rows whose target id is not configured are never deleted: they are parked
|
||||
(error logged, `deleted_at` pushed forward) so the loop terminates and the
|
||||
rows survive.
|
||||
|
||||
Returns a map `{:deleted rows :parked count}`, or nil when there is nothing
|
||||
left to reclaim."
|
||||
[cfg]
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(fn [{:keys [::db/conn ::sto/storage]}]
|
||||
;; NOTE: db/exec! returns an empty vector when there are no
|
||||
;; rows left; use not-empty to detect it.
|
||||
(when-let [chunk (not-empty (get-pending-chunk conn chunk-size))]
|
||||
(doseq [{:keys [id]} chunk]
|
||||
(db/exec-one! conn [sql:delete-pending-sobject id]))
|
||||
chunk))))
|
||||
(reduce-kv
|
||||
(fn [acc [backend-id target] rows]
|
||||
(if (sto/target-resolvable? storage backend-id target)
|
||||
(do
|
||||
(doseq [{:keys [id]} rows]
|
||||
(db/exec-one! conn [sql:delete-pending-sobject id]))
|
||||
(update acc :deleted into rows))
|
||||
(do
|
||||
(park-unresolvable! conn backend-id target (mapv :id rows))
|
||||
(update acc :parked + (count rows)))))
|
||||
{:deleted [] :parked 0}
|
||||
(group-by-route chunk))))))
|
||||
|
||||
(defn- delete-blobs!
|
||||
"Best-effort removal of the orphaned blobs. Runs after the pending rows
|
||||
have been committed so a failure here never blocks their reclamation."
|
||||
have been committed so a failure here never blocks their reclamation.
|
||||
|
||||
The `:storage-target` metadata is load-bearing: the S3 backend reads it
|
||||
(`s3/target-id`) to resolve the bucket/prefix/client for `del-object`."
|
||||
[storage rows]
|
||||
(doseq [{:keys [id backend target]} rows]
|
||||
(try
|
||||
@ -69,12 +116,16 @@
|
||||
|
||||
(defn- process!
|
||||
[{::sto/keys [storage] :as cfg}]
|
||||
(loop [total 0]
|
||||
(if-let [rows (delete-pending-rows! cfg)]
|
||||
(do
|
||||
(delete-blobs! storage rows)
|
||||
(recur (long (+ total (count rows)))))
|
||||
total)))
|
||||
(loop [total-deleted 0
|
||||
total-parked 0]
|
||||
(if-let [result (delete-pending-rows! cfg)]
|
||||
(let [removed (:deleted result)
|
||||
parked (:parked result)]
|
||||
(delete-blobs! storage removed)
|
||||
(recur (long (+ total-deleted (count removed)))
|
||||
(long (+ total-parked parked))))
|
||||
{:processed total-deleted
|
||||
:parked total-parked})))
|
||||
|
||||
(defmethod ig/assert-key ::handler
|
||||
[_ params]
|
||||
@ -84,6 +135,7 @@
|
||||
(defmethod ig/init-key ::handler
|
||||
[_ cfg]
|
||||
(fn [_]
|
||||
(let [total (process! cfg)]
|
||||
(l/inf :hint "task finished" :total total)
|
||||
{:processed total})))
|
||||
(let [{:keys [processed parked]} (process! cfg)]
|
||||
(l/inf :hint "task finished" :total processed :parked parked)
|
||||
{:processed processed
|
||||
:parked parked})))
|
||||
|
||||
@ -88,7 +88,7 @@
|
||||
|
||||
;; --- BACKEND INIT
|
||||
|
||||
(def ^:private schema:target
|
||||
(def schema:target
|
||||
[:map {:title "s3-target"}
|
||||
[:bucket ::sm/text]
|
||||
[:region {:optional true} :keyword]
|
||||
@ -123,6 +123,17 @@
|
||||
:presigner presigner
|
||||
:close-fn #(.close ^java.lang.AutoCloseable client)}))
|
||||
|
||||
(defn- build-client-pair-or-cleanup
|
||||
"Builds a client pair, closing the pairs already built when the build
|
||||
fails so a failed backend init does not leak clients."
|
||||
[acc params region endpoint]
|
||||
(try
|
||||
(build-client-pair params region endpoint)
|
||||
(catch Throwable cause
|
||||
(doseq [f (:close-fns acc)]
|
||||
(ex/ignoring (f)))
|
||||
(throw cause))))
|
||||
|
||||
(defn- build-targets
|
||||
"Resolves the implicit `:default` target plus the declared targets, sharing
|
||||
one S3 client/presigner pair per distinct `[region endpoint]`."
|
||||
@ -140,7 +151,7 @@
|
||||
(fn [acc id {:keys [region endpoint bucket prefix]}]
|
||||
(let [k [region endpoint]
|
||||
pair (or (get-in acc [:pairs k])
|
||||
(build-client-pair params region endpoint))
|
||||
(build-client-pair-or-cleanup acc params region endpoint))
|
||||
acc (cond-> acc
|
||||
(nil? (get-in acc [:pairs k]))
|
||||
(-> (assoc-in [:pairs k] pair)
|
||||
@ -209,10 +220,9 @@
|
||||
(keyword? target-id) target-id
|
||||
:else (keyword target-id))]
|
||||
(or (get (::targets backend) tid)
|
||||
(get (::targets backend) (::default-target backend))
|
||||
(ex/raise :type :internal
|
||||
:code :invalid-storage-target
|
||||
:hint "storage target not configured"
|
||||
:hint "storage target is not configured"
|
||||
:target target-id
|
||||
:available (vec (keys (::targets backend)))))))
|
||||
|
||||
@ -220,6 +230,14 @@
|
||||
[backend object]
|
||||
(resolve-target-by-id backend (target-id backend object)))
|
||||
|
||||
(defmethod impl/target-resolvable? :s3
|
||||
[backend target-id]
|
||||
(let [tid (cond
|
||||
(nil? target-id) (::default-target backend)
|
||||
(keyword? target-id) target-id
|
||||
:else (keyword target-id))]
|
||||
(contains? (::targets backend) tid)))
|
||||
|
||||
;; --- API IMPL
|
||||
|
||||
(defmethod impl/put-object :s3
|
||||
|
||||
@ -79,3 +79,19 @@
|
||||
(t/is (some? (load-error
|
||||
"{:targets {:temp {:bucket \"x\"}} :routes {\"tempfile\" :temp}}"
|
||||
{:objects-storage-backend :fs}))))
|
||||
|
||||
(t/deftest rejects-route-pointing-to-default
|
||||
(let [err (load-error
|
||||
"{:targets {:temp {:bucket \"x\"}} :routes {\"tempfile\" :default}}"
|
||||
{})]
|
||||
(t/is (some? err))
|
||||
(t/is (= :unknown-storage-target (:code (ex-data err))))))
|
||||
|
||||
(t/deftest loads-empty-targets-map
|
||||
(let [file (write-routes! "{:targets {}}")]
|
||||
(try
|
||||
(let [result (load-with {:objects-storage-s3-routes-file (.getAbsolutePath file)})]
|
||||
(t/is (= {} (:targets result)))
|
||||
(t/is (nil? (:routes result))))
|
||||
(finally
|
||||
(delete! file)))))
|
||||
|
||||
@ -879,15 +879,40 @@
|
||||
(t/is (= "temp" (::sto.s3/bucket @captured)))
|
||||
(t/is (= "tmp/" (::sto.s3/prefix @captured))))))
|
||||
|
||||
(t/deftest s3-bulk-delete-falls-back-to-default-target
|
||||
(t/deftest s3-bulk-delete-rejects-unknown-target
|
||||
(let [backend (fake-s3-backend-with-targets)
|
||||
captured (atom nil)]
|
||||
(with-mocks [_mock {:target 'app.storage.s3/del-object-in-bulk
|
||||
:return (fn [target _ids]
|
||||
(reset! captured target)
|
||||
(p/resolved nil))}]
|
||||
(impl/del-objects-in-bulk backend :missing #{(uuid/next)})
|
||||
(t/is (= "default" (::sto.s3/bucket @captured))))))
|
||||
(let [ex (try
|
||||
(impl/del-objects-in-bulk backend :missing #{(uuid/next)})
|
||||
nil
|
||||
(catch Throwable cause cause))]
|
||||
(t/is (some? ex))
|
||||
(t/is (= :invalid-storage-target (:code (ex-data ex))))
|
||||
(t/is (nil? @captured))))))
|
||||
|
||||
(t/deftest s3-get-object-url-rejects-unknown-target
|
||||
(let [backend (fake-s3-backend-with-targets)
|
||||
ex (try
|
||||
(impl/get-object-url backend {:id (uuid/next)
|
||||
:storage-target "ghost"} {})
|
||||
nil
|
||||
(catch Throwable cause cause))]
|
||||
(t/is (some? ex))
|
||||
(t/is (= :invalid-storage-target (:code (ex-data ex))))))
|
||||
|
||||
(t/deftest s3-target-resolvable-checks-configured-targets
|
||||
(let [backend (fake-s3-backend-with-targets)]
|
||||
(t/is (true? (impl/target-resolvable? backend nil)))
|
||||
(t/is (true? (impl/target-resolvable? backend "default")))
|
||||
(t/is (true? (impl/target-resolvable? backend :temp)))
|
||||
(t/is (false? (impl/target-resolvable? backend "ghost")))))
|
||||
|
||||
(t/deftest fs-target-is-always-resolvable
|
||||
(t/is (true? (impl/target-resolvable? {::sto/type :fs} "anything"))))
|
||||
|
||||
(t/deftest s3-build-targets-shares-clients-and-closes-them
|
||||
(let [clients (atom [])
|
||||
@ -923,6 +948,34 @@
|
||||
(ig/halt-key! :app.storage.s3/backend backend)
|
||||
(t/is (= 2 @close-count))))))
|
||||
|
||||
(t/deftest s3-build-targets-closes-clients-on-failure
|
||||
(let [close-count (atom 0)
|
||||
mk-client (fn [params]
|
||||
(when (= :us-east-1 (:app.storage.s3/region params))
|
||||
(throw (RuntimeException. "boom")))
|
||||
(reify
|
||||
clojure.lang.IDeref
|
||||
(deref [_] (Object.))
|
||||
java.lang.AutoCloseable
|
||||
(close [_] (swap! close-count inc))))]
|
||||
(with-mocks [_c {:target 'app.storage.s3/build-s3-client
|
||||
:return mk-client}
|
||||
_p {:target 'app.storage.s3/build-s3-presigner
|
||||
:return (fn [_] (reify S3Presigner))}]
|
||||
(let [ex (try
|
||||
(ig/init-key :app.storage.s3/backend
|
||||
{:app.storage.s3/region :eu-central-1
|
||||
:app.storage.s3/bucket "main"
|
||||
:app.worker/netty-io-executor :executor
|
||||
:app.storage.s3/targets
|
||||
{:same {:bucket "same"}
|
||||
:other {:bucket "other" :region :us-east-1}}})
|
||||
nil
|
||||
(catch Throwable cause cause))]
|
||||
(t/is (some? ex))
|
||||
;; the pair built before the failing one is closed
|
||||
(t/is (= 1 @close-count))))))
|
||||
|
||||
(t/deftest s3-exists-object-returns-true-on-found
|
||||
(with-mocks [mock {:target 'app.storage.s3/head-object
|
||||
:return (p/resolved {})}]
|
||||
@ -1000,20 +1053,43 @@
|
||||
(t/is (nil? (:target row)))))
|
||||
|
||||
(t/deftest dedup-is-isolated-per-storage-target
|
||||
(let [routed (configure-s3-storage {"file-data" :temp})
|
||||
unrouted (configure-s3-storage {})
|
||||
content (-> (sto/content "content")
|
||||
(sto/wrap-with-hash "same-hash"))]
|
||||
(with-mocks [_mock {:target 'app.storage.impl/put-object
|
||||
:return (fn [_ object _] object)}]
|
||||
(let [object1 (sto/put-object! routed {::sto/content content
|
||||
::sto/deduplicate? true
|
||||
:bucket "file-data"
|
||||
:content-type "text/plain"})
|
||||
object2 (sto/put-object! unrouted {::sto/content content
|
||||
::sto/deduplicate? true
|
||||
:bucket "file-data"
|
||||
:content-type "text/plain"})
|
||||
row (th/db-exec-one! ["select count(*) from storage_object"])]
|
||||
;; same semantic bucket, different target: no dedup hit, two rows
|
||||
(t/is (not= (:id object1) (:id object2)))
|
||||
(t/is (= 2 (:count row)))))))
|
||||
|
||||
(t/deftest dedup-reuses-object-within-same-storage-target
|
||||
(let [storage (configure-s3-storage {"file-data" :temp})
|
||||
content (-> (sto/content "content")
|
||||
(sto/wrap-with-hash "same-hash"))]
|
||||
(with-mocks [_mock {:target 'app.storage.impl/put-object
|
||||
:return (fn [_ object _] object)}]
|
||||
(with-mocks [_p {:target 'app.storage.impl/put-object
|
||||
:return (fn [_ object _] object)}
|
||||
_e {:target 'app.storage.impl/exists-object?
|
||||
:return (fn [_ _] true)}]
|
||||
(let [object1 (sto/put-object! storage {::sto/content content
|
||||
::sto/deduplicate? true
|
||||
:bucket "file-data"
|
||||
:content-type "text/plain"})
|
||||
object2 (sto/put-object! storage {::sto/content content
|
||||
::sto/deduplicate? true
|
||||
:bucket "file-media-object"
|
||||
:bucket "file-data"
|
||||
:content-type "text/plain"})]
|
||||
(t/is (not= (:id object1) (:id object2)))))))
|
||||
;; same semantic bucket and target: dedup reuses the object
|
||||
(t/is (= (:id object1) (:id object2)))))))
|
||||
|
||||
(t/deftest dedup-hit-carries-storage-target-metadata
|
||||
(let [storage (configure-s3-storage {"file-data" :temp})
|
||||
@ -1037,6 +1113,37 @@
|
||||
(t/is (some? @captured))
|
||||
(t/is (= "temp" (:storage-target (meta @captured)))))))
|
||||
|
||||
(t/deftest dedup-repair-carries-storage-target-metadata
|
||||
(let [storage (configure-s3-storage {"file-data" :temp})
|
||||
content (-> (sto/content "content")
|
||||
(sto/wrap-with-hash "same-hash"))
|
||||
calls (atom [])]
|
||||
(with-mocks [_p {:target 'app.storage.impl/put-object
|
||||
:return (fn [_ object _]
|
||||
(swap! calls conj object)
|
||||
object)}
|
||||
_h {:target 'app.storage.s3/head-object
|
||||
:return (p/rejected (-> (NoSuchKeyException/builder)
|
||||
(.message "no key")
|
||||
(.build)))}]
|
||||
(let [object1 (sto/put-object! storage {::sto/content content
|
||||
::sto/deduplicate? true
|
||||
:bucket "file-data"
|
||||
:content-type "text/plain"})
|
||||
;; second put finds the row but the real exists-object? sees a
|
||||
;; missing blob and repairs it in place
|
||||
object2 (sto/put-object! storage {::sto/content content
|
||||
::sto/deduplicate? true
|
||||
:bucket "file-data"
|
||||
:content-type "text/plain"})
|
||||
row (th/db-exec-one!
|
||||
["select status from storage_object where id = ?" (:id object1)])
|
||||
count (th/db-exec-one! ["select count(*) from storage_object"])]
|
||||
(t/is (= (:id object1) (:id object2)))
|
||||
(t/is (= "valid" (:status row)))
|
||||
(t/is (= 1 (:count count)))
|
||||
(t/is (= "temp" (:storage-target (meta (last @calls)))))))))
|
||||
|
||||
;; --- GC target routing
|
||||
|
||||
(defn- storage-with-s3-targets
|
||||
@ -1059,11 +1166,85 @@
|
||||
:return (fn [_ target _ids]
|
||||
(reset! captured target)
|
||||
#{})}]
|
||||
(t/is (= 1 (#'sto.gc-deleted/clean-deleted! cfg)))
|
||||
(t/is (= 1 (:deleted (#'sto.gc-deleted/clean-deleted! cfg))))
|
||||
(t/is (= "temp" @captured)))))
|
||||
|
||||
(t/deftest pending-gc-deletes-from-routed-target
|
||||
(t/deftest gc-deleted-refuses-unknown-target-and-keeps-row
|
||||
(let [storage (storage-with-s3-targets)
|
||||
cfg {::db/pool th/*pool* ::sto/storage storage}
|
||||
id (uuid/next)
|
||||
logged (atom nil)]
|
||||
(th/db-exec! ["insert into storage_object (id, size, backend, metadata, deleted_at, status)
|
||||
values (?, 1, 's3', ?, ?, 'valid')"
|
||||
id
|
||||
(db/tjson {:bucket "file-data" :storage-target "ghost"})
|
||||
(ct/in-past {:minutes 1})])
|
||||
(with-mocks [mock {:target 'app.storage.impl/del-objects-in-bulk
|
||||
:return (fn [_ _ _] #{})}]
|
||||
(with-redefs [app.storage.gc-deleted/log-refusal!
|
||||
(fn [backend-id target ids]
|
||||
(reset! logged [backend-id target (vec ids)]))]
|
||||
(let [result (#'sto.gc-deleted/clean-deleted! cfg)
|
||||
row (th/db-exec-one!
|
||||
["select status, deleted_at, deletion_attempts
|
||||
from storage_object where id = ?" id])]
|
||||
(t/is (= 0 (:deleted result)))
|
||||
(t/is (= 1 (:parked result)))
|
||||
(t/is (= 0 (:call-count @mock)))
|
||||
(t/is (= "valid" (:status row)))
|
||||
(t/is (ct/is-after? (:deleted-at row) (ct/now)))
|
||||
(t/is (= 0 (:deletion-attempts row)))
|
||||
(t/is (= [:s3 "ghost" [id]] @logged)))))))
|
||||
|
||||
(t/deftest gc-deleted-give-up-not-applied-to-unknown-target
|
||||
(let [storage (storage-with-s3-targets)
|
||||
cfg {::db/pool th/*pool* ::sto/storage storage}
|
||||
id (uuid/next)]
|
||||
(th/db-exec! ["insert into storage_object (id, size, backend, metadata, deleted_at, status, deletion_attempts)
|
||||
values (?, 1, 's3', ?, ?, 'valid', 10)"
|
||||
id
|
||||
(db/tjson {:bucket "file-data" :storage-target "ghost"})
|
||||
(ct/in-past {:minutes 1})])
|
||||
(with-mocks [_mock {:target 'app.storage.impl/del-objects-in-bulk
|
||||
:return (fn [_ _ _] #{})}]
|
||||
(with-redefs [app.storage.gc-deleted/log-refusal! (fn [& _] nil)]
|
||||
(#'sto.gc-deleted/clean-deleted! cfg)))
|
||||
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" id])]
|
||||
(t/is (= 1 (:count row))))))
|
||||
|
||||
(t/deftest gc-deleted-normal-failure-defers-and-gives-up
|
||||
(let [storage (storage-with-s3-targets)
|
||||
cfg {::db/pool th/*pool* ::sto/storage storage}
|
||||
id (uuid/next)]
|
||||
(th/db-exec! ["insert into storage_object (id, size, backend, metadata, deleted_at, status)
|
||||
values (?, 1, 's3', ?, ?, 'valid')"
|
||||
id
|
||||
(db/tjson {:bucket "file-data" :storage-target "temp"})
|
||||
(ct/in-past {:minutes 1})])
|
||||
(with-mocks [_mock {:target 'app.storage.impl/del-objects-in-bulk
|
||||
:return (fn [_ _ ids] (set ids))}]
|
||||
(let [result (#'sto.gc-deleted/clean-deleted! cfg)
|
||||
row (th/db-exec-one!
|
||||
["select deleted_at, deletion_attempts
|
||||
from storage_object where id = ?" id])]
|
||||
(t/is (= 0 (:deleted result)))
|
||||
(t/is (= 0 (:parked result)))
|
||||
(t/is (ct/is-after? (:deleted-at row) (ct/now)))
|
||||
(t/is (= 1 (:deletion-attempts row))))
|
||||
|
||||
;; force the give-up threshold and let the next pass remove the row
|
||||
(th/db-update! :storage-object
|
||||
{:deletion-attempts 7
|
||||
:deleted-at (ct/in-past {:minutes 1})}
|
||||
{:id id})
|
||||
(let [result (#'sto.gc-deleted/clean-deleted! cfg)]
|
||||
(t/is (= 0 (:deleted result)))
|
||||
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" id])]
|
||||
(t/is (= 0 (:count row))))))))
|
||||
|
||||
(t/deftest pending-gc-deletes-resolvable-target
|
||||
(let [storage (storage-with-s3-targets)
|
||||
cfg {::db/pool th/*pool* ::sto/storage storage}
|
||||
id (uuid/next)
|
||||
captured (atom nil)]
|
||||
(th/db-exec! ["insert into storage_object (id, size, backend, metadata, created_at, status)
|
||||
@ -1071,13 +1252,84 @@
|
||||
id
|
||||
(db/tjson {:storage-target "temp"})
|
||||
(ct/in-past {:days 2})])
|
||||
(let [rows (th/db-exec!
|
||||
["select id, backend,
|
||||
coalesce(metadata->>'~:storage-target','default') as target
|
||||
from storage_object where id = ?" id])]
|
||||
(with-mocks [_mock {:target 'app.storage.impl/del-object
|
||||
:return (fn [_ object]
|
||||
(reset! captured object)
|
||||
nil)}]
|
||||
(#'sto.pending-gc/delete-blobs! storage rows)
|
||||
(t/is (= "temp" (:storage-target (meta @captured))))))))
|
||||
(with-mocks [_mock {:target 'app.storage.impl/del-object
|
||||
:return (fn [_ object]
|
||||
(reset! captured object)
|
||||
nil)}]
|
||||
(let [result (#'sto.pending-gc/process! cfg)]
|
||||
(t/is (= 1 (:processed result)))
|
||||
(t/is (= 0 (:parked result)))
|
||||
(t/is (= "temp" (:storage-target (meta @captured))))
|
||||
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" id])]
|
||||
(t/is (= 0 (:count row))))))))
|
||||
|
||||
(t/deftest pending-gc-refuses-unknown-target-and-keeps-row
|
||||
(let [storage (storage-with-s3-targets)
|
||||
cfg {::db/pool th/*pool* ::sto/storage storage}
|
||||
id (uuid/next)
|
||||
logged (atom nil)
|
||||
captured (atom nil)]
|
||||
(th/db-exec! ["insert into storage_object (id, size, backend, metadata, created_at, status)
|
||||
values (?, 1, 's3', ?, ?, 'pending')"
|
||||
id
|
||||
(db/tjson {:storage-target "ghost"})
|
||||
(ct/in-past {:days 2})])
|
||||
(with-mocks [mock {:target 'app.storage.impl/del-object
|
||||
:return (fn [_ object]
|
||||
(reset! captured object)
|
||||
nil)}]
|
||||
(with-redefs [app.storage.pending-gc/log-refusal!
|
||||
(fn [backend-id target ids]
|
||||
(reset! logged [backend-id target (vec ids)]))]
|
||||
(let [result (#'sto.pending-gc/process! cfg)
|
||||
row (th/db-exec-one!
|
||||
["select status, deleted_at from storage_object where id = ?" id])]
|
||||
(t/is (= 0 (:processed result)))
|
||||
(t/is (= 1 (:parked result)))
|
||||
(t/is (= 0 (:call-count @mock)))
|
||||
(t/is (nil? @captured))
|
||||
(t/is (= "pending" (:status row)))
|
||||
(t/is (ct/is-after? (:deleted-at row) (ct/now)))
|
||||
(t/is (= [:s3 "ghost" [id]] @logged)))))))
|
||||
|
||||
(t/deftest gc-deleted-legacy-rows-delete-from-default-target
|
||||
(let [storage (storage-with-s3-targets)
|
||||
cfg {::db/pool th/*pool* ::sto/storage storage}
|
||||
id (uuid/next)
|
||||
captured (atom nil)]
|
||||
(th/db-exec! ["insert into storage_object (id, size, backend, metadata, deleted_at, status)
|
||||
values (?, 1, 's3', ?, ?, 'valid')"
|
||||
id
|
||||
(db/tjson {:bucket "file-data"})
|
||||
(ct/in-past {:minutes 1})])
|
||||
(with-mocks [_mock {:target 'app.storage.impl/del-objects-in-bulk
|
||||
:return (fn [_ target _ids]
|
||||
(reset! captured target)
|
||||
#{})}]
|
||||
(let [result (#'sto.gc-deleted/clean-deleted! cfg)]
|
||||
(t/is (= 1 (:deleted result)))
|
||||
(t/is (= 0 (:parked result)))
|
||||
(t/is (= "default" @captured))
|
||||
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" id])]
|
||||
(t/is (= 0 (:count row))))))))
|
||||
|
||||
(t/deftest pending-gc-legacy-rows-delete-from-default-target
|
||||
(let [storage (storage-with-s3-targets)
|
||||
cfg {::db/pool th/*pool* ::sto/storage storage}
|
||||
id (uuid/next)
|
||||
captured (atom nil)]
|
||||
(th/db-exec! ["insert into storage_object (id, size, backend, metadata, created_at, status)
|
||||
values (?, 1, 's3', ?, ?, 'pending')"
|
||||
id
|
||||
(db/tjson {:bucket "file-data"})
|
||||
(ct/in-past {:days 2})])
|
||||
(with-mocks [_mock {:target 'app.storage.impl/del-object
|
||||
:return (fn [_ object]
|
||||
(reset! captured object)
|
||||
nil)}]
|
||||
(let [result (#'sto.pending-gc/process! cfg)]
|
||||
(t/is (= 1 (:processed result)))
|
||||
(t/is (= 0 (:parked result)))
|
||||
(t/is (= "default" (:storage-target (meta @captured))))
|
||||
(let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" id])]
|
||||
(t/is (= 0 (:count row))))))))
|
||||
|
||||
@ -584,8 +584,15 @@ Targets and routing are declared in an external EDN file referenced by the
|
||||
objects use the default bucket and the behavior is unchanged.
|
||||
|
||||
The target id is stored in the object metadata, so it must stay stable. If you
|
||||
remove or rename a target id, objects already written with it can no longer be
|
||||
read or deleted.
|
||||
remove or rename a target id, objects already written with it cannot be located:
|
||||
reads and serving of those objects fail, and the garbage-collection tasks refuse
|
||||
to delete them. Those rows are parked for one day and retried on later runs,
|
||||
without counting deletion attempts and without ever reaching the give-up window,
|
||||
so the row is never removed until the routes file declares the target id again.
|
||||
|
||||
All backend replicas must mount the same routes file with the same target ids. A
|
||||
replica with a stale file misroutes new writes and fails reads and garbage
|
||||
collection for the targets it does not declare.
|
||||
|
||||
### File Data Storage
|
||||
|
||||
|
||||
@ -91,6 +91,10 @@ reads, URL signing, deduplication and garbage collection. Objects without a
|
||||
stored target use the default target. See the configuration guide for the
|
||||
routing file format.
|
||||
|
||||
If an object references a target id that is not configured anymore, it cannot be
|
||||
located: reads and serving fail, and garbage collection refuses to delete the
|
||||
object, keeping the database row until the target is declared again.
|
||||
|
||||
## Sharing and deleting objects
|
||||
|
||||
To save storage space, duplicated objects wre shared. So, if for example
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user