mirror of
https://github.com/penpot/penpot.git
synced 2026-11-14 22:24:56 +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
309 lines
13 KiB
Clojure
309 lines
13 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 INC
|
|
|
|
(ns backend-tests.jobs-request-test
|
|
(:require
|
|
[app.common.generic-pool :as gpool]
|
|
[app.common.json :as json]
|
|
[app.common.time :as ct]
|
|
[app.config :as cf]
|
|
[app.db :as db]
|
|
[app.jobs :as jobs]
|
|
[app.main :as main]
|
|
[app.metrics :as mtx]
|
|
[app.metrics.definition :as-alias mdef]
|
|
[app.redis :as rds]
|
|
[app.worker :as wrk]
|
|
[backend-tests.helpers :as th]
|
|
[clojure.test :as t]
|
|
[cuerdas.core :as str]
|
|
[integrant.core :as ig])
|
|
(:import
|
|
io.lettuce.core.api.sync.RedisCommands
|
|
io.prometheus.client.Counter
|
|
io.prometheus.client.Counter$Child
|
|
java.lang.AutoCloseable))
|
|
|
|
(t/use-fixtures :once th/state-init)
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; HELPERS
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defn- make-metrics []
|
|
(ig/init-key :app.metrics/metrics
|
|
{:default (select-keys main/default-metrics
|
|
[:jobs-requests-total
|
|
:jobs-request-timing])}))
|
|
|
|
(defn- make-pool
|
|
([] (make-pool (get th/*system* :app.metrics/metrics)))
|
|
([metrics]
|
|
(-> {::jobs/request-pool {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics metrics}}
|
|
(ig/expand)
|
|
(ig/init)
|
|
(get ::jobs/request-pool))))
|
|
|
|
(defn- queue-key
|
|
[queue]
|
|
(wrk/queue-key (cf/get :tenant) queue))
|
|
|
|
(defn- make-cfg
|
|
[pool]
|
|
{::jobs/request-pool pool})
|
|
|
|
(defn- clear-queues
|
|
[]
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)
|
|
keys (vec (.keys ^RedisCommands cmd
|
|
(str jobs/reply-key-prefix ":" (cf/get :tenant) ":*")))]
|
|
(when (seq keys)
|
|
(rds/del conn keys))
|
|
(rds/del conn (queue-key :media)))
|
|
(finally
|
|
(rds/close conn)))))
|
|
|
|
(defn- test-fixture [next]
|
|
(th/database-reset
|
|
(fn []
|
|
(clear-queues)
|
|
(next))))
|
|
|
|
(t/use-fixtures :each test-fixture)
|
|
|
|
(defn- with-responder
|
|
"Start a responder on :media that answers the next request with the
|
|
provided response map (usually {:ok ...} or {:error ...}). Returns a
|
|
future that resolves to the decoded request payload [request-id,
|
|
reply-key, cmd, params]."
|
|
[pool response-fn & {:keys [delay-ms timeout-ms]}]
|
|
(let [timeout-ms (or timeout-ms 10000)
|
|
out (promise)]
|
|
(let [fut (future
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(when delay-ms
|
|
(Thread/sleep (long delay-ms)))
|
|
(let [key (queue-key :media)
|
|
[_ payload] (rds/blpop conn [key] (ct/duration {:millis timeout-ms}))]
|
|
(if (nil? payload)
|
|
(deliver out ::no-request)
|
|
(let [[request-id reply-key cmd params :as decoded] (json/decode payload)]
|
|
(deliver out decoded)
|
|
(jobs/reply {::rds/pool (::jobs/pool pool)
|
|
::mtx/metrics (::mtx/metrics pool)}
|
|
reply-key
|
|
(response-fn params)))))
|
|
(finally
|
|
(rds/close conn)))))]
|
|
{:future fut
|
|
:payload out})))
|
|
|
|
(defn- get-ttl
|
|
[key]
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)]
|
|
(.ttl ^RedisCommands cmd key))
|
|
(finally
|
|
(rds/close conn)))))
|
|
|
|
(defn- key-exists?
|
|
[key]
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)]
|
|
(pos? (.exists ^RedisCommands cmd (make-array String 1 key))))
|
|
(finally
|
|
(rds/close conn)))))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; TESTS
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(defn- counter-value [metrics id labels]
|
|
(let [collector (mtx/get-collector metrics id)
|
|
instance (::mdef/instance collector)
|
|
child (.labels ^Counter instance (into-array String labels))]
|
|
(.get ^Counter$Child child)))
|
|
|
|
(t/deftest request-pool-keeps-its-own-metrics
|
|
(let [metrics-a (make-metrics)
|
|
metrics-b (make-metrics)
|
|
pool-a (make-pool metrics-a)
|
|
pool-b (make-pool metrics-b)]
|
|
(t/is (identical? metrics-a (::mtx/metrics pool-a)))
|
|
(t/is (identical? metrics-b (::mtx/metrics pool-b)))))
|
|
|
|
(t/deftest request-metrics-use-the-final-outcome
|
|
(let [metrics (make-metrics)
|
|
pool (make-pool metrics)
|
|
cfg (make-cfg pool)]
|
|
(with-responder pool (fn [_params] {:ok {:value 1}}))
|
|
(t/is (= {:value 1}
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {}})))
|
|
(t/is (= 1.0 (counter-value metrics :jobs-requests-total ["replied"])))
|
|
(t/is (thrown-with-msg? Exception #"timeout waiting for the job reply"
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {}
|
|
::jobs/timeout (ct/duration {:millis 100})})))
|
|
(t/is (= 1.0 (counter-value metrics :jobs-requests-total ["timeout"])))
|
|
(t/is (= 0.0 (counter-value metrics :jobs-requests-total ["sent"])))))
|
|
|
|
(t/deftest request-returns-the-ok-reply
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
payload (with-responder pool (fn [_params] {:ok {:value 42}}))]
|
|
|
|
(let [result (jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}})]
|
|
(t/is (= {:value 42} result)))
|
|
|
|
(t/testing "the worker receives the expected payload shape"
|
|
(let [[request-id reply-key cmd params :as decoded] (deref (:payload payload) 10000 ::timeout)]
|
|
(t/is (string? request-id))
|
|
(t/is (string? reply-key))
|
|
(t/is (str/starts-with? reply-key (str jobs/reply-key-prefix ":" (cf/get :tenant) ":")))
|
|
(t/is (= "process" cmd))
|
|
(t/is (= {"x" 1} params)
|
|
(str "params should be plain JSON: " (pr-str decoded)))))))
|
|
|
|
(t/deftest request-propagates-the-error-reply
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
_ (with-responder pool
|
|
(fn [_params]
|
|
{:error {:code :processing-failed
|
|
:hint "something went wrong"}}))]
|
|
(t/is (thrown-with-msg? Exception #"something went wrong"
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}})))))
|
|
|
|
(t/deftest request-times-out-and-cleans-the-reply-key
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)]
|
|
(t/is (thrown-with-msg? Exception #"timeout waiting for the job reply"
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}
|
|
::jobs/timeout (ct/duration {:millis 500})})))
|
|
|
|
(t/testing "no orphan reply key is left behind"
|
|
;; there is no way to know the exact reply key (random request-id);
|
|
;; check no reply keys exist at all after the timeout
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)
|
|
keys (.keys ^RedisCommands cmd
|
|
(str jobs/reply-key-prefix ":" (cf/get :tenant) ":*"))]
|
|
(t/is (empty? (vec keys))))
|
|
(finally
|
|
(rds/close conn)))))))
|
|
|
|
(t/deftest request-pool-returns-connections-on-success-and-failure
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
default-timeout (-> (cf/get-jobs-request-timeout)
|
|
(ct/plus (ct/duration {:seconds 30})))]
|
|
|
|
(t/testing "successful request does not consume the connection"
|
|
(with-responder pool (fn [_params] {:ok {:a 1}}))
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}})
|
|
|
|
(t/testing "and the next request works with a pooled connection"
|
|
(with-responder pool (fn [_params] {:ok {:b 2}}))
|
|
(t/is (= {:b 2} (jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 2}})))
|
|
|
|
(t/testing "connection command timeout is restored after use"
|
|
(with-open [^AutoCloseable pooled (gpool/get (::jobs/pool pool))]
|
|
(let [conn @pooled]
|
|
(t/is (= (.toMillis ^java.time.Duration default-timeout)
|
|
(.toMillis ^java.time.Duration (rds/get-timeout conn)))))))))))
|
|
|
|
(t/deftest request-supports-larger-per-call-timeout
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
_ (with-responder pool (fn [_params] {:ok {:slow 1}}))]
|
|
|
|
(t/testing "an override bigger than the default works (command timeout raised per call)"
|
|
(t/is (= {:slow 1}
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}
|
|
::jobs/timeout (ct/duration {:seconds 130})}))))))
|
|
|
|
(t/deftest request-raises-connection-timeout-to-per-call-override
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
seen (atom nil)
|
|
orig rds/set-timeout
|
|
override (ct/duration {:seconds 999})
|
|
_ (with-responder pool (fn [_params] {:ok {:fast 1}}))]
|
|
;; an override above the pool default is honored, not capped
|
|
(with-redefs [rds/set-timeout (fn [conn timeout]
|
|
(reset! seen timeout)
|
|
(orig conn timeout))]
|
|
(t/is (= {:fast 1}
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}
|
|
::jobs/timeout override}))))
|
|
(t/is (= (.toMillis ^java.time.Duration (ct/plus override (ct/duration {:seconds 30})))
|
|
(.toMillis ^java.time.Duration @seen)))))
|
|
|
|
(t/deftest late-reply-is-cleaned-by-the-worker-expire
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
;; the responder replies 2s after the caller timeout (500ms)
|
|
responder (with-responder pool
|
|
(fn [_params] {:ok {:late 1}})
|
|
:delay-ms 2000)]
|
|
|
|
(t/is (thrown-with-msg? Exception #"timeout waiting for the job reply"
|
|
(jobs/request cfg {::jobs/queue :media
|
|
::jobs/cmd :process
|
|
::jobs/params {:x 1}
|
|
::jobs/timeout (ct/duration {:millis 500})})))
|
|
|
|
(t/testing "the late reply key exists (recreated by the worker) with a short TTL"
|
|
(deref (:future responder) 10000 ::timeout)
|
|
(let [conn (rds/connect {::rds/client (get th/*system* :app.redis/client)
|
|
::mtx/metrics (get th/*system* :app.metrics/metrics)})]
|
|
(try
|
|
(let [cmd (.-cmd conn)
|
|
keys (vec (.keys ^RedisCommands cmd
|
|
(str jobs/reply-key-prefix ":" (cf/get :tenant) ":*")))]
|
|
(t/is (= 1 (count keys)))
|
|
(let [ttl (.ttl ^RedisCommands cmd (first keys))]
|
|
(t/is (pos? ttl))
|
|
(t/is (<= ttl 60))))
|
|
(finally
|
|
(rds/close conn)))))))
|
|
|
|
(t/deftest request-creates-no-job-row
|
|
(let [pool (make-pool)
|
|
cfg (make-cfg pool)
|
|
_ (with-responder pool (fn [_params] {:ok {}}))]
|
|
(jobs/request cfg {::jobs/queue :media ::jobs/cmd :process ::jobs/params {}})
|
|
(t/is (zero? (:cnt (th/db-exec-one! ["SELECT count(*) AS cnt FROM job"]))))))
|