Merge pull request #7682 from penpot/niwinz-staging-worker-runner-exceptions

🐛 Fix incorrect status return on worker runner
This commit is contained in:
Alejandro Alonso 2025-11-05 07:44:28 +01:00 committed by GitHub
commit 7c529eedd4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 31 additions and 21 deletions

View File

@ -319,5 +319,9 @@
([key default] ([key default]
(c/get config key default))) (c/get config key default)))
(defn logging-context
[]
{:version/backend (:full version)})
;; Set value for all new threads bindings. ;; Set value for all new threads bindings.
(alter-var-root #'*assert* (constantly (contains? flags :backend-asserts))) (alter-var-root #'*assert* (constantly (contains? flags :backend-asserts)))

View File

@ -25,15 +25,14 @@
(let [claims (-> {} (let [claims (-> {}
(into (::session/token-claims request)) (into (::session/token-claims request))
(into (::actoken/token-claims request)))] (into (::actoken/token-claims request)))]
{:request/path (:path request) (-> (cf/logging-context)
:request/method (:method request) (assoc :request/path (:path request))
:request/params (:params request) (assoc :request/method (:method request))
:request/user-agent (yreq/get-header request "user-agent") (assoc :request/params (:params request))
:request/ip-addr (inet/parse-request request) (assoc :request/user-agent (yreq/get-header request "user-agent"))
:request/profile-id (:uid claims) (assoc :request/ip-addr (inet/parse-request request))
:version/frontend (or (yreq/get-header request "x-frontend-version") "unknown") (assoc :request/profile-id (:uid claims))
:version/backend (:full cf/version)})) (assoc :version/frontend (or (yreq/get-header request "x-frontend-version") "unknown")))))
(defmulti handle-error (defmulti handle-error
(fn [cause _ _] (fn [cause _ _]

View File

@ -13,6 +13,7 @@
[app.common.schema :as sm] [app.common.schema :as sm]
[app.common.time :as ct] [app.common.time :as ct]
[app.common.transit :as t] [app.common.transit :as t]
[app.config :as cf]
[app.db :as db] [app.db :as db]
[app.metrics :as mtx] [app.metrics :as mtx]
[app.redis :as rds] [app.redis :as rds]
@ -60,7 +61,8 @@
(defn get-error-context (defn get-error-context
[_ item] [_ item]
{:params item}) (-> (cf/logging-context)
(assoc :params item)))
(defn- get-task (defn- get-task
[{:keys [::db/pool]} task-id] [{:keys [::db/pool]} task-id]
@ -131,6 +133,11 @@
[{:keys [::id ::timeout] :as cfg} task-id scheduled-at] [{:keys [::id ::timeout] :as cfg} task-id scheduled-at]
(loop [task (get-task cfg task-id)] (loop [task (get-task cfg task-id)]
(cond (cond
(nil? task)
(l/wrn :hint "no task found on the database"
:runner-id id
:task-id task-id)
(ex/exception? task) (ex/exception? task)
(if (or (db/connection-error? task) (if (or (db/connection-error? task)
(db/serialization-error? task)) (db/serialization-error? task))
@ -153,11 +160,6 @@
:task-id task-id :task-id task-id
:runner-id id) :runner-id id)
(nil? task)
(l/wrn :hint "no task found on the database"
:runner-id id
:task-id task-id)
:else :else
(let [result (run-task cfg task)] (let [result (run-task cfg task)]
(with-meta result (with-meta result
@ -213,6 +215,7 @@
:payload payload))) :payload payload)))
(catch Throwable cause (catch Throwable cause
(l/err :hint "unable to decode payload" (l/err :hint "unable to decode payload"
::l/context (cf/logging-context)
:payload payload :payload payload
:length (alength ^String/1 payload) :length (alength ^String/1 payload)
:cause cause)))) :cause cause))))
@ -224,11 +227,11 @@
"failed" (handle-task-failure result) "failed" (handle-task-failure result)
"completed" (handle-task-completion result) "completed" (handle-task-completion result)
(throw (IllegalArgumentException. (throw (IllegalArgumentException.
(str "invalid status received: " status)))))) (str "invalid status received: '" status "'"))))))
(run-task-loop [[task-id scheduled-at]] (run-task-loop [[task-id scheduled-at]]
(loop [result (run-task! cfg task-id scheduled-at)] (loop [result (run-task! cfg task-id scheduled-at)]
(when-let [cause (process-result result)] (when-let [cause (some-> result process-result)]
(if (or (db/connection-error? cause) (if (or (db/connection-error? cause)
(db/serialization-error? cause)) (db/serialization-error? cause))
(do (do
@ -236,9 +239,9 @@
:cause cause) :cause cause)
(px/sleep timeout) (px/sleep timeout)
(recur result)) (recur result))
(do (l/err :hint "unhandled exception on processing task result"
(l/err :hint "unhandled exception on processing task result" ::l/context (cf/logging-context)
:cause cause))))))] :cause cause)))))]
(try (try
(let [key (str/ffmt "penpot.worker.queue:%" queue) (let [key (str/ffmt "penpot.worker.queue:%" queue)
@ -254,11 +257,14 @@
(if (rds/timeout-exception? cause) (if (rds/timeout-exception? cause)
(do (do
(l/err :hint "redis pop operation timeout, consider increasing redis timeout (will retry in some instants)" (l/err :hint "redis pop operation timeout, consider increasing redis timeout (will retry in some instants)"
::l/context (cf/logging-context)
:timeout timeout :timeout timeout
:cause cause) :cause cause)
(px/sleep timeout)) (px/sleep timeout))
(l/err :hint "unhandled exception" :cause cause)))))) (l/err :hint "unhandled exception"
::l/context (cf/logging-context)
:cause cause))))))
(defn- start-thread! (defn- start-thread!
[{:keys [::id ::queue ::wrk/tenant] :as cfg}] [{:keys [::id ::queue ::wrk/tenant] :as cfg}]
@ -284,6 +290,7 @@
:queue queue)) :queue queue))
(catch Throwable cause (catch Throwable cause
(l/err :hint "unexpected exception" (l/err :hint "unexpected exception"
::l/context (cf/logging-context)
:id id :id id
:queue queue :queue queue
:cause cause)) :cause cause))