diff --git a/.serena/memories/exporter/core.md b/.serena/memories/exporter/core.md index f90d0314c3..1ef13cfb62 100644 --- a/.serena/memories/exporter/core.md +++ b/.serena/memories/exporter/core.md @@ -42,13 +42,16 @@ on this side: the row of the job, its cancel and its result are the backend's. - `exporter.main` mirrors the backend's `app.main`: a `system-config` def reading env at the wiring layer, a public `system` atom with the running map, `start`/`stop`/`restart` (`^:async`, resolving `:started`/`:stopped`/`:restarted`), `start-custom` as the test seam, idempotent `install-process-handlers` for the `uncaughtException`/SIGTERM/SIGINT wiring (runs on `start`, never on namespace load), `^:dev/before-load-async`/`^:dev/after-load-async` hooks bridging the promise lifecycle to shadow's `done` callbacks (a hook failure logs, never sticks the reload), and `main` as the entry point: shadow boots `exporter.main/main`, and the old `app.core` reload path (`stop` and the `:devtools` wiring) sits commented out until `app.*` is fully ported. The entry leaves info logging enabled for the new tree (`setup!` on the `exporter` logger, mirroring the old `{:app :info}`); without it the default level filters everything below fatal. - Service config maps take plain (non-namespaced) param keys (`{:exporter.browser/pool {:max 2}}`): the top-level key already scopes the component, so namespacing the params only adds noise. Component keys and error reasons stay namespaced. - `exporter.wasm.pool` follows the backend `db/run!` shape: `run [cfg f & args]` resolves the pool from `cfg` (the `:exporter.wasm.pool/pool` service instance, or a raw pool), acquires a worker, applies `f` to the cfg carrying it under `:exporter.wasm.pool/worker` plus any extra args, and returns the worker — or destroys it when `f` fails, like a rollback. Pure pool mechanics only (`init`/`halt`, `terminate`, `await-ready`); the render protocol lives one level up. -- `exporter.wasm.scope` owns the lease: `run [sys on-scope]` lends one worker for the whole batch (serializer, per-lease signal buffer, mid-render abort interval), and the public low-level `render-on [cfg params]` renders on the leased worker — asserting `:worker-not-leased`/`:timeout-ms-missing` (`:assertion`, never a silent default: a bare `::key` resolves to the CURRENT ns, never to an alias, so cross-ns keys are literals or `::alias/key`) — with the rest preconfigured and namespaced by setter (`:exporter.renderer/on-object` from the caller, `:exporter.wasm.scope/cancel-buffer` from the lease); the check is mandatory at the lease entry and runs here when configured. -- `exporter.renderer.browser` merges the bitmap/pdf/svg flows into one `render [ctx params on-object]` (branching by type, `on-object` always awaited, unknown types raise); `ctx` carries what the job params do not freeze (`:browser-pool`, `:base-uri`, `:public-uri`, `:svgo?`). Dead xml predicates from the old tree did not survive the merge. -- `exporter.browser.scope` owns the browser batch: `run [sys task]` fires every export through its own checkout, all at once, bounded by the pool. No lease — a browser batch shares nothing but the pool, so fan-out only. The facade validates once, partitions by backend and joins both scopes. +- `exporter.wasm.scope` owns the lease: `run [cfg on-scope]` lends one worker for the whole batch (serializer, per-lease signal buffer, mid-render abort interval), and the public low-level `render-on [cfg params]` renders on the leased worker — asserting `:worker-not-leased`/`:timeout-ms-missing` (`:assertion`, never a silent default: a bare `::key` resolves to the CURRENT ns, never to an alias, so cross-ns keys are literals or `::alias/key`) — with the rest preconfigured and namespaced by setter (`:exporter.renderer/on-object` from the caller, `:exporter.wasm.scope/cancel-buffer` from the lease); the check is mandatory at the lease entry and runs here when configured. +- `exporter.renderer.browser` merges the bitmap/pdf/svg flows into one `render [cfg params on-object check-cancelled]` (branching by type, `on-object` always awaited, unknown types raise); `cfg` carries what the job params do not freeze (`:exporter.browser/pool`, `:base-uri`, `:public-uri`, `:svgo?`). Dead xml predicates from the old tree did not survive the merge. +- `exporter.browser.scope` owns the browser batch: `run [cfg exports]` fires every export through its own checkout, all at once, bounded by the pool. No lease — a browser batch shares nothing but the pool, so fan-out only. The facade validates once, partitions by backend and joins both scopes. - `exporter.renderer.wasm` is a thin single-shot over the scope lease. The scope fn survives one level down, as `exporter.wasm.scope/run`. - `exporter.renderer` is the malli-validated facade with one entry point: `render [cfg & task]` takes a task map (`:exports :on-object :check-cancelled`) and fires the whole batch at once — the browser exports through `exporter.browser.scope`, the wasm ones through one leased worker — resolving nil once every export landed (objects travel through `on-object`). `cfg` is task-free infrastructure with process lifetime (pools, uris, flags); everything per-task rides in the validated task map. One closed schema validates data plus injections, so a bad task — or a misspelled key — never touches a pool; task validation raises `:data-validation`. Each driver takes the same `[cfg params on-object check-cancelled]` shape (`headless?` is now `wasm?`: the browser also renders headless, so the old name lied) and implements cancel internally: cooperative checkpoints on the browser, checkpoints plus a mid-render abort of its own leased worker on wasm. The lease lives one level down, in `exporter.wasm.scope/run`, which the facade calls with the task keys set once in `cfg`; the browser batch goes through `exporter.browser.scope/run [cfg exports]`. -- The runner stays backend-blind: plan → render batch → settle, with beats/watchdog/settle untouched. Its `on-object` calls the same injected check the drivers use. When `exporter.jobs` is ported, its only cancel surface toward renderers is one zero-arg check constructor per job; the old `cancel-fns`/terminate-registration apparatus does not survive (abort ownership moved to the lease holder). -- `exporter.main` wires both pools (`:exporter.browser/pool` and `:exporter.wasm.pool/pool`, each sized to the worker concurrency). No test boots the production wiring: the wasm pool warms one worker eagerly and the worker script in a test process would be the test bundle itself; the wiring is asserted as data and the lifecycle rides fake configs. +- `exporter.jobs` is the local cancel arm, smaller than the legacy `app.jobs`: one flag per claimed job (`register`/`mark-cancelled` returning true/nil/`cancelled?`/`release`) and one zero-arg check constructor per job (`check-cancelled`, raising `:job-cancelled`). No terminate callbacks and no shared-array signal travel through here — the abort belongs to whoever holds the render (the per-lease interval of `exporter.wasm.scope` terminates its own worker). The runner releases the registry entry on every settle; the legacy runner never did (one leaked entry per job). +- `exporter.consumer.plan` is the render plan (`prepare-exports`, `count-objects`, `sanitize-file-regex`), dispatching on `exporter.renderer/wasm?` instead of the legacy `headless?`. +- `exporter.consumer` is the runner: `run-export [cfg claim params]` keeps the legacy beats/watchdog/settle (first breath, render session, multipart complete or `fail-job`, temp release) but renders through the facade (`render [cfg & task]`) with one check from `exporter.jobs` shared between units of work and drivers. `cfg` is the render view; the claim names the job, `params` carries the frozen job params. The settle never rejects; only bugs of this process do. +- `exporter.consumer.worker` is the queue poller (`read-payload`, `process`, `poll-once`, `start`/`stop` with a running flag, no promesa): K connections over the `:exporter` queue, claim → `run-export` through the `:render` view → settle before the next pop. `stop` quits every connection and ends every loop (the legacy loops kept failing turns after the stop); boot cleans orphaned temp files first. +- `exporter.main` wires both pools (`:exporter.browser/pool` and `:exporter.wasm.pool/pool`, each sized to the worker concurrency) plus `:exporter.consumer/worker` (`:concurrency`, `:queue-key`, the `:render` view with both pools as `ref`s and the static uris/flags). No test boots the production wiring: the wasm pool warms one worker eagerly and the worker script in a test process would be the test bundle itself; the wiring is asserted as data and the lifecycle rides fake configs. - Async test vars prove their chain only when really awaited: a test passing with 0 assertions means the chain broke upstream. ## Render details diff --git a/exporter/src/exporter/consumer.cljs b/exporter/src/exporter/consumer.cljs new file mode 100644 index 0000000000..6351bc6823 --- /dev/null +++ b/exporter/src/exporter/consumer.cljs @@ -0,0 +1,371 @@ +;; 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 exporter.consumer + "The runner of the `:export-assets` job. + + Runs one export end to end: the render plan of + `exporter.consumer.plan` (the name transducers and the partitioning + the export surfaces always kept), the pdf assembly of the frames + run, the beats shaped by the progress contract (`:preparing`, + `:rendering`, `:packaging`, with the `objects` and `pages` counters + of every partition) and one settle: a `complete-job` multipart call + that stores the artifact and closes the session the backend minted + in the same step. The render runs as the job's owner through the + session `create-job-session` opened. + + 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, with the same check the drivers run; inside a render the + lease holder aborts its own worker. The local registry entry is + released on every settle — unlike the legacy runner, which only + released the temp files and leaked one entry per job. + + 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 + reject. There are no promesa chains in this namespace." + (:require + [app.common.exceptions :as ex] + [app.common.logging :as l] + [app.common.uuid :as uuid] + [app.consumer.api :as api] + [app.handlers.resources :as rsc] + [app.jobs.utils :as job.utils] + [app.util.shell :as sh] + [cuerdas.core :as str] + [exporter.consumer.plan :as plan] + [exporter.jobs :as jobs] + [exporter.renderer :as renderer])) + +(def ^:private report-throttle-ms 250) +(def ^:private watchdog-interval-ms 1000) + +;; ---- PROGRESS AND CANCEL (the milestones of the contract) + +(defn- start-beats + "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 marks the local flag the + checks read; the in-flight render aborts itself) 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. The beat never rejects." + [job-id on-cancel] + (let [last-beat (atom 0) + last (atom nil) + fired (atom false) + check (fn [answer] + (when (= :skip (:action answer)) + (when (compare-and-set! fired false true) + (l/info :hint "job cancelled by backend" + :job-id (str job-id)) + (on-cancel))) + answer) + report (^:async fn [stage counters] + (try + (check (await (api/report-job-progress job-id {:stage stage + :counters counters}))) + (catch :default cause + (l/warn :hint "beat failed to land" + :job-id (str job-id) + :cause cause) + {:action :run}))) + beat (^:async 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}) + (await (report stage counters))) + {:action :run :throttled? true})))] + {:beat beat :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." + [beat last] + (let [timer (js/setInterval + #(beat (: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`." + [kind current total] + {kind {:current current :total total}}) + +;; ---- THE PARAMS + +(defn- frame-item? + "The frames export names pages of a file without a `type`: the pdf + of their pages. A shape item always names its type." + [item] + (nil? (:type item))) + +(defn- ->uuid + "One id of the claim into what the renderer wants: the claim carries + the job row JSON, every id a string, while the render spec only + takes uuid objects. Values that already are uuids (tests, callers in + process) pass through." + [v] + (if (string? v) (uuid/uuid v) v)) + +(defn- normalize-items + "The items the claim delivered are the plain JSON the job row holds: + the type arrives as text and the renderer dispatches on the keywords + the legacy surface received; every id arrives as text and the render + spec only takes uuid objects. The frames items are pages, not typed + shapes: they name the pdf of a page each." + [items] + (->> items + (mapv (fn [item] + (cond-> item + (string? (:type item)) (update :type keyword) + (string? (:file-id item)) (update :file-id ->uuid) + (string? (:page-id item)) (update :page-id ->uuid) + (string? (:object-id item)) (update :object-id ->uuid) + (string? (:share-id item)) (update :share-id ->uuid) + (nil? (:type item)) (assoc :type :pdf :scale 1 :suffix "")))))) + +(defn- make-plan + "The render plan the legacy handlers prepare: the same transducers of + names and partition size they served the old surface with, the same + `single?` rule (one prepared export with one object, not forced), + and the artifact the whole run fills." + [job-id token {:keys [exports force-multiple name skip-children is-wasm]}] + (let [items (normalize-items exports) + frames? (boolean (every? frame-item? items)) + prepared (plan/prepare-exports items token is-wasm) + single? (and (not frames?) + (not (true? force-multiple)) + (= 1 (count prepared)) + (= 1 (count (-> prepared first :objects)))) + kind (cond + single? (-> prepared first :type) + frames? :pdf + :else :zip)] + {:frames? frames? + :single? single? + :job-id (->uuid job-id) + :prepared prepared + :total (plan/count-objects prepared) + :skip-children skip-children + :is-wasm (boolean is-wasm) + :counter-kind (if frames? :pages :objects) + :resource (rsc/create kind + (or name (-> prepared first :name)))})) + +;; ---- 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)) + (throw (ex/error :type :internal + :code :job-cancelled + :hint "export job was cancelled"))) + answer) + +(defn- ^:async run-single + "One render, one object: the artifact IS the object, moved into the + resource path the multipart settle will name." + [cfg beat plan check-cancelled] + (let [{:keys [job-id resource counter-kind]} plan + export (-> plan :prepared first) + object (atom nil)] + (job.utils/track job-id (:path resource)) + (check-beat (await (beat :rendering (counter counter-kind 0 1) :force? true))) + (await (renderer/render cfg + :exports [(assoc export + :job-id job-id + :skip-children (:skip-children plan))] + :on-object (fn [obj] + (reset! object obj) + (job.utils/track job-id (:path obj)) + nil) + :check-cancelled check-cancelled)) + (await (sh/move (:path @object) (:path resource))) + (check-beat (await (beat :packaging (counter counter-kind 1 1) :force? true))) + resource)) + +(defn- ^:async run-multiple + "The multi-object render: every prepared export renders through the + one batch of the facade, appending each object to the zip as it + lands, and 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." + [cfg beat plan check-cancelled] + (let [{:keys [job-id resource prepared total counter-kind]} plan + failure (volatile! nil) + rendered (volatile! 0) + zip (rsc/create-zip :resource resource + :on-error (fn [cause] (vreset! failure cause)) + :on-progress (fn [_] nil)) + append (fn [{:keys [filename path]}] + (check-cancelled) + (job.utils/track job-id path) + (vswap! rendered inc) + (beat :rendering + (counter counter-kind @rendered total)) + (rsc/add-to-zip zip path + (str/replace filename + plan/sanitize-file-regex + "_")) + nil)] + (check-beat (await (beat :rendering (counter counter-kind 0 total) :force? true))) + (await (renderer/render cfg + :exports (mapv #(assoc % :job-id job-id) prepared) + :on-object append + :check-cancelled check-cancelled)) + (when-let [cause @failure] + (throw cause)) + (await (rsc/close-zip zip)) + (check-beat (await (beat :packaging (counter counter-kind total total) :force? true))) + resource)) + +(defn- ^:async 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"))] + (await (sh/run-cmd "pdfunite" (into [] (concat (vec paths) [path])))) + path)) + +(defn- ^:async run-frames + "The frames render: a file per page, joined into the pdf of the file + once every page has landed." + [cfg beat plan check-cancelled] + (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) + (job.utils/track job-id path) + (vswap! paths conj path) + (vswap! rendered inc) + (beat :rendering + (counter :pages @rendered total)) + nil)] + (check-beat (await (beat :rendering (counter :pages 0 total) :force? true))) + (await (renderer/render cfg + :exports (mapv #(assoc % :job-id job-id + :is-wasm (:is-wasm plan)) + prepared) + :on-object on-object + :check-cancelled check-cancelled)) + (let [joined (await (join-pdf job-id file-id @paths))] + (await (sh/move joined (:path resource))) + (check-beat (await (beat :packaging (counter :pages total total) :force? true))) + resource))) + +(defn- run-prepared + [cfg beat check-cancelled plan] + (cond + (:single? plan) (run-single cfg beat plan check-cancelled) + (:frames? plan) (run-frames cfg beat plan check-cancelled) + :else (run-multiple cfg beat plan check-cancelled))) + +;; ---- THE SETTLE + +(defn- failure-report + "The error shape the backend stores for `job.error`: type and code + keywords, hint the human message. A cause that is not a penpot error + becomes an internal one, with only its message as the hint." + [cause] + (let [data (ex-data cause)] + {:type (or (:type data) :internal) + :code (or (:code data) :export-failed) + :hint (or (:hint data) (ex-message cause) "export failed")})) + +(defn cancelled? + "Whether `cause` ends a run the user asked to end: a cancelled + export is a normal ending, reported and logged as one, never as a + failure with a trace." + [cause] + (= :job-cancelled (:code (ex-data cause)))) + +(defn- ^:async settle-failure + "Settles the job as failed, after releasing the temp files the run + owns. The settle never rejects: if the failure cannot reach the + backend, the row keeps running until the lease GC turns it `aborted`, + which is the honest state for a worker that cannot talk." + [job-id session cause] + (await (job.utils/release job-id)) + (jobs/release job-id) + (try + (await (api/fail-job job-id (failure-report cause) + :session-id (some-> @session :session-id))) + (catch :default settle-cause + (l/warn :hint "unable to settle the failed job" + :job-id job-id :cause settle-cause)))) + +(defn ^:async run-export + "Runs one claimed export to its settle: first breath, render + session, plan, render batch, multipart complete. `cfg` is the render + config (a view over the running system); the claim names the job and + `params` carries the frozen job params." + [cfg {:keys [job-id]} params] + (let [session (atom nil) + stop-watchdog (atom nil) + _ (jobs/register job-id) + stop-cmd (fn [] (jobs/mark-cancelled job-id)) + {:keys [beat last]} (start-beats job-id stop-cmd) + check-cancelled (jobs/check-cancelled job-id)] + (try + (reset! stop-watchdog (with-watchdog beat last)) + ;; the first breath right after the claim, before the + ;; session roundtrip: the widget moves from queued to + ;; exporting on it, and a cancel that landed meanwhile + ;; raises here instead of after wasted work + (check-beat (await (beat :preparing {} :force? true))) + (let [session' (await (api/create-job-session job-id))] + (reset! session session') + (l/info :hint "render session minted" + :job-id (str job-id)) + (let [plan (make-plan job-id (:session-token session') params) + resource (await (run-prepared cfg beat check-cancelled plan))] + ;; the watchdog outlives the render, so the run stops it on + ;; the way out: a run that ends badly stops beating too, or + ;; its interval would knock on a dead job forever, one HTTP + ;; call a second that only ever answers skip + (@stop-watchdog) + (let [artifact {:path (str (:path resource)) + :filename (:filename resource) + :mtype (:mtype resource)}] + (await (api/complete-job-with-artifact + {:job-id job-id :session-id (:session-id session')} + artifact)) + (await (job.utils/release job-id)) + (jobs/release job-id) + (l/info :hint "export job settled" :job-id (str job-id) + :outcome "completed") + nil))) + (catch :default cause + (when-let [stop @stop-watchdog] (stop)) + (if (cancelled? cause) + (l/info :hint "export job cancelled" :job-id (str job-id)) + (l/error :hint "export job failed" :job-id (str job-id) :cause cause)) + (await (settle-failure job-id session cause)))))) diff --git a/exporter/src/exporter/consumer/plan.cljs b/exporter/src/exporter/consumer/plan.cljs new file mode 100644 index 0000000000..5019dc4308 --- /dev/null +++ b/exporter/src/exporter/consumer/plan.cljs @@ -0,0 +1,91 @@ +;; 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 exporter.consumer.plan + "The render plan of an assets run. + + The items the backend froze in the job become the renders this + process runs: file names built once and kept free of duplicates, then + grouped by scale and type, one render per group on the wasm engine + and one DOM page per partition of fifty on the browser engine. + `exporter.consumer` runs whatever this namespace plans, and this + namespace only plans." + (:require + [app.common.data :as d] + [app.util.mime :as mime] + [cuerdas.core :as str] + [exporter.renderer :as renderer])) + +;; Regex to clean namefiles +(def sanitize-file-regex #"[\\/:*?\"<>|]") + +(defn count-objects + [exports] + (reduce + 0 (map (comp count :objects) exports))) + +(defn- assoc-file-name + "A transducer that assocs a candidate filename and avoid duplicates" + [] + (letfn [(find-candidate [params used] + (loop [index 0] + (let [candidate (str (:name params) + (:suffix params "") + (when (pos? index) + (str/concat "-" (inc index))) + (mime/get-extension (:type params)))] + (if (contains? used candidate) + (recur (inc index)) + candidate))))] + (fn [rf] + (let [used (volatile! #{})] + (fn + ([] (rf)) + ([result] (rf result)) + ([result params] + (let [candidate (find-candidate params @used) + params (assoc params :filename candidate)] + (vswap! used conj candidate) + (rf result params)))))))) + +(def ^:const ^:private + default-partition-size 50) + +(defn prepare-exports + [exports token is-wasm] + (letfn [(process-group [[part1 :as group]] + ;; The browser path renders one DOM page per partition; wasm + ;; takes the whole group in one render. + (if (renderer/wasm? {:is-wasm is-wasm :type (:type part1)}) + [(build-render group)] + (sequence (comp (partition-all default-partition-size) + (map build-render)) + group))) + + (build-render [[part1 :as part]] + (cond-> {:file-id (:file-id part1) + :page-id (:page-id part1) + :name (:name part1) + :token token + :type (:type part1) + :scale (:scale part1) + :objects (mapv part-entry->object part)} + ;; a nil share-id is not an absent one for the render + ;; spec: only name it when the item came through one + (some? (:share-id part1)) (assoc :share-id (:share-id part1)))) + + (part-entry->object [entry] + {:id (:object-id entry) + :filename (:filename entry) + :name (:name entry) + :suffix (:suffix entry)})] + + (let [xform (comp + (map #(assoc % :token token)) + (assoc-file-name))] + (->> (sequence xform exports) + (d/group-by (juxt :scale :type)) + (map second) + (into [] (mapcat process-group)))))) diff --git a/exporter/src/exporter/consumer/worker.cljs b/exporter/src/exporter/consumer/worker.cljs new file mode 100644 index 0000000000..087979a966 --- /dev/null +++ b/exporter/src/exporter/consumer/worker.cljs @@ -0,0 +1,149 @@ +;; 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 exporter.consumer.worker + "The consumer of the backend `:exporter` queue. + + K pollers run in parallel, each one owning its Redis connection: the + blocking BLPOP of a poller holds that connection for the whole wait, + and the poller does not come back to the queue until the job it + claimed has settled, so the connection is both the semaphore of the + slot and the backpressure — nothing the queue receives can appear + twice taken, and no process ever runs more exports than it has + connections. + + The dispatcher pushes `[job-id, scheduled-at]` as JSON into + `penpot.worker.queue::exporter`. Corrupt payloads are logged + and dropped; a Redis hiccup is a lost turn, not a lost job: the + poller waits a bit and pops again (what was already delivered is + claimed, and the loop does not stop the process). Stopping quits + every connection and ends every loop: unlike the legacy worker, + whose loops kept failing turns after the stop, no poller survives + its handle. + + The worker is a system component (`:exporter.consumer/worker`): its + config carries the queue knobs (`:concurrency`, `:queue-key`, the + `:connect` factory) and the `:render` view the runner renders + through. There are no promesa chains in this namespace." + (:require + ["ioredis" :as redis] + [app.common.logging :as l] + [app.config :as cf] + [app.consumer.api :as api] + [app.consumer.config :as ccfg] + [app.jobs.utils :as job.utils] + [exporter.consumer :as consumer] + [exporter.utils.system :as system])) + +(def ^:private poll-timeout-s 5) +(def ^:private reconnect-delay-ms 1000) + +(defn- wait + [ms] + (js/Promise. (fn [resolve] (js/setTimeout resolve ms)))) + +(defn read-payload + "The JSON payload of the dispatcher: `job-id` and the `scheduled-at` + of the row when it was pushed. Anything else is a corrupt payload: + none, and the caller drops it." + [payload] + (try + (let [[job-id scheduled-at] (js->clj (js/JSON.parse payload))] + {:job-id job-id :scheduled-at scheduled-at}) + (catch :default cause + (l/warn :hint "corrupt queue payload dropped" + :payload (str payload) :cause cause) + nil))) + +(defn ^:async process + "One payload of the queue: claim the job, and when the backend gives + it to this poller, run the export to its settle. `skip` means + something else won it (or it is gone): the payload is dropped, with + the state the backend says." + [cfg {:keys [job-id scheduled-at]}] + (let [answer (await (api/claim-job job-id scheduled-at))] + (if (= :skip (:action answer)) + (l/info :hint "skipping job" :job-id (str job-id) + :name (:name answer) :status (:status answer)) + (do + (l/info :hint "running job" :job-id (str job-id) :name (:name answer)) + (await (consumer/run-export (:render cfg) {:job-id job-id} (:params answer))) + (l/info :hint "job settled" :job-id (str job-id)) + nil)))) + +;; ---- THE POLLER + +(defn ^:async poll-once + "One turn: pop the queue, claim what came, and wait for the settle. + An empty pop and a dropped payload both resolve; a claimed job is + awaited the whole run." + [conn key on-payload] + (let [answer (await (.blpop ^js conn key poll-timeout-s)) + payload (second answer) + claim (when (some? payload) (read-payload payload))] + (when (some? claim) + (await (on-payload claim)) + nil))) + +(defn- ^:async poll-loop + "The wait of one poller, turn after turn, until the handle stops. A + failed turn (Redis down, a bug) is logged, never allowed to end the + loop; a delay keeps the retry from spinning while the connection + rebuilds." + [handle conn key on-payload] + (try + (await (poll-once conn key on-payload)) + (catch :default cause + (l/error :hint "poller turn failed" :key key :cause cause) + (await (wait reconnect-delay-ms)))) + (when @(:running handle) + (await (poll-loop handle conn key on-payload)))) + +(defn- open-connection + "One Redis connection of one poller. ioredis reconnects on its own + strategy, so a fallen Redis comes back with this process doing + nothing but wait (and the failed pop comes right after)." + [] + (new redis/default (cf/get :redis-uri))) + +;; ---- THE START + +(defn ^:async start + "Launches the pollers over the queue and resolves the handle their + `stop` needs. A previous process leaves no temp files behind: the + boot cleans what the crash owned before the first pop." + [{:keys [concurrency queue-key connect] :as cfg}] + (await (job.utils/clean-orphans)) + (let [running (atom true) + handle {:running running :conns (atom [])} + on-payload (fn [claim] (process cfg claim)) + open (or connect open-connection) + k (or concurrency (ccfg/concurrency))] + (l/info :hint "worker started" :pollers k :queue queue-key) + (doseq [_ (range k)] + (let [conn (open)] + (swap! (:conns handle) conj conn) + (poll-loop handle conn queue-key on-payload))) + handle)) + +(defn ^:async stop + "Ends the pollers, closing their connections and their loops: the + running flag falls first, so a turn that ends after the quit does + not pop again. Idempotent: a second stop quits nothing." + [handle] + (when (and (some? handle) + (compare-and-set! (:running handle) true false)) + (await (js/Promise.allSettled + (mapv (fn [conn] (.quit ^js conn)) @(:conns handle)))) + nil)) + +(defmethod system/init-key :exporter.consumer/worker + [_ cfg] + (start cfg)) + +(defmethod system/halt-key :exporter.consumer/worker + [_ handle] + (stop handle)) diff --git a/exporter/src/exporter/jobs.cljs b/exporter/src/exporter/jobs.cljs new file mode 100644 index 0000000000..82cece45d8 --- /dev/null +++ b/exporter/src/exporter/jobs.cljs @@ -0,0 +1,69 @@ +;; 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 exporter.jobs + "The local arm of the job life this process runs. + + The durable row of a job is the backend's business: the worker claims + it, beats against it and settles it through the management API. This + namespace keeps only the runtime bit that cannot live in a row — the + cancel flag the parts of a run read between renders — keyed by the + job id of the claimed job. + + This is deliberately smaller than the legacy `app.jobs`: the abort + itself belongs to whoever holds the render (the per-lease interval of + `exporter.wasm.scope` terminates its own worker when the check + fails), so no terminate callbacks and no shared-array signal travel + through here. The only surface toward the renderers is one zero-arg + check constructor per job." + (:require + [app.common.exceptions :as ex])) + +(defonce ^:private registry (atom {})) + +(defn register + "The runtime record of a job this process runs: one cancel flag, + down until something marks it." + [job-id] + (swap! registry assoc (str job-id) {:cancelled? false}) + nil) + +(defn cancelled? + "True once the job has been marked cancelled in this process. The + check between objects reads it; inside a render the leased worker + reads its own abort instead." + [job-id] + (boolean (:cancelled? (get @registry (str job-id))))) + +(defn mark-cancelled + "Signals one job of this process as cancelled: the flag the checks + read goes up. A job this process does not own, or a settled one, is + a no-op. Returns true when this call marked it, nil otherwise." + [job-id] + (let [k (str job-id) + rt (get @registry k)] + (when (some? rt) + (when-not (:cancelled? rt) + (swap! registry assoc-in [k :cancelled?] true) + true)))) + +(defn release + "Drops the local runtime entry once the job settled. The row stays in + the backend until it is read once more or expires." + [job-id] + (swap! registry dissoc (str job-id)) + nil) + +(defn check-cancelled + "The zero-arg check of one job: silent while it lives, raising + `:job-cancelled` once marked. The same fn travels into the render + task and runs between units of work." + [job-id] + (fn [] + (when (cancelled? job-id) + (throw (ex/error :type :internal + :code :job-cancelled + :hint "export job was cancelled"))))) diff --git a/exporter/src/exporter/main.cljs b/exporter/src/exporter/main.cljs index 4e3948e065..38d363b001 100644 --- a/exporter/src/exporter/main.cljs +++ b/exporter/src/exporter/main.cljs @@ -19,8 +19,9 @@ [app.config :as cf] [app.consumer.config :as ccfg] ;; Loaded for their init-key/halt-key methods, which is what makes - ;; the pools part of the system below. + ;; the pools and the queue consumer part of the system below. [exporter.browser] + [exporter.consumer.worker] [exporter.utils.system :as system] [exporter.wasm.pool])) @@ -28,9 +29,18 @@ (def system-config "The production wiring. Env-derived values are read here, at the - wiring layer; the components themselves take plain data." - {:exporter.browser/pool {:max (ccfg/concurrency)} - :exporter.wasm.pool/pool {:max (ccfg/concurrency)}}) + wiring layer; the components themselves take plain data. The worker + renders through a view of the running pools, resolved by refs, plus + the static render config." + {:exporter.browser/pool {:max (ccfg/concurrency)} + :exporter.wasm.pool/pool {:max (ccfg/concurrency)} + :exporter.consumer/worker {:concurrency (ccfg/concurrency) + :queue-key (ccfg/queue-key) + :render {:exporter.browser/pool (system/ref :exporter.browser/pool) + :exporter.wasm.pool/pool (system/ref :exporter.wasm.pool/pool) + :base-uri (cf/get-internal-uri) + :public-uri (cf/get :public-uri) + :svgo? (contains? cf/flags :exporter-svgo)}}}) ;; The running system map, or nil when nothing is started. ;; Counterpart of the backend's `app.system/system` var. diff --git a/exporter/test/exporter_tests/consumer_test.cljs b/exporter/test/exporter_tests/consumer_test.cljs new file mode 100644 index 0000000000..1b03e0f2ba --- /dev/null +++ b/exporter/test/exporter_tests/consumer_test.cljs @@ -0,0 +1,266 @@ +;; 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 exporter-tests.consumer-test + "The export runner of the new tree: plan → render batch → settle, + over a fake management http and a stub renderer. The beats paint the + stage vocabulary in order and the settle carries the artifact; a + failure or a cancel settles as fail-job and never leaks the beat." + (:require + ["node:fs/promises" :as fsp] + ["undici" :as http] + [app.common.transit :as transit] + [app.common.uuid :as uuid] + [app.consumer.api :as api] + [app.util.shell :as sh] + [cljs.test :as t :include-macros true] + [cuerdas.core :as cstr] + [exporter.consumer :as consumer] + [exporter.jobs :as jobs] + [exporter.renderer :as renderer])) + +;; ---- THE MANAGEMENT FAKE + +(def ^:private calls + "One entry per management call: [method, body]." + (atom [])) + +(defn- method-of + [uri] + (last (cstr/split (str uri) "/"))) + +(defn- steps-of + [] + (mapv first @calls)) + +(defn- call-of + [method] + (some (fn [[m body]] + (when (= m method) body)) + @calls)) + +(defn- fake-fetch + [respond] + (reset! calls []) + (let [original api/fetch] + (set! api/fetch + (fn [uri request] + (let [method (method-of uri) + body (if (instance? http/FormData (.-body request)) + (.-body request) + (transit/decode-str (.-body request)))] + (swap! calls conj [method body]) + (js/Promise.resolve #js {:status 200 + :text (constantly + (js/Promise.resolve + (transit/encode-str + (respond method body))))})))) + (fn [] (set! api/fetch original)))) + +;; ---- THE RENDER STUB + +(defn- stub-render + "A renderer stub in place of the facade: writes one file of + `content` per object of every export and calls back with it. When + `captured` is an atom, it also keeps the last task it received." + ([content] (stub-render content nil)) + ([content captured] + (let [original renderer/render] + (set! renderer/render + (fn [_cfg & {:as task}] + (when (some? captured) + (reset! captured task)) + (js/Promise. + (fn [resolve reject] + (let [writes (mapv (fn [_] + (let [path (sh/tempfile :prefix "penpot.render." + :suffix ".png")] + (.then (fsp/writeFile path content) + (fn [_] + ((:on-object task) + {:path path + :filename "rendered.png"}))))) + (:exports task))] + (-> (js/Promise.all (clj->js writes)) + (.then (fn [_] (resolve nil)) reject))))))) + (fn [] (set! renderer/render original))))) + +(defn- stub-render-fail + [] + (let [original renderer/render] + (set! renderer/render + (fn [_cfg & _task] + (js/Promise.reject (ex-info "render boom" {})))) + (fn [] (set! renderer/render original)))) + +;; ---- THE FIXTURE + +(defn- export-params + [] + {:exports [{:file-id (str (uuid/next)) + :page-id (str (uuid/next)) + :object-id (str (uuid/next)) + :type "png" + :name "test shape" + :suffix "" + :scale 1}] + :name "the export"}) + +(defn- answer-with + [sid stages] + (fn [method body] + (when (= method "report-job-progress") + (vswap! stages conj (get-in body [:progress :stage]))) + (case method + "create-job-session" {:session-id sid :session-token "session-token"} + {:action :run}))) + +(t/deftest ^:async run-exports-one-shape-to-a-multipart-settle + (let [job-id (uuid/next) + sid (uuid/next) + stages (volatile! []) + restore-fetch (fake-fetch + (fn [method body] + (when (= method "report-job-progress") + (vswap! stages conj (get-in body [:progress :stage]))) + (case method + "create-job-session" {:session-id sid :session-token "session-token"} + {:action :run}))) + restore-render (stub-render "rendered!")] + (try + (await (consumer/run-export {} {:job-id job-id} (export-params))) + (t/testing "the run went by first breath, session, beats, settle" + (t/is (= ["report-job-progress" "create-job-session" + "report-job-progress" "report-job-progress" + "complete-job"] + (steps-of)))) + (t/testing "the settle carried the real artifact, named and typed" + (let [fd (call-of "complete-job")] + (t/is (instance? http/FormData fd)) + (t/is (= (str job-id) (.get fd "job-id"))) + (t/is (= (str sid) (.get fd "session-id"))) + (t/is (= "image/png" (.get fd "mtype"))))) + (t/testing "the beats painted the vocabulary in order" + (t/is (= [:preparing :rendering :packaging] @stages))) + (t/testing "the local registry entry is gone with the settle" + (t/is (false? (jobs/cancelled? job-id)))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore-fetch) + (restore-render))))) + +(t/deftest ^:async run-hands-the-renderer-uuids-not-claim-strings + (let [restore-fetch (fake-fetch (answer-with (uuid/next) (volatile! []))) + captured (atom nil) + restore-render (stub-render "rendered!" captured) + job-id (uuid/next) + share-id (uuid/next)] + (try + (await (consumer/run-export + {} + {:job-id job-id} + {:exports [{:file-id (str (uuid/next)) + :page-id (str (uuid/next)) + :object-id (str (uuid/next)) + :share-id (str share-id) + :type "png" + :name "test shape" + :suffix "" + :scale 1}] + :name "the export"})) + (let [export (-> @captured :exports first)] + (t/testing "every id arrived as a uuid object" + (t/is (uuid? (:file-id export))) + (t/is (uuid? (:page-id export))) + (t/is (uuid? (:share-id export))) + (t/is (= share-id (:share-id export))) + (t/is (uuid? (-> export :objects first :id)))) + (t/testing "the job-id traveled with the render" + (t/is (= job-id (:job-id export)))) + (t/testing "the task carried both injections" + (t/is (fn? (:on-object @captured))) + (t/is (fn? (:check-cancelled @captured))))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore-fetch) + (restore-render))))) + +(t/deftest ^:async run-packs-two-shapes-as-zip + (let [restore-fetch (fake-fetch (answer-with (uuid/next) (volatile! []))) + restore-render (stub-render "rendered!") + job-id (uuid/next)] + (try + (await (consumer/run-export + {} + {:job-id job-id} + {:exports [{:file-id (str (uuid/next)) + :page-id (str (uuid/next)) + :object-id (str (uuid/next)) + :type "png" + :name "one" + :suffix "" + :scale 1} + {:file-id (str (uuid/next)) + :page-id (str (uuid/next)) + :object-id (str (uuid/next)) + :type "png" + :name "two" + :suffix "" + :scale 1}] + :name "the export"})) + (t/testing "the artifact is the zip of both objects" + (let [fd (call-of "complete-job")] + (t/is (instance? http/FormData fd)) + (t/is (= "application/zip" (.get fd "mtype"))))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore-fetch) + (restore-render))))) + +(t/deftest ^:async a-failed-run-settles-fail-job-and-stops-beating + (let [restore-fetch (fake-fetch (answer-with (uuid/next) (volatile! []))) + restore-render (stub-render-fail) + job-id (uuid/next)] + (try + (await (consumer/run-export {} {:job-id job-id} (export-params))) + (await (js/Promise. (fn [resolve] (js/setTimeout resolve 1400)))) + (t/testing "the settle was a fail, not a complete" + (t/is (= "fail-job" (last (steps-of))))) + (t/testing "the failure is the shape the backend stores" + (let [fail (call-of "fail-job")] + (t/is (= :internal (get-in fail [:error :type]))) + (t/is (= :export-failed (get-in fail [:error :code]))) + (t/is (= "render boom" (get-in fail [:error :hint]))))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore-fetch) + (restore-render))))) + +(t/deftest ^:async cancel-before-the-render-settles-cancelled + (let [restore-fetch (fake-fetch + (fn [method _body] + (case method + "create-job-session" {:session-id (uuid/next) + :session-token "session-token"} + "report-job-progress" {:action :skip} + {:action :run}))) + restore-render (stub-render "should never render") + job-id (uuid/next)] + (try + (await (consumer/run-export {} {:job-id job-id} (export-params))) + (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]))))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore-fetch) + (restore-render))))) diff --git a/exporter/test/exporter_tests/consumer_worker_test.cljs b/exporter/test/exporter_tests/consumer_worker_test.cljs new file mode 100644 index 0000000000..896b74aa57 --- /dev/null +++ b/exporter/test/exporter_tests/consumer_worker_test.cljs @@ -0,0 +1,173 @@ +;; 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 exporter-tests.consumer-worker-test + "The queue loop of the new tree: the payload it reads, the claim it + makes, and the handoff to the runner, over a fake redis connection + and a fake management http." + (:require + [app.common.transit :as transit] + [app.common.uuid :as uuid] + [app.consumer.api :as api] + [cljs.test :as t :include-macros true] + [cuerdas.core :as cstr] + [exporter.consumer :as consumer] + [exporter.consumer.worker :as worker])) + +;; ---- THE MANAGEMENT FAKE + +(def ^:private calls + "One entry per management call: [method, body]." + (atom [])) + +(defn- fake-fetch + [respond] + (reset! calls []) + (let [original api/fetch] + (set! api/fetch + (fn [uri request] + (let [method (last (cstr/split (str uri) "/")) + body (transit/decode-str (.-body request))] + (swap! calls conj [method body]) + (js/Promise.resolve #js {:status 200 + :text (constantly + (js/Promise.resolve + (transit/encode-str + (respond method body))))})))) + (fn [] (set! api/fetch original)))) + +;; ---- THE QUEUE FAKE + +(defn- fake-conn + [queue] + #js {:blpop (fn [_key _timeout-s] + (if-let [payload (first @queue)] + (do + (swap! queue rest) + (js/Promise.resolve #js ["the.queue.key" payload])) + (js/Promise.resolve nil))) + :quit (fn [] (js/Promise.resolve nil))}) + +(defn- encoded-claim + [job-id scheduled-at] + (js/JSON.stringify #js [(str job-id) scheduled-at])) + +(defn- test-cfg + [extra] + (merge {:concurrency 1 + :queue-key "the.queue.key" + :render {}} + extra)) + +(t/deftest read-decodes-the-dispatcher-payload + (t/testing "a payload in shape names the job and when it was pushed" + (let [job-id (uuid/next)] + (t/is (= {:job-id (str job-id) + :scheduled-at "2026-10-07T09:00:00Z"} + (worker/read-payload (encoded-claim job-id + "2026-10-07T09:00:00Z")))))) + (t/testing "a corrupt payload is dropped, not thrown at the loop" + (t/is (nil? (worker/read-payload "{not a payload"))))) + +(t/deftest ^:async poll-once-awaits-what-it-claims + (try + (let [queue (atom [(encoded-claim (uuid/next) "2026-10-07T09:00:00Z")]) + conn (fake-conn queue) + handled (atom [])] + (await (worker/poll-once conn + "the.queue.key" + (fn [claim] + (swap! handled conj claim) + (js/Promise.resolve nil)))) + (t/is (= 1 (count @handled)))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))))) + +(t/deftest ^:async poll-once-drops-empty-pops-and-corrupt-payloads + (try + (let [queue (atom ["{corrupt"]) + conn (fake-conn queue)] + (await (worker/poll-once conn + "the.queue.key" + (fn [_claim] + (t/is false "corrupt must not be handled")))) + (await (worker/poll-once conn + "the.queue.key" + (fn [_claim] + (t/is false "a corrupt pop has nothing to handle")))) + (t/is (empty? @queue))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))))) + +(t/deftest ^:async process-drops-a-claim-the-worker-never-won + (let [restore-fetch (fake-fetch + (fn [method _body] + (case method + "claim-job" {:action :skip :status "cancelled"} + {:action :run})))] + (try + (await (worker/process (test-cfg {}) + {:job-id (uuid/next) + :scheduled-at "2026-10-07T09:00:00Z"})) + (t/testing "only the claim traveled: no run, no render" + (t/is (= ["claim-job"] (mapv first @calls)))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore-fetch))))) + +(t/deftest ^:async process-hands-the-claim-to-the-runner + (let [restore-fetch (fake-fetch + (fn [method _body] + (case method + "claim-job" {:action :run + :name "export-assets" + :params {:exports [] :name "x"}} + {:action :run}))) + ran (atom nil) + restore-run (let [original consumer/run-export] + (set! consumer/run-export + (fn [cfg claim params] + (reset! ran [cfg claim params]) + (js/Promise.resolve nil))) + (fn [] (set! consumer/run-export original))) + render {:some "render-cfg"}] + (try + (let [job-id (uuid/next)] + (await (worker/process (test-cfg {:render render}) + {:job-id (str job-id) + :scheduled-at "2026-10-07T09:00:00Z"})) + (t/testing "the runner got the render view, the job and the params" + (t/is (= render (nth @ran 0))) + (t/is (= {:job-id (str job-id)} (nth @ran 1))) + (t/is (= {:exports [] :name "x"} (nth @ran 2))))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore-fetch) + (restore-run))))) + +(t/deftest ^:async start-opens-one-connection-per-poller-and-stop-quits-them + (try + (let [opened (atom []) + quit (atom []) + conn (fn [] + (let [c (fake-conn (atom []))] + (swap! opened conj c) + (set! (.-quit c) (fn [] (swap! quit conj c) (js/Promise.resolve nil))) + ;; the first pop never answers: the pollers sit in + ;; it, so the test never races their loop + (set! (.-blpop c) (fn [_k _t] (js/Promise. (fn [_res _rej])))) + c)) + handle (await (worker/start (test-cfg {:concurrency 2 :connect conn})))] + (t/is (= 2 (count @opened))) + (await (worker/stop handle)) + (t/is (= 2 (count @quit))) + (t/testing "a second stop is a no-op" + (await (worker/stop handle)) + (t/is (= 2 (count @quit))))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))))) diff --git a/exporter/test/exporter_tests/exporter_consumer_plan_test.cljs b/exporter/test/exporter_tests/exporter_consumer_plan_test.cljs new file mode 100644 index 0000000000..22568fbf45 --- /dev/null +++ b/exporter/test/exporter_tests/exporter_consumer_plan_test.cljs @@ -0,0 +1,44 @@ +;; 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 exporter-tests.exporter-consumer-plan-test + "The render plan of the new tree: same chunks and names as the + legacy planner, dispatching on the wasm flag." + (:require + [app.common.uuid :as uuid] + [cljs.test :as t :include-macros true] + [exporter.consumer.plan :as plan])) + +(defn- exports + [n type scale] + (let [file-id (uuid/next) + page-id (uuid/next)] + (mapv (fn [i] + {:file-id file-id + :page-id page-id + :object-id (uuid/next) + :name (str "shape-" i) + :suffix "" + :scale scale + :type type}) + (range n)))) + +(t/deftest browser-exports-are-chunked + (let [parts (plan/prepare-exports (exports 120 :png 1) "token" false)] + (t/is (= 3 (count parts))) + (t/is (= [50 50 20] (mapv (comp count :objects) parts))))) + +(t/deftest wasm-exports-render-as-one-group + (let [parts (plan/prepare-exports (exports 120 :png 1) "token" true)] + (t/is (= 1 (count parts))) + (t/is (= 120 (count (:objects (first parts))))))) + +(t/deftest duplicate-names-get-suffixed + (let [items [(assoc ((exports 1 :png 1) 0) :name "same") + (assoc ((exports 1 :png 1) 0) :name "same")] + parts (plan/prepare-exports items "token" false)] + (t/is (= ["same.png" "same-2.png"] + (mapv :filename (:objects (first parts))))))) diff --git a/exporter/test/exporter_tests/exporter_jobs_test.cljs b/exporter/test/exporter_tests/exporter_jobs_test.cljs new file mode 100644 index 0000000000..9d45c8f876 --- /dev/null +++ b/exporter/test/exporter_tests/exporter_jobs_test.cljs @@ -0,0 +1,41 @@ +;; 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 exporter-tests.exporter-jobs-test + "The local cancel arm of the new tree: one flag per claimed job, + and the zero-arg check the renderers run." + (:require + [app.common.uuid :as uuid] + [cljs.test :as t :include-macros true] + [exporter.jobs :as jobs])) + +(t/deftest register-mark-and-release + (let [job-id (uuid/next)] + (t/testing "a fresh job is not cancelled" + (jobs/register job-id) + (t/is (false? (jobs/cancelled? job-id)))) + (t/testing "the first mark flips the flag and reports it" + (t/is (true? (jobs/mark-cancelled job-id))) + (t/is (true? (jobs/cancelled? job-id)))) + (t/testing "a released job is gone, not cancelled" + (jobs/release job-id) + (t/is (false? (jobs/cancelled? job-id))) + (t/is (nil? (jobs/mark-cancelled job-id)))))) + +(t/deftest check-cancelled-raises-the-cooperative-code + (let [job-id (uuid/next) + check (jobs/check-cancelled job-id)] + (jobs/register job-id) + (t/testing "a live job passes the check silently" + (t/is (nil? (check)))) + (t/testing "a marked job raises what the runner settles as cancel" + (jobs/mark-cancelled job-id) + (try + (check) + (t/is false "the check should have raised") + (catch :default cause + (t/is (= :job-cancelled (-> cause ex-data :code)))))) + (jobs/release job-id))) diff --git a/exporter/test/exporter_tests/main_test.cljs b/exporter/test/exporter_tests/main_test.cljs index 3bf973dfa6..c225962410 100644 --- a/exporter/test/exporter_tests/main_test.cljs +++ b/exporter/test/exporter_tests/main_test.cljs @@ -13,7 +13,8 @@ [app.consumer.config :as ccfg] [cljs.test :as t :include-macros true] [exporter.browser :as browser] - [exporter.main :as main])) + [exporter.main :as main] + [exporter.utils.system :as system])) (t/deftest new-tree-loads (t/testing "exporter.main resolves once required" @@ -89,7 +90,15 @@ (t/is (= {:max (ccfg/concurrency)} (:exporter.browser/pool main/system-config))) (t/is (= {:max (ccfg/concurrency)} - (:exporter.wasm.pool/pool main/system-config))))) + (:exporter.wasm.pool/pool main/system-config)))) + (t/testing "the queue consumer renders through a view of the running pools" + (let [worker (:exporter.consumer/worker main/system-config)] + (t/is (= (ccfg/concurrency) (:concurrency worker))) + (t/is (= (ccfg/queue-key) (:queue-key worker))) + (t/is (= {:exporter.browser/pool (system/ref :exporter.browser/pool) + :exporter.wasm.pool/pool (system/ref :exporter.wasm.pool/pool)} + (select-keys (:render worker) + [:exporter.browser/pool :exporter.wasm.pool/pool])))))) ;; NOTE: no test boots the production wiring: the wasm pool warms one ;; worker eagerly (min 1), and in this process the worker script would diff --git a/exporter/test/exporter_tests/runner.cljs b/exporter/test/exporter_tests/runner.cljs index b5f200a129..de67ff0ce9 100644 --- a/exporter/test/exporter_tests/runner.cljs +++ b/exporter/test/exporter_tests/runner.cljs @@ -15,6 +15,10 @@ [exporter-tests.browser-test] [exporter-tests.consumer-config-test] [exporter-tests.consumer-plan-test] + [exporter-tests.consumer-test] + [exporter-tests.consumer-worker-test] + [exporter-tests.exporter-consumer-plan-test] + [exporter-tests.exporter-jobs-test] [exporter-tests.jobs-test] [exporter-tests.main-test] [exporter-tests.renderer-browser-test] @@ -42,6 +46,10 @@ ['exporter-tests.browser-scope-test 'exporter-tests.browser-test 'exporter-tests.consumer-plan-test + 'exporter-tests.consumer-test + 'exporter-tests.consumer-worker-test + 'exporter-tests.exporter-consumer-plan-test + 'exporter-tests.exporter-jobs-test 'exporter-tests.jobs-test 'exporter-tests.main-test 'exporter-tests.api-test