github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-11-10 03:50:33 +00:00
Code Issues Packages Projects Releases Wiki Activity
penpot/backend/test/backend_tests/jobs_gc_test.clj
Andrey Antukh d3ca9a0b83
✨ Add unified jobs substrate replacing legacy one-shot task system (#11542) 
* 🎉 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
2026-10-06 07:33:12 +02:00

362 lines
17 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 INC
(ns backend-tests.jobs-gc-test
(:require
[app.common.schema :as sm]
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.jobs :as jobs]
[app.jobs.gc :as gc]
[app.main :as main]
[app.metrics :as-alias mtx]
[app.metrics.definition :as-alias mdef]
[app.storage :as sto]
[backend-tests.helpers :as th]
[backend-tests.storage-test :refer [configure-storage-backend]]
[clojure.test :as t]
[integrant.core :as ig])
(:import
io.prometheus.client.Counter
io.prometheus.client.Counter$Child))
(t/use-fixtures :once th/state-init)
(t/use-fixtures :each (th/serial
th/database-reset
th/clean-storage))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; HELPERS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn- make-metrics []
(ig/init-key :app.metrics/metrics
{:default (select-keys main/default-metrics
[:jobs-gc-rows])}))
(defn- counter-value [metrics labels]
(let [collector (mtx/get-collector metrics :jobs-gc-rows)
instance (::mdef/instance collector)
child (.labels ^Counter instance (into-array String labels))]
(.get ^Counter$Child child)))
(defn- mk-storage-object
[]
(let [storage (-> (:app.storage/storage th/*system*)
(configure-storage-backend))]
(sto/put-object! storage {::sto/content (sto/content "content")
:content-type "text/plain"})))
(defn- mk-job
[{:keys [status profile-id resource-id expires-at modified-at]
:or {status "new"}}]
(let [id (uuid/next)]
(th/db-insert! :job {:id id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status status
:profile-id profile-id
:resource-id resource-id
:expires-at expires-at
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (or modified-at (ct/now))})
id))
(t/deftest gc-records-deleted-rows
(let [metrics (make-metrics)
cfg {::db/pool th/*pool*
::mtx/metrics metrics}]
(th/db-insert! :job {:id (uuid/next)
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "completed"
:expires-at (ct/in-past {:minutes 5})
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
(gc/execute-jobs-gc cfg {})
(t/is (= 1.0 (counter-value metrics ["expired" "deleted"])))))
(t/deftest job-rows-are-protected-against-unintended-deletes
(let [job-id (mk-job {:expires-at (ct/in-past {:minutes 5})})]
(t/testing "a stray DELETE is blocked by the deletion protection trigger"
(t/is (thrown-with-msg? Exception #"unable to proceed to delete row"
(th/db-exec! ["DELETE FROM job WHERE id = ?" job-id])))
(t/is (some? (th/db-get :job {:id job-id} :id :status))))
(t/testing "the jobs GC is the intended deleter and disables the guard"
(th/run-task! :jobs-gc {})
(t/is (nil? (th/db-get :job {:id job-id} :id :status))))))
(t/deftest gc-deletes-expired-jobs-and-touches-their-resources
(let [old-object (mk-storage-object)
live-object (mk-storage-object)
expired-id (uuid/next)
live-id (uuid/next)]
(th/db-insert! :job {:id expired-id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "completed"
:resource-id (:id old-object)
:expires-at (ct/in-past {:minutes 5})
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
(th/db-insert! :job {:id live-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 live-object)
:expires-at (ct/in-future {:minutes 5})
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
(th/run-task! :jobs-gc {})
(t/is (nil? (th/db-get :job {:id expired-id} :id :status)))
(t/is (some? (th/db-get :job {:id live-id} :id :status)))
(t/is (some? (:touched-at (th/db-get :storage-object {:id (:id old-object)}
:id :touched-at))))
(t/is (nil? (:touched-at (th/db-get :storage-object {:id (:id live-object)}
:id :touched-at)))))
;; a row without resource is deleted without touching anything
(let [expired-id (uuid/next)]
(th/db-insert! :job {:id expired-id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "new"
:expires-at (ct/in-past {:minutes 5})
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
(th/run-task! :jobs-gc {})
(t/is (nil? (th/db-get :job {:id expired-id} :id :status)))))
(t/deftest gc-params-accept-string-min-age
(t/testing "duration strings validate like durations and millis"
(t/is ((sm/validator gc/schema:jobs-gc-params) {:min-age "1h"}))
(t/is ((sm/validator gc/schema:jobs-gc-params) {:min-age 3600000}))
(t/is ((sm/validator gc/schema:jobs-gc-params) {:min-age (ct/duration {:hours 1})})))
(t/testing "string min-age runs like a duration"
(t/is (map? (th/run-task! :jobs-gc {:min-age "1h"})))))
(t/deftest submit-normalizes-durations-to-millis
(let [defs {:jobs-gc (ig/init-key ::gc/job-def {})}
cfg {::jobs/defs defs
::db/pool th/*pool*
::mtx/metrics (get th/*system* :app.metrics/metrics)}
job-id (jobs/submit cfg {::jobs/name :jobs-gc
::jobs/params {:min-age (ct/duration {:hours 1})}})]
(t/testing "a Duration object does not reach JSON encoding"
(t/is (= 3600000 (:min-age (:params (jobs/get-job cfg job-id))))))))
(t/deftest gc-retention-deletes-old-internal-terminal-rows
(let [profile (th/create-profile* 1 {})
;; old internal terminal row -> deleted
old-id (uuid/next)]
(th/db-insert! :job {:id old-id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "completed"
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/in-past {:days 10})})
;; recent internal terminal row: inside the retention window, survives
(let [recent-id (uuid/next)]
(th/db-insert! :job {:id recent-id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "completed"
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
;; non-terminal internal row: never swept by retention, survives
(let [running-id (uuid/next)]
(th/db-insert! :job {:id running-id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "running"
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/in-past {:days 10})})
;; old USER terminal row (profile set): retention does not sweep
;; the user ledger, survives
(let [user-id (uuid/next)]
(th/db-insert! :job {:id user-id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "completed"
:profile-id (:id profile)
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/in-past {:days 10})})
(th/run-task! :jobs-gc {})
(t/is (nil? (th/db-get :job {:id old-id} :id :status)))
(t/is (= 3 (count (th/db-query :job {:queue "default"}))))
(t/is (some? (th/db-get :job {:id recent-id} :id :status)))
(t/is (some? (th/db-get :job {:id running-id} :id :status)))
(t/is (some? (th/db-get :job {:id user-id} :id :status))))))))
(t/deftest gc-retention-deletes-old-aborted-rows
"Aborted jobs are terminal like failed/cancelled: old internal aborted
rows are swept by retention, recent ones survive."
(let [old-id (uuid/next)
recent-id (uuid/next)]
(doseq [[id modified-at] [[old-id (ct/in-past {:days 10})]
[recent-id (ct/now)]]]
(th/db-insert! :job {:id id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status "aborted"
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at modified-at}))
(th/run-task! :jobs-gc {})
(t/is (nil? (th/db-get :job {:id old-id} :id :status)))
(t/is (some? (th/db-get :job {:id recent-id} :id :status)))))
(t/deftest gc-retention-touches-resources-of-retained-rows
(let [object (mk-storage-object)
job-id (uuid/next)]
(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 "failed"
:resource-id (:id object)
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/in-past {:days 10})})
(th/run-task! :jobs-gc {})
(t/is (nil? (th/db-get :job {:id job-id} :id :status)))
(t/is (some? (:touched-at (th/db-get :storage-object {:id (:id object)}
:id :touched-at))))))
(t/deftest gc-deletes-expired-jobs-in-bounded-batches
(let [total 2500]
(th/db-exec! [(str "INSERT INTO job (name, tenant, queue, status, expires_at) "
"SELECT 'test-job', '" (cf/get :tenant) "', 'default', 'completed', "
"now() - interval '1 hour' "
"FROM generate_series(1, " total ")")])
;; spread a few resources across batches so every batch touches
(let [objects (repeatedly 3 mk-storage-object)]
(doseq [[object n] (map vector objects (range))]
(th/db-exec! ["UPDATE job SET resource_id = ?
WHERE id = (SELECT id FROM job OFFSET ? LIMIT 1)"
(:id object) (* n 1000)]))
(let [{:keys [deleted-expired touched-expired]} (th/run-task! :jobs-gc {})]
(t/is (= total deleted-expired))
(t/is (= (count objects) touched-expired))
(t/is (= 0 (count (th/db-query :job {:queue "default"}))))
(doseq [object objects]
(t/is (some? (:touched-at (th/db-get :storage-object {:id (:id object)}
:id :touched-at)))))))))
(t/deftest gc-commits-each-batch-separately
(let [total 2500
calls (atom 0)
orig @#'gc/touch-resources]
(th/db-exec! [(str "INSERT INTO job (name, tenant, queue, status, expires_at) "
"SELECT 'test-job', '" (cf/get :tenant) "', 'default', 'completed', "
"now() - interval '1 hour' "
"FROM generate_series(1, " total ")")])
;; blow up on the 3rd batch: earlier batches must stay committed
(alter-var-root #'gc/touch-resources
(constantly (fn [conn ids]
(when (= 3 (swap! calls inc))
(throw (ex-info "boom" {})))
(orig conn ids))))
(try
(t/is (thrown? Exception (th/run-task! :jobs-gc {})))
(t/is (= 500 (count (th/db-query :job {:queue "default"}))))
(finally
(alter-var-root #'gc/touch-resources (constantly orig))))
;; the next tick completes the sweep
(let [{:keys [deleted-expired]} (th/run-task! :jobs-gc {})]
(t/is (= 500 deleted-expired))
(t/is (zero? (count (th/db-query :job {:queue "default"})))))))
(t/deftest gc-expiration-skips-active-rows
(let [running-id (uuid/next)
retry-id (uuid/next)]
(doseq [[id status] [[running-id "running"] [retry-id "retry"]]]
(th/db-insert! :job {:id id
:name "test-job"
:tenant (cf/get :tenant)
:queue "default"
:params (db/json {})
:priority 100
:max-retries 3
:retry-num 0
:status status
:expires-at (ct/in-past {:minutes 5})
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)}))
(th/run-task! :jobs-gc {})
(t/testing "active rows survive expiration"
(t/is (some? (th/db-get :job {:id running-id} :id :status)))
(t/is (some? (th/db-get :job {:id retry-id} :id :status))))))
Reference in New Issue View Git Blame Copy Permalink