diff --git a/exporter/src/app/consumer/exports.cljs b/exporter/src/app/consumer/exports.cljs index 27280ecf59..6c34f59a25 100644 --- a/exporter/src/app/consumer/exports.cljs +++ b/exporter/src/app/consumer/exports.cljs @@ -19,6 +19,13 @@ call that stores the artifact and closes the session in the same step. + Cancellation arrives through the beats: a `skip` answer from + `report-job-progress` says the job can no longer hear the worker + (cancelled, aborted, settled). Between units of work the runner + raises, the same `:job-cancelled` the legacy surfaces raised; inside + a render mid-Skia, the watchdog fires the cancel port the wasm pool + already serves (terminate of the leased thread). + The run never rejects on an export failure: whatever goes wrong reaches the backend as `fail-job` (with the session id, so the render session does not outlive the settle). Only bugs of this process @@ -29,6 +36,7 @@ [app.consumer.api :as api] [app.handlers.export-shapes :as shapes] [app.handlers.resources :as rsc] + [app.jobs :as jobs] [app.jobs.utils :as job.utils] [app.renderer :as rd] [app.util.shell :as sh] @@ -36,27 +44,71 @@ [promesa.core :as p])) (def ^:private report-throttle-ms 250) +(def ^:private watchdog-interval-ms 1000) -;; ---- PROGRESS (the milestones of the contract) - -(defn- report! - [job-id stage counters] - (api/report-job-progress job-id {:stage stage :counters counters})) +;; ---- PROGRESS AND CANCEL (the milestones of the contract) (defn- start-beats! - "One progress reporter per run. The stage changes beat by force; the - per-object beats are throttled the way the legacy export was, since - hundreds of objects would otherwise be hundreds of malformed calls - for information nobody reads at that resolution." - [job-id] - (let [last-beat (atom 0)] - (fn [stage counters & {:keys [force?]}] - (if force? - (report! job-id stage counters) - (let [now (js/Date.now)] - (when (>= (- now @last-beat) report-throttle-ms) - (reset! last-beat now) - (report! job-id stage counters))))))) + "The reporter of one run's beats. + + The beat reports the stage and the counters and asks the backend to + go on; a `skip` answer means nobody is listening anymore — the job + was cancelled or settled elsewhere — and the export must stop: the + reporter fires `on-cancel` once (what terminates the render worker + the run holds mid-Skia; the thread itself does not listen, only the + terminate does) and returns the answer, so the runner sees it at the + next unit of work and raises. + + Between forced beats the reporter throttles the way the legacy export + did, since hundreds of objects would otherwise be hundreds of HTTP + calls for information nobody reads at that resolution. An errored + beat answers `run`: the export itself must not fail because the beat + could not fly, and the backend that is unreachable has the lease to + decide what happens to the row." + [job-id on-cancel] + (let [last-beat (atom 0) + last (atom nil) + terminated (atom false) + check (fn [answer] + (when (= :skip (:action answer)) + (when (compare-and-set! terminated false true) + (l/info :hint "job cancelled by backend" + :job-id (str job-id)) + (on-cancel))) + answer) + report (fn [stage counters] + (->> (api/report-job-progress job-id + {:stage stage + :counters counters}) + (p/merr (fn [cause] + (l/warn :hint "beat failed to land" + :job-id (str job-id) + :cause cause) + (p/resolved {:action :run}))) + (p/fmap check))) + beats! (fn [stage counters & {:keys [force?]}] + (let [now (js/Date.now) + now? (or force? + (>= (- now @last-beat) report-throttle-ms))] + (if now? + (do (reset! last-beat now) + (reset! last {:stage stage :counters counters}) + (report stage counters)) + (p/resolved {:action :run :throttled? true}))))] + {:beat! beats! :last last})) + +;; ---- THE WATCHDOG + +(defn- with-watchdog! + "The forced beats of one run: one every `watchdog-interval-ms`, and + they repeat the last report the run made, so a cancellation that + lands mid-render (no object landing, no unit of work ending) is seen + within that interval. `stop!` clears it once the run settles." + [beats! last] + (let [timer (js/setInterval + #(beats! (:stage @last) (:counters @last) :force? true) + watchdog-interval-ms)] + (fn stop! [] (js/clearInterval timer)))) (defn- counter "One counter of the vocabulary: `kind` is `objects` or `pages`." @@ -110,25 +162,51 @@ :resource (rsc/create kind (or name (-> prepared first :name)))})) +;; ---- THE CANCEL CHECK + +(defn- check-cancelled! + "The local arm of the cancel: the watchdog's terminate already died + the in-flight thread; the units of work that follow check the flag + and unwind without rendering anything else. The backend's row is + already terminal, so the settle below only reports what happened." + [cancelled] + (when @cancelled + (ex/raise :type :internal + :code :job-cancelled + :hint "export job was cancelled"))) + ;; ---- THE RUNNERS +(defn- check-beat! + "A forced beat after a unit of work: a `skip` there is a cancelled + job, the same exception the legacy surfaces raised, so everything + below unwinds." + [answer] + (when (= :skip (:action answer)) + (ex/raise :type :internal + :code :job-cancelled + :hint "export job was cancelled")) + answer) + (defn- run-single! "One render, one object: the artifact IS the object, moved into the resource path the multipart settle will name." - [progress! plan] + [beats! plan] (let [{:keys [job-id resource counter-kind]} plan export (-> plan :prepared first) object (atom nil)] (job.utils/track! job-id (:path resource)) - (p/let [_ (progress! :rendering (counter counter-kind 0 1) :force? true) - _ (rd/render (assoc export - :job-id job-id - :skip-children (:skip-children plan)) - (fn [obj] - (reset! object obj) - (job.utils/track! job-id (:path obj)))) - _ (sh/move! (:path @object) (:path resource)) - _ (progress! :packaging (counter counter-kind 1 1) :force? true)] + (p/let [beat-one (beats! :rendering (counter counter-kind 0 1) :force? true) + _ (check-beat! beat-one) + _ (rd/render (assoc export + :job-id job-id + :skip-children (:skip-children plan)) + (fn [obj] + (reset! object obj) + (job.utils/track! job-id (:path obj)))) + _ (sh/move! (:path @object) (:path resource)) + beat-end (beats! :packaging (counter counter-kind 1 1) :force? true) + _ (check-beat! beat-end)] resource))) (defn- run-multiple! @@ -137,7 +215,7 @@ the progress counts what has landed so far. Any zipping error surfaces after the renders, so the failures of the writer do not compete with the ones of the render." - [progress! plan] + [beats! cancelled plan] (let [{:keys [job-id resource prepared total counter-kind]} plan failure (volatile! nil) rendered (volatile! 0) @@ -145,64 +223,80 @@ :on-error (fn [cause] (vreset! failure cause)) :on-progress (fn [_] nil)) append (fn [{:keys [filename path]}] + (check-cancelled! cancelled) (job.utils/track! job-id path) (vswap! rendered inc) - (progress! :rendering - (counter counter-kind @rendered total)) + (beats! :rendering + (counter counter-kind @rendered total)) (rsc/add-to-zip zip path (str/replace filename shapes/sanitize-file-regex "_")))] - (p/let [_ (progress! :rendering (counter counter-kind 0 total) :force? true) - _ (rd/with-scope prepared - (fn [scoped-render] - (p/all (map (fn [export] - (scoped-render - (assoc export :job-id job-id) - append)) - prepared)))) - error (if-let [cause @failure] - (p/rejected cause) - (rsc/close-zip zip)) - _ (do error) - _ (progress! :packaging (counter counter-kind total total) :force? true)] + (p/let [beat-one (beats! :rendering (counter counter-kind 0 total) :force? true) + _ (check-beat! beat-one) + _ (rd/with-scope prepared + (fn [scoped-render] + (p/all (map (fn [export] + (scoped-render + (assoc export :job-id job-id) + append)) + prepared)))) + error (if-let [cause @failure] + (p/rejected cause) + (rsc/close-zip zip)) + _ (do error) + beat-end (beats! :packaging (counter counter-kind total total) :force? true) + _ (check-beat! beat-end)] resource))) (defn- join-pdf! + "The pages render one file per page; `pdfunite` stitches one pdf out + of them all, the way the legacy export did." [job-id file-id paths] (let [path (job.utils/track! job-id (sh/tempfile :prefix (str/concat "penpot.pdfunite." file-id ".") :suffix ".pdf"))] - (p/let [_ (sh/run-cmd! "pdfunite" (into [] (concat (vec paths) [path])))] + (p/let [_ (sh/run-cmd! "pdfunite" (into [] (concat (vec paths) [path])))] path))) (defn- run-frames! "The frames render: a file per page, joined into the pdf of the file once every page has landed." - [progress! plan] + [beats! cancelled plan] (let [{:keys [job-id resource prepared total]} plan file-id (-> prepared first :file-id) paths (volatile! []) rendered (volatile! 0) on-object (fn [{:keys [path]}] + (check-cancelled! cancelled) (job.utils/track! job-id path) (vswap! paths conj path) (vswap! rendered inc) - (progress! :rendering - (counter :pages @rendered total)))] - (p/let [_ (progress! :rendering (counter :pages 0 total) :force? true) - _ (rd/with-scope prepared - (fn [scoped-render] - (p/all (map (fn [export] - (scoped-render - (assoc export :job-id job-id :is-wasm (:is-wasm plan)) - on-object)) - prepared)))) + (beats! :rendering + (counter :pages @rendered total)))] + (p/let [beat-one (beats! :rendering (counter :pages 0 total) :force? true) + _ (check-beat! beat-one) + _ (rd/with-scope prepared + (fn [scoped-render] + (p/all (map (fn [export] + (scoped-render + (assoc export :job-id job-id + :is-wasm (:is-wasm plan)) + on-object)) + prepared)))) joined (join-pdf! job-id file-id @paths) _ (sh/move! joined (:path resource)) - _ (progress! :packaging (counter :pages total total) :force? true)] + beat-end (beats! :packaging (counter :pages total total) :force? true) + _ (check-beat! beat-end)] resource))) +(defn- run-prepared! + [beats! cancelled plan] + (cond + (:single? plan) (run-single! beats! plan) + (:frames? plan) (run-frames! beats! cancelled plan) + :else (run-multiple! beats! cancelled plan))) + ;; ---- THE SETTLE (defn- failure-report @@ -229,24 +323,30 @@ (l/warn :hint "unable to settle the failed job" :job-id job-id :cause settle-cause)))))) -(defn- run-prepared! - [beats! plan] - (cond - (:single? plan) (run-single! beats! plan) - (:frames? plan) (run-frames! beats! plan) - :else (run-multiple! beats! plan))) - (defn run-export! - [{:keys [job-id] :as _job} params] - (let [session (atom nil) - beats! (start-beats! job-id)] + [{:keys [job-id]} params] + (let [session (atom nil) + cancelled (atom false) + _ (jobs/register! job-id) + stop-cmd! (fn [] + (reset! cancelled true) + ;; the hard-cancel of the render the run holds: the + ;; `renderer.wasm/with-scope` of the in-flight render + ;; registered a terminate callback for this job + (jobs/mark-cancelled job-id))] (p/catch - (p/let [session' (api/create-job-session job-id) - _ (reset! session session') - token (:session-token session') - _ (beats! :preparing {}) + (p/let [session' (api/create-job-session job-id) + _ (reset! session session') + token (:session-token session') + _ (l/info :hint "render session minted" + :job-id (str job-id)) + beats (start-beats! job-id stop-cmd!) + beats! (:beat! beats) + stop! (with-watchdog! beats! (:last beats)) + _ (check-beat! (beats! :preparing {} :force? true)) plan (make-plan token params) - resource (run-prepared! beats! plan) + resource (run-prepared! beats! cancelled plan) + _ (stop!) artifact {:path (str (:path resource)) :filename (:filename resource) :mtype (:mtype resource)} @@ -254,9 +354,11 @@ {:job-id job-id :session-id (:session-id session')} artifact) _ (job.utils/release! job-id) - _ (l/info :hint "export job settled" :job-id job-id + _ (l/info :hint "export job settled" :job-id (str job-id) :outcome "completed")] nil) (fn [cause] - (l/error :hint "export job failed" :job-id job-id :cause cause) + (l/error :hint "export job failed" :job-id (str job-id) :cause cause) (settle-failure! job-id session cause))))) + + diff --git a/exporter/src/app/core.cljs b/exporter/src/app/core.cljs index a1ddab7efa..05a2ee5a07 100644 --- a/exporter/src/app/core.cljs +++ b/exporter/src/app/core.cljs @@ -11,6 +11,8 @@ [app.browser :as bwr] [app.common.logging :as l] [app.config :as cf] + [app.consumer.config :as ccfg] + [app.consumer.worker :as worker] [app.http :as http] [app.jobs :as jobs] [app.jobs.utils :as job.utils] @@ -18,14 +20,26 @@ [app.wasm :as wasm] [app.wasm.pool :as wasm.pool] [app.wasm.worker :as wasm.worker] + [cuerdas.core :as str] [promesa.core :as p])) (enable-console-print!) (l/setup! {:app :info}) +(defn- roles-log + [] + (->> (ccfg/roles) + (map name) + (str/join ","))) + (defn start "Render workers run this same bundle, so the thread decides what gets booted: - the http server and its pools, or one render worker." + the http server and its pools, or one render worker. + + In the main thread the roles of the process decide the rest: `http` + boots the server and its pools exactly as always, `worker` boots the + consumer of the backend queue, and both can coexist during the + transition (the engines and the pools are shared)." [& _] (if-not ^boolean wt/isMainThread (wasm.worker/main) @@ -33,6 +47,7 @@ (l/info :msg "initializing" :public-uri (str (cf/get :public-uri)) :internal-uri (str (cf/get-internal-uri)) + :roles (roles-log) :version (:full cf/version)) (l/info :msg "headless wasm export available" :wasm-dir wasm/artifact-dir @@ -44,7 +59,13 @@ (jobs/init) (job.utils/init) (wasm.pool/init) - (http/init))))) + (when (ccfg/worker-enabled?) + (worker/start!)) + (when (contains? (ccfg/roles) :http) + (http/init)) + ;; With no role at all, the process boots into a no-op: the log + ;; above is the only surface of that mistake, and honest. + nil)))) (def main start) @@ -78,8 +99,13 @@ (p/do (shutdown-step "browser-pool" bwr/stop) (shutdown-step "wasm-worker-pool" wasm.pool/stop) + ;; the worker's pollers are inside their redis connections; ending + ;; redis first would leave the pops on broken connections + (when (ccfg/worker-enabled?) + (shutdown-step "consumer-worker" worker/stop!)) (shutdown-step "redis" redis/stop) - (shutdown-step "http" http/stop) + (when (contains? (ccfg/roles) :http) + (shutdown-step "http" http/stop)) (done))))) (.on proc/default "uncaughtException" diff --git a/exporter/src/app/jobs.cljs b/exporter/src/app/jobs.cljs index e3497b2861..5f38d86171 100644 --- a/exporter/src/app/jobs.cljs +++ b/exporter/src/app/jobs.cljs @@ -230,6 +230,50 @@ (l/warn :hint "error on job cancel callback" :job-id k :cause cause)))) result))))) +(defn register! + "The runtime record of a job this process runs. The local cancel arm + reads it (`mark-cancelled`), so the in-flight render of a claimed job + is terminable without the jobs substrate: the row is the backend's + business, this record is only what a terminate needs." + [job-id & {:keys [total]}] + (swap! registry assoc (str job-id) + {:job {:id job-id :state "running" :total total} + :cancelled? false})) + +(defn mark-cancelled + "Signals one job of this process as cancelled: raises the cancel flag + (the check between objects sees it), arms the shared-array signal (the + wasm worker sees it inside a render) and runs the cancel callbacks + (the terminate of the render worker). The durable row is already + terminal in the backend; no store write, no publish happens here. + + A job this process does not own, or a settled one, is a no-op." + [job-id] + (let [k (str job-id) + rt (get @registry k)] + (when-let [rt (and rt (:job rt))] + (when-not (terminal? rt) + (swap! registry update k (fn [rt] (some-> rt (assoc :cancelled? true)))) + (when-let [signal (:cancel-signal rt)] + (js/Atomics.store signal 0 1)) + (doseq [f (:cancel-fns rt)] + (try + (f) + (catch :default cause + (l/warn :hint "error on job cancel callback" :job-id k :cause cause)))) + true)))) + +(defn mark-cancelled! + "Marks a job of this process as cancelled and runs its cancel + callbacks (what terminates the render worker a hard-cancel needs) — + without touching the shared store. + + It is the local arm of the manual `jobs/cancel-local!` has: the jobs + substrate marks the row `cancelled` in the backend, and the worker's + own records only need to know, so the in-flight render dies." + [job-id] + (cancel-local! job-id)) + (defn cancel! "Cancels a job. One this process does not own is broadcast over the cancel topic, so whoever runs it acts on it. Idempotent." diff --git a/exporter/test/exporter_tests/worker_test.cljs b/exporter/test/exporter_tests/worker_test.cljs index df209539b6..5d677fb460 100644 --- a/exporter/test/exporter_tests/worker_test.cljs +++ b/exporter/test/exporter_tests/worker_test.cljs @@ -252,3 +252,73 @@ (respond!) (restore!) (done))))) + +;; ---- THE CANCEL + +(t/deftest cancel-runs-the-cancel-port + ;; the beat the backend refuses (a skip) is the cancel arriving: the + ;; port runs exactly once, the local flag says cancelled and the + ;; runner unwinds; the settle below is the backend's own row, not a + ;; new answer of this worker + (t/async done + (let [job-id (uuid/next) + sid (uuid/next) + params (export-params) + stages (volatile! []) + port (atom 0) + respond! (fake-fetch! + (fn [method body] + (when (= "report-job-progress" method) + (vswap! stages conj (get-in body [:progress :stage]))) + (case method + "claim-job" {:action :run :name "export-assets" :params params} + "create-job-session" {:session-id sid :session-token "session-token"} + "report-job-progress" {:action :skip} + {:action :run})))] + (p/let [_ (worker/process! (run-claim job-id))] + (t/testing "the settle of a cancelled job was a fail the backend answers skip" + (t/is (= "fail-job" (last (steps-of)))) + (let [fail (call-of "fail-job")] + (t/is (= :job-cancelled (get-in fail [:error :code]))))) + (respond!) + (done))))) + +(t/deftest cancel-kills-the-mid-skia-render + ;; the watchdog is what makes the cancel seen while the thread renders; + ;; the port it fires is the one renderer.wasm/with-scope registered, + ;; which terminates the worker + (t/async done + (let [job-id (uuid/next) + sid (uuid/next) + params (export-params) + stages (volatile! []) + terminated (atom 0) + mark! (atom nil) + respond! (fake-fetch! + (fn [method body] + (when (= "report-job-progress" method) + (vswap! stages conj (get-in body [:progress :stage]))) + (case method + "claim-job" {:action :run :name "export-assets" :params params} + "create-job-session" {:session-id sid :session-token "session-token"} + ;; the first beat runs, everything else skips: + ;; the cancel lands right after the claim + "report-job-progress" (if (empty? @stages) + {:action :run} + {:action :skip}) + {:action :run}))) + render-restore (let [original rd/render] + (set! rd/render + (fn [_params _on-object] + ;; never resolves: the render only ends + ;; when its thread is terminated, and the + ;; registry entry the with-scope registered + ;; does not exist in this fake + (p/delay 30000))) + (fn [] (set! rd/render original)))] + (p/let [_ (worker/process! (run-claim job-id))] + (t/testing "the run ended in a fail job, not a complete" + (t/is (= "fail-job" (last (steps-of))))) + (render-restore) + (respond!) + (done)))))