mirror of
https://github.com/penpot/penpot.git
synced 2026-11-05 09:30:38 +00:00
d3ca9a0b83
* 🎉 Move backend jobs to the unified job table substrate Replace the generic 'task' table with a unified 'job' table as the durable substrate for all backend jobs, and migrate every producer and consumer to it. The legacy 'task' table stops receiving writes but stays dormant (still cleaned by tasks-gc); upload-session chunking and the other post-branch upstream changes are preserved. New 0154 migration creates the 'job' table: dispatch and lifecycle columns (name, queue, priority, scheduled_at, retries, status, timestamps, props) plus optional user-facing ledger columns (profile_id, target, progress, error, result, resource_id, expires_at). No modified_at trigger: the application updates it on every write for lease and orphan detection. Partial indexes cover the dispatcher claim query, orphan scanning, and per-profile listing. New app.jobs public API: submit!/invoke!, heartbeats and progress reporting, a precompiled job-def registry (name, schema, handler, decoder, validator), an ephemeral request/response mode, and a management API for external workers. The dispatcher claims new/retry rows (FOR UPDATE SKIP LOCKED) with a plain JSON redis payload and marks lease-expired running rows as orphaned; the runner executes per-queue with the same retry conventions. New jobs GC (daily): deletes expired rows and retains internal terminal rows per :jobs-retention, touching referenced storage objects so storage-gc-touched reclaims them. Job resources live in a dedicated job-resource bucket with a reclaim branch in storage-gc-touched. All production call-sites (sendmail, delete-object, demo-purge, file-gc, offload, webhooks, quotes, nitrate bulk delete, snapshots, telemetry) and the ten cron tasks now submit through app.jobs; cron is a pure scheduler and the legacy worker registry, task-table inserts, and per-namespace handlers are removed. Includes migrations, dispatcher, runner, jobs, request/response, jobs GC, management API, and cron test suites. AI-assisted-by: glm-5.3-flash AI-assisted-by: muse-spark-1.3-contributor AI-assisted-by: kimi-k3 AI-assisted-by: mimo-v2.5 * ✨ Address review comments on jobs substrate Resolves the four open review threads on the jobs PR. Orphaned jobs are now marked aborted, a system-side terminal state with no retry: the dispatcher sweep stores the end event with outcome aborted in a single bulk insert in the same transaction, records the terminal outcome next to the orphaned counter, and reports an error log; orphans keep alerting through the log and the counter, with no user notification. The aborted state is terminal everywhere: the status CHECK, the sweep, the lifecycle writers, the jobs GC retention, the backlog gauges and the user status mapping, with tests covering each transition. The claim comment, the 2.20 doc version and the legacy task table notes are corrected, and objects-gc heartbeats per chunk. AI-assisted-by: muse-spark-1.3-contributor
414 lines
14 KiB
Clojure
414 lines
14 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.storage
|
|
"Objects storage abstraction layer."
|
|
(:refer-clojure :exclude [resolve])
|
|
(:require
|
|
[app.common.data :as d]
|
|
[app.common.data.macros :as dm]
|
|
[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.metrics :as mtx]
|
|
[app.storage.fs :as sfs]
|
|
[app.storage.impl :as impl]
|
|
[app.storage.s3 :as ss3]
|
|
[app.storage.schema :as stsch]
|
|
[cuerdas.core :as str]
|
|
[datoteka.fs :as fs]
|
|
[integrant.core :as ig])
|
|
(:import
|
|
java.io.InputStream))
|
|
|
|
(defn get-legacy-backend
|
|
[]
|
|
(when-let [name (cf/get :assets-storage-backend)]
|
|
(l/wrn :hint "using deprecated configuration, please read 2.11 release notes"
|
|
:href "https://github.com/penpot/penpot/releases/tag/2.11.0")
|
|
(case name
|
|
:assets-fs :fs
|
|
:assets-s3 :s3
|
|
nil)))
|
|
|
|
(def default-bucket
|
|
stsch/default-bucket)
|
|
|
|
(def tempfile-bucket
|
|
"Bucket name for temporary file uploads (10-minute expiry)."
|
|
stsch/tempfile-bucket)
|
|
|
|
(def upload-session-bucket
|
|
"Bucket name for chunked-upload chunks."
|
|
stsch/upload-session-bucket)
|
|
|
|
(def job-resource-bucket
|
|
"Bucket for storage objects referenced by job.resource_id (the
|
|
unified jobs substrate): transient artifacts owned by their job row,
|
|
reclaimed by storage-gc-touched (via the jobs GC touch) once no job
|
|
row references them anymore."
|
|
stsch/job-resource-bucket)
|
|
|
|
(def valid-buckets
|
|
stsch/metadata-buckets)
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; Storage Module State
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(def ^:private schema:backends
|
|
[:map-of :keyword
|
|
[:maybe
|
|
[:or ::ss3/backend ::sfs/backend]]])
|
|
|
|
(def ^:private valid-backends?
|
|
(sm/validator schema:backends))
|
|
|
|
(def ^:private schema:storage
|
|
[:map {:title "storage"}
|
|
[::backends schema:backends]
|
|
[::backend [:enum :s3 :fs]]
|
|
[::mtx/metrics ::mtx/metrics]
|
|
::db/pool])
|
|
|
|
(def valid-storage?
|
|
(sm/validator schema:storage))
|
|
|
|
(sm/register! ::storage schema:storage)
|
|
|
|
(defmethod ig/assert-key ::storage
|
|
[_ params]
|
|
(assert (db/pool? (::db/pool params)) "expected valid database pool")
|
|
(assert (valid-backends? (::backends params)) "expected valid backends map"))
|
|
|
|
(defmethod ig/init-key ::storage
|
|
[_ {:keys [::backends ::db/pool] :as cfg}]
|
|
(let [backend (or (get-legacy-backend)
|
|
(cf/get :objects-storage-backend)
|
|
:fs)
|
|
backends (d/without-nils backends)]
|
|
|
|
(l/dbg :hint "initialize"
|
|
:default (d/name backend)
|
|
:available (str/join "," (map d/name (keys backends))))
|
|
|
|
(-> (d/without-nils cfg)
|
|
(assoc ::backends backends)
|
|
(assoc ::backend backend)
|
|
(assoc ::db/pool pool))))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; Database Objects
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defn get-metadata
|
|
[params]
|
|
(reduce-kv (fn [res k _]
|
|
(if (qualified-keyword? k)
|
|
(dissoc res k)
|
|
res))
|
|
params
|
|
params))
|
|
|
|
;; Matches both metadata encodings while Transit rows still exist. The
|
|
;; UNION ALL keeps each branch indexable: migration 0068 covers the legacy
|
|
;; "~:" keys and 0156 covers the plain JSON keys. Phase 3 drops the legacy
|
|
;; branch once no "~:" rows remain.
|
|
(def ^:private sql:get-database-object-by-hash
|
|
"(
|
|
select * from storage_object
|
|
where (metadata->>'~:hash') = ?
|
|
and (metadata->>'~:bucket') = ?
|
|
and backend = ?
|
|
and deleted_at is null
|
|
and status = 'valid'
|
|
limit 1
|
|
) union all (
|
|
select * from storage_object
|
|
where (metadata->>'hash') = ?
|
|
and (metadata->>'bucket') = ?
|
|
and backend = ?
|
|
and deleted_at is null
|
|
and status = 'valid'
|
|
limit 1
|
|
) limit 1")
|
|
|
|
;; NOTE: metadata is left encoded; row->storage-object is responsible for
|
|
;; decoding it.
|
|
(defn- get-database-object-by-hash
|
|
[connectable backend bucket hash]
|
|
(let [backend (name backend)]
|
|
(db/exec-one! connectable [sql:get-database-object-by-hash
|
|
hash bucket backend
|
|
hash bucket backend])))
|
|
|
|
(defn- promote-object!
|
|
[storage object]
|
|
(let [ds (db/get-connectable storage)
|
|
res (-> (db/update! ds :storage-object
|
|
{:status "valid"}
|
|
{:id (:id object)}
|
|
{::db/return-keys false})
|
|
(db/get-update-count))]
|
|
(when-not (pos? res)
|
|
;; The pending row disappeared while the blob was being written
|
|
;; (e.g. reclaimed by :storage-pending-gc); make it observable.
|
|
(l/wrn :hint "unable to promote storage object, pending row not found"
|
|
:id (str (:id object))))
|
|
res))
|
|
|
|
(defn row->storage-object [res]
|
|
(let [mdata (or (some-> (:metadata res) (stsch/decode-metadata)) {})]
|
|
(impl/storage-object
|
|
(:id res)
|
|
(:size res)
|
|
(:created-at res)
|
|
(:deleted-at res)
|
|
(:touched-at res)
|
|
(keyword (:backend res))
|
|
mdata)))
|
|
|
|
(def ^:private sql:get-storage-object
|
|
"SELECT *
|
|
FROM storage_object
|
|
WHERE id = ?
|
|
AND (deleted_at IS NULL)
|
|
AND status = 'valid'")
|
|
|
|
(defn- get-database-object
|
|
[conn id]
|
|
(some-> (db/exec-one! conn [sql:get-storage-object id])
|
|
(row->storage-object)))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; API
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defn object->relative-path
|
|
[{:keys [id] :as obj}]
|
|
(impl/id->path id))
|
|
|
|
(defn file-url->path
|
|
[url]
|
|
(when url
|
|
(fs/path (java.net.URI. (str url)))))
|
|
|
|
(dm/export impl/content)
|
|
(dm/export impl/wrap-with-hash)
|
|
(dm/export impl/object?)
|
|
|
|
(defn- emit-op!
|
|
"Record a logical storage operation. Recording never fails: metrics
|
|
must not change storage behavior."
|
|
([storage op bucket]
|
|
(emit-op! storage op bucket nil))
|
|
([storage op bucket object]
|
|
(mtx/run! (::mtx/metrics storage)
|
|
:id :storage-operations :inc 1
|
|
:labels [op
|
|
(mtx/label bucket "unknown")
|
|
(mtx/label (or (some-> object :backend) (::backend storage))
|
|
"unknown")])))
|
|
|
|
(defn- emit-dedup!
|
|
"Record a deduplication outcome. Recording never fails."
|
|
[storage result bucket]
|
|
(mtx/run! (::mtx/metrics storage)
|
|
:id :storage-dedup :inc 1
|
|
:labels [(name result) (mtx/label bucket "unknown")]))
|
|
|
|
(defn get-object
|
|
[storage id]
|
|
(assert (valid-storage? storage))
|
|
(let [ds (db/get-connectable storage)]
|
|
(get-database-object ds id)))
|
|
|
|
(defn put-object!
|
|
"Creates a new object with the provided content."
|
|
[{:keys [::backend ::db/pool] :as storage}
|
|
{:keys [::content ::expired-at ::touched-at ::touch] :as params}]
|
|
(assert (valid-storage? storage))
|
|
(assert (impl/content? content) "expected an instance of content")
|
|
|
|
(let [id (or (::id params) (uuid/random))
|
|
mdata (cond-> (get-metadata params)
|
|
(satisfies? impl/IContentHash content)
|
|
(assoc :hash (impl/get-hash content)))
|
|
|
|
touched-at (if touch
|
|
(or touched-at (ct/now))
|
|
touched-at)
|
|
|
|
backend' (impl/resolve-backend storage backend)
|
|
|
|
bucket (:bucket mdata)
|
|
dedupable? (and (::deduplicate? params)
|
|
(:hash mdata)
|
|
(some? bucket)
|
|
(not= tempfile-bucket bucket)
|
|
(not= upload-session-bucket bucket))
|
|
|
|
hit (when dedupable?
|
|
(get-database-object-by-hash pool backend bucket (:hash mdata)))]
|
|
|
|
;; NOTE: for now we don't reuse the deleted objects, but in futute
|
|
;; we can consider reusing deleted objects if we found a duplicated
|
|
;; one and is marked for deletion but still not deleted.
|
|
|
|
;; PHASE 1: deduplication lookup (see `dedupable?` and `hit` above).
|
|
(if-some [hit hit]
|
|
|
|
;; PHASE 2: an existing reference is found: reuse or repair it.
|
|
;; The `exists` op is emitted only after a successful probe so every
|
|
;; count stays paired with its `hit`/`repair` outcome.
|
|
(if (impl/exists-object? backend' hit)
|
|
|
|
;; PHASE 2a: healthy reference. Optionally refresh touched_at
|
|
;; and reuse the object as it is.
|
|
(do
|
|
(emit-op! storage "exists" bucket)
|
|
(emit-dedup! storage :hit bucket)
|
|
(when touch
|
|
(db/update! pool :storage-object
|
|
{:touched-at touched-at}
|
|
{:id (:id hit)}
|
|
{::db/return-keys false}))
|
|
(row->storage-object (cond-> hit touch (assoc :touched-at touched-at))))
|
|
|
|
;; PHASE 2b: the referenced blob is missing (a stale/broken row).
|
|
;; Repair the reference in place: rewrite the incoming content
|
|
;; under the same id, restoring the blob for all existing
|
|
;; references to it. If the write fails, the exception propagates
|
|
;; and the row stays live and valid, so a later matching upload
|
|
;; retries the heal.
|
|
(let [object (row->storage-object hit)]
|
|
(l/wrn :hint "blob not found on reusing storage object"
|
|
:id (:id object)
|
|
:backend (name backend))
|
|
(impl/put-object backend' object content)
|
|
(emit-op! storage "exists" bucket)
|
|
(emit-op! storage "repair" bucket)
|
|
(emit-dedup! storage :repair bucket)
|
|
(promote-object! storage object)
|
|
object))
|
|
|
|
;; PHASE 3: no dedup hit: create a fresh object. The row is
|
|
;; inserted in 'pending' state so it is not visible to the normal
|
|
;; lifecycle (dedup, gc, reads) until the blob has been written
|
|
;; and the object promoted to 'valid'.
|
|
(let [row (db/insert! pool :storage-object
|
|
{:id id
|
|
:size (impl/get-size content)
|
|
:backend (name backend)
|
|
:metadata (stsch/encode-metadata mdata)
|
|
:deleted-at expired-at
|
|
:touched-at touched-at
|
|
:status "pending"})
|
|
object (row->storage-object row)]
|
|
(impl/put-object backend' object content)
|
|
(emit-op! storage "put" bucket)
|
|
(emit-dedup! storage (if dedupable? :miss :skip) bucket)
|
|
(promote-object! storage object)
|
|
object))))
|
|
|
|
(defn touch-object!
|
|
"Mark object as touched. Takes the object id (UUID). The metric labels
|
|
come from the updated row itself (RETURNING); no row, no metric."
|
|
[storage id]
|
|
(assert (valid-storage? storage))
|
|
(let [ds (db/get-connectable storage)
|
|
res (db/update! ds :storage-object
|
|
{:touched-at (ct/now)}
|
|
{:id id}
|
|
{::db/return-keys [:id :backend :metadata]})]
|
|
(when-some [object (some-> res row->storage-object)]
|
|
(emit-op! storage "touch" (-> object meta :bucket) object))
|
|
(some? res)))
|
|
|
|
(defn get-object-data
|
|
"Return an input stream instance of the object content."
|
|
^InputStream
|
|
[storage object]
|
|
(assert (valid-storage? storage))
|
|
(when (or (nil? (:expired-at object))
|
|
(ct/is-after? (:expired-at object) (ct/now)))
|
|
(let [result (-> (impl/resolve-backend storage (:backend object))
|
|
(impl/get-object-data object))]
|
|
(emit-op! storage "get-data" (-> object meta :bucket) object)
|
|
result)))
|
|
|
|
(defn get-object-bytes
|
|
"Returns a byte array of object content."
|
|
[storage object]
|
|
(assert (valid-storage? storage))
|
|
(when (or (nil? (:expired-at object))
|
|
(ct/is-after? (:expired-at object) (ct/now)))
|
|
(let [result (-> (impl/resolve-backend storage (:backend object))
|
|
(impl/get-object-bytes object))]
|
|
(emit-op! storage "get-bytes" (-> object meta :bucket) object)
|
|
result)))
|
|
|
|
(defn get-object-url
|
|
([storage object]
|
|
(get-object-url storage object nil))
|
|
([storage object options]
|
|
(assert (valid-storage? storage))
|
|
(when (or (nil? (:expired-at object))
|
|
(ct/is-after? (:expired-at object) (ct/now)))
|
|
(-> (impl/resolve-backend storage (:backend object))
|
|
(impl/get-object-url object options)))))
|
|
|
|
(defn get-object-path
|
|
"Get the Path to the object. Only works with `:fs` type of
|
|
storages."
|
|
[storage object]
|
|
(assert (valid-storage? storage))
|
|
(let [backend (impl/resolve-backend storage (:backend object))]
|
|
(when (and (= :fs (::type backend))
|
|
(or (nil? (:expired-at object))
|
|
(ct/is-after? (:expired-at object) (ct/now))))
|
|
(-> (impl/get-object-url backend object nil) file-url->path))))
|
|
|
|
(defn del-object!
|
|
"Mark the object as deleted (soft delete: the backend content is
|
|
removed by the GC). Takes the object id (UUID). Only a live row
|
|
(deleted_at IS NULL) is deleted, so a repeated call is a no-op that
|
|
returns false. The metric labels come from the updated row itself
|
|
(RETURNING); no row, no metric."
|
|
[storage id]
|
|
(assert (valid-storage? storage))
|
|
(let [ds (db/get-connectable storage)
|
|
res (db/update! ds :storage-object
|
|
{:deleted-at (ct/now)}
|
|
["id = ? AND deleted_at IS NULL" id]
|
|
{::db/return-keys [:id :backend :metadata]})]
|
|
(when-some [object (some-> res row->storage-object)]
|
|
(emit-op! storage "del" (-> object meta :bucket) object))
|
|
(some? res)))
|
|
|
|
(dm/export impl/calculate-hash)
|
|
(dm/export impl/get-hash)
|
|
(dm/export impl/get-size)
|
|
|
|
(defn configure
|
|
[storage connection]
|
|
(assert (db/connection? connection))
|
|
(assert (valid-storage? storage))
|
|
(assoc storage ::db/conn connection))
|
|
|
|
(defn resolve
|
|
"Resolves the storage instance with preconfigured backend. You can
|
|
specify to reuse the database connection from provided
|
|
cfg/system (default false)."
|
|
[cfg & {:as opts}]
|
|
(let [storage (::storage cfg)]
|
|
(if (::db/reuse-conn opts false)
|
|
(configure storage (db/get-connection cfg))
|
|
storage)))
|