mirror of
https://github.com/penpot/penpot.git
synced 2026-10-29 22:10:34 +00:00
✨ Cancel exports through the beats and boot by roles
A cancellation reaches the worker through its beats: a `skip` answer from `report-job-progress` says the job can no longer hear it. The runner raises `:job-cancelled` between units of work (the same exception the legacy surfaces raised), and the render mid-Skia dies by terminate: the watchdog of the beats fires the cancel port the wasm pool already serves (`jobs/mark-cancelled` raises the flag, arms the shared-array signal and runs the terminate callbacks `renderer.wasm/with-scope` registered), and the run settles as `fail-job` with the session id, which the backend answers skip on a row it already cancelled. The settle is what ends the render session's life, whatever the write says: complete and fail now close the named session even when they lose the write, so no settle of the worker leaves a session behind. The process boots by roles (`PENPOT_EXPORTER_ROLES`): `http` boots the server and its pools exactly as today, `worker` boots K pollers of the backend queue, and both roles coexist during the transition sharing the engines; the shutdown unwinds the worker before redis, so no poller pops on a closing connection. AI-assisted-by: glm-5.3-flash
This commit is contained in:
Andrey Antukh
parent
c0b4430293
commit
8391fbdf07
@ -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)))))
|
||||
|
||||
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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."
|
||||
|
||||
@ -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)))))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user