mirror of
https://github.com/penpot/penpot.git
synced 2026-11-12 13:00:37 +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
244 lines
9.2 KiB
Clojure
244 lines
9.2 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.loggers.database
|
|
"A specific logger impl that persists errors on the database."
|
|
(:require
|
|
[app.common.data :as d]
|
|
[app.common.exceptions :as ex]
|
|
[app.common.logging :as l]
|
|
[app.common.pprint :as pp]
|
|
[app.common.schema :as sm]
|
|
[app.common.uri :as u]
|
|
[app.config :as cf]
|
|
[app.db :as db]
|
|
[app.loggers.audit :as audit]
|
|
[app.rpc.rlimit :as-alias rlimit]
|
|
[clojure.spec.alpha :as s]
|
|
[integrant.core :as ig]
|
|
[promesa.exec :as px]
|
|
[promesa.exec.csp :as sp]))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; Error Listener
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(declare handle-event)
|
|
|
|
(defonce enabled (atom true))
|
|
|
|
(defn- persist-on-database!
|
|
[pool id source report]
|
|
(when-not (db/read-only? pool)
|
|
(db/insert! pool :server-error-report
|
|
{:id id
|
|
:source source
|
|
:version source ;; backward compatibility with old code that reads version column
|
|
:content (db/tjson report)})))
|
|
|
|
(defn- concurrent-exception?
|
|
[cause]
|
|
(or (instance? java.util.concurrent.CompletionException cause)
|
|
(instance? java.util.concurrent.ExecutionException cause)))
|
|
|
|
(defn- log-record->report
|
|
[{:keys [::l/context ::l/message ::l/props ::l/logger ::l/level ::l/cause] :as record}]
|
|
(assert (l/valid-record? record) "expectd valid log record")
|
|
(let [data (if (concurrent-exception? cause)
|
|
(ex-data (ex-cause cause))
|
|
(ex-data cause))
|
|
|
|
ctx (-> context
|
|
(assoc :backend/tenant (cf/get :tenant))
|
|
(assoc :backend/host (cf/get :host))
|
|
(assoc :backend/public-uri (str (cf/get :public-uri)))
|
|
(assoc :backend/version (:full cf/version))
|
|
(assoc :logger/name logger)
|
|
(assoc :logger/level level)
|
|
(dissoc :request/params :value :params :data))
|
|
|
|
href (if-let [path (:request/path context)]
|
|
(str (u/join (cf/get :public-uri) path))
|
|
(str (cf/get :public-uri)))]
|
|
|
|
(merge
|
|
{:context (-> (into (sorted-map) ctx)
|
|
(pp/pprint-str :length 50))
|
|
:props (pp/pprint-str props :length 50)
|
|
:hint (or (when-let [message (ex-message cause)]
|
|
(if-let [props-hint (:hint props)]
|
|
(str props-hint ": " message)
|
|
message))
|
|
@message)
|
|
:trace (or (::trace record)
|
|
(some-> cause (ex/format-throwable :data? true :explain? false :header? false :summary? false)))
|
|
:tenant (cf/get :tenant)
|
|
:version (:full cf/version)
|
|
:profile-id (some-> (:request/profile-id context) str)
|
|
:href href}
|
|
|
|
(when-let [params (or (:request/params context) (:params context))]
|
|
{:params (pp/pprint-str params :length 20 :level 20)})
|
|
|
|
(when-let [value (:value context)]
|
|
{:value (pp/pprint-str value :length 30 :level 13)})
|
|
|
|
(when-let [data (some-> data (dissoc ::s/problems ::s/value ::s/spec ::sm/explain :hint))]
|
|
{:data (pp/pprint-str data :length 30 :level 13)})
|
|
|
|
(when-let [explain (ex/explain data :length 30 :level 13)]
|
|
{:explain explain}))))
|
|
|
|
(defn- handle-log-record
|
|
"Convert the log record into a report object and persist it on the database"
|
|
[{:keys [::db/pool]} {:keys [::l/id] :as record}]
|
|
(try
|
|
(let [report (-> record log-record->report d/without-nils)]
|
|
(l/dbg :hint "registering error on database"
|
|
:id (str id)
|
|
:src "logging"
|
|
:uri (cf/get-public-uri "dbg/error/" (str id)))
|
|
(persist-on-database! pool id 3 report))
|
|
(catch Throwable cause
|
|
(l/warn :hint "unexpected exception on database error logger" :cause cause))))
|
|
|
|
(defn- audit-event->report
|
|
[{:keys [context props ip-addr profile-id] :as record}]
|
|
(let [context
|
|
(reduce-kv (fn [context k v]
|
|
(let [k' (keyword "frontend" (name k))]
|
|
(-> context
|
|
(dissoc k)
|
|
(assoc k' v))))
|
|
context
|
|
context)
|
|
|
|
context
|
|
(-> context
|
|
(assoc :backend/tenant (cf/get :tenant))
|
|
(assoc :backend/host (cf/get :host))
|
|
(assoc :backend/public-uri (str (cf/get :public-uri)))
|
|
(assoc :backend/version (:full cf/version))
|
|
(assoc :frontend/ip-addr ip-addr))]
|
|
|
|
{:context (-> (into (sorted-map) context)
|
|
(pp/pprint-str :length 50))
|
|
:kind (:name record)
|
|
:profile-id (some-> profile-id str)
|
|
:href (get props :href)
|
|
:hint (get props :hint)
|
|
:trace (get props :report)
|
|
:tenant (cf/get :tenant)
|
|
:version (:full cf/version)}))
|
|
|
|
(defn- handle-audit-event
|
|
"Convert the log record into a report object and persist it on the database"
|
|
[{:keys [::db/pool]} {:keys [id] :as event}]
|
|
(try
|
|
(let [report (-> event audit-event->report d/without-nils)]
|
|
(l/dbg :hint "registering error on database"
|
|
:id (str id)
|
|
:src "audit-log"
|
|
:uri (cf/get-public-uri "dbg/error/" (str id)))
|
|
(persist-on-database! pool id 4 report))
|
|
(catch Throwable cause
|
|
(l/warn :hint "unexpected exception on database error logger" :cause cause))))
|
|
|
|
(defn- rlimit-event->report
|
|
[event]
|
|
(let [context
|
|
(-> {}
|
|
(assoc :rlimit/uid (::rlimit/uid event))
|
|
(assoc :rlimit/method (::rlimit/method event))
|
|
(assoc :backend/tenant (cf/get :tenant))
|
|
(assoc :backend/host (cf/get :host))
|
|
(assoc :backend/public-uri (str (cf/get :public-uri)))
|
|
(assoc :backend/version (:full cf/version)))
|
|
|
|
result
|
|
(->> (::rlimit/results event)
|
|
(mapv (fn [result]
|
|
(-> (into (sorted-map) result)
|
|
(dissoc ::rlimit/method)))))]
|
|
|
|
{:hint (str "Rate Limit Rejection: " (::rlimit/method event) " for " (::rlimit/uid event))
|
|
:context (-> (into (sorted-map) context)
|
|
(pp/pprint-str :length 50))
|
|
:value (pp/pprint-str result :length 50)
|
|
:tenant (cf/get :tenant)
|
|
:version (:full cf/version)
|
|
:href (cf/get-public-uri)}))
|
|
|
|
(defn- handle-rlimit-event
|
|
"Convert the log record into a report object and persist it on the database"
|
|
[{:keys [::db/pool]} {:keys [::rlimit/id] :as event}]
|
|
(try
|
|
(let [report (-> event rlimit-event->report d/without-nils)]
|
|
(l/dbg :hint "registering rate limit rejection"
|
|
:id (str id)
|
|
:src "rlimit"
|
|
:uri (cf/get-public-uri "dbg/error/" (str id)))
|
|
(persist-on-database! pool id 5 report))
|
|
(catch Throwable cause
|
|
(l/warn :hint "unexpected exception on database error logger" :cause cause))))
|
|
|
|
(defmethod ig/assert-key ::reporter
|
|
[_ params]
|
|
(assert (db/pool? (::db/pool params)) "expect valid database pool"))
|
|
|
|
(defmethod ig/init-key ::reporter
|
|
[_ cfg]
|
|
(when (contains? cf/flags :error-reporting)
|
|
(let [input (sp/chan :buf (sp/sliding-buffer 256))
|
|
thread (px/thread
|
|
{:name "penpot/reporter/database"}
|
|
(l/info :hint "initializing database error persistence")
|
|
(try
|
|
(loop []
|
|
(when-let [item (sp/take! input)]
|
|
(when @enabled
|
|
(cond
|
|
(::l/id item)
|
|
(handle-log-record cfg item)
|
|
|
|
(::rlimit/id item)
|
|
(handle-rlimit-event cfg item)
|
|
|
|
(-> item meta ::audit/event)
|
|
(handle-audit-event cfg item)
|
|
|
|
:else
|
|
(l/warn :hint "received unexpected item" :item item)))
|
|
|
|
(recur)))
|
|
|
|
(catch InterruptedException _
|
|
(l/debug :hint "reporter interrupted"))
|
|
(catch Throwable cause
|
|
(l/error :hint "unexpected error" :cause cause))
|
|
(finally
|
|
(l/info :hint "reporter terminated"))))]
|
|
|
|
(add-watch l/log-record ::reporter
|
|
(fn [_ _ _ record]
|
|
(when (= :error (::l/level record))
|
|
(sp/put! input record))))
|
|
|
|
{::input input
|
|
::thread thread})))
|
|
|
|
(defmethod ig/halt-key! ::reporter
|
|
[_ {:keys [::input ::thread]}]
|
|
(remove-watch l/log-record ::reporter)
|
|
(some-> input sp/close!)
|
|
(some-> thread px/interrupt!))
|
|
|
|
(defn emit
|
|
"Emit an event/report into the database reporter"
|
|
[cfg event]
|
|
(when-let [{:keys [::input]} (get cfg ::reporter)]
|
|
(sp/put! input event)))
|