diff --git a/.serena/memories/exporter/core.md b/.serena/memories/exporter/core.md index 1ef13cfb62..afe02cf027 100644 --- a/.serena/memories/exporter/core.md +++ b/.serena/memories/exporter/core.md @@ -50,6 +50,8 @@ on this side: the row of the job, its cancel and its result are the backend's. - `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.api` is the backend management client (claim/progress/session/settle, transit bodies, multipart artifact), a faithful `^:async` port with no promesa chains and the same `fetch` def seam for hermetic tests. +- `exporter.consumer.config` holds the consumer knobs (`concurrency` floored at 1, `queue-key`, `queue-name`): the runner, the worker and `exporter.main` read it, so no namespace under `exporter.*` depends on `app.consumer.*` anymore. - `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. diff --git a/exporter/src/exporter/consumer.cljs b/exporter/src/exporter/consumer.cljs index 6351bc6823..7bc8b03829 100644 --- a/exporter/src/exporter/consumer.cljs +++ b/exporter/src/exporter/consumer.cljs @@ -33,11 +33,11 @@ [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.api :as api] [exporter.consumer.plan :as plan] [exporter.jobs :as jobs] [exporter.renderer :as renderer])) diff --git a/exporter/src/exporter/consumer/api.cljs b/exporter/src/exporter/consumer/api.cljs new file mode 100644 index 0000000000..2507f8d85b --- /dev/null +++ b/exporter/src/exporter/consumer/api.cljs @@ -0,0 +1,152 @@ +;; 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.api + "The backend management API client. + + The only place of the worker that speaks HTTP: every method of the + backend ledger (claim, progress, session, settle) travels through + here, with the shared key the exporter was configured with and, when + a call renders as a user, the Bearer token of its session. Request + bodies travel as transit+json (except the artifact form of + `complete-job`, which is multipart, like every other upload) and the + answers come back as transit too. There are no promesa chains in + this namespace." + (:require + ["node:fs/promises" :as fsp] + ["undici" :as http] + [app.common.exceptions :as ex] + [app.common.transit :as t] + [app.common.uri :as u] + [app.config :as cf] + [cljs.core :as c])) + +(def fetch + "The network entry point of the client: a plain def, redefined by the + tests of the client, which never need a network." + http/fetch) + +;; ---- THE REQUEST + +(defn- dispatcher + "The agent of one request: a self-signed certificate happens in + deployments the same way the existing uploads already accept." + [] + (new http/Agent #js {:connect #js {:rejectUnauthorized false}})) + +(defn- method-uri + [method] + (-> (cf/get-internal-uri) + (u/ensure-path-slash) + (u/join (str "api/management/methods/" (c/name method))) + (str))) + +(defn- shared-key + "What every management call carries: the key id of this exporter plus + the key, and the wish to read transit back." + [] + #js {"X-Shared-Key" (str "exporter " cf/management-key) + "Accept" "application/transit+json"}) + +(defn- ^:async parse-response + "The answer of a management call: transit when it lands. A non-200 is + the failure of the call, with the error body the backend answered; + a body that does not decode stays as the plain text it came as." + [response] + (let [status (.-status response) + text (await (.text response)) + body (let [decoded (ex/try! (t/decode-str text))] + (if-not (ex/exception? decoded) decoded text))] + (if (= status 200) + body + (throw (ex/error :type :internal + :code :failed-management-request + :status status + :response body))))) + +(defn- ^:async request + "POST `params` to a management method as transit+json. The content + type travels on every call: without it the backend never parses the + body as transit and validates an empty map." + [method params] + (let [response (await (fetch (method-uri method) + #js {:method "POST" + :headers (doto (shared-key) + (unchecked-set "Content-Type" "application/transit+json")) + :body (t/encode-str params) + :dispatcher (dispatcher)}))] + (await (parse-response response)))) + +(defn- ^:async request-multipart + "POST a FormData body to a management method. The content type is + settled by the FormData itself (boundary included), so only the + shared key goes in the headers." + [method form-data] + (let [response (await (fetch (method-uri method) + #js {:method "POST" + :headers (shared-key) + :body form-data + :dispatcher (dispatcher)}))] + (await (parse-response response)))) + +;; ---- THE METHODS + +(defn ^:async claim-job + "Request a queued job from the backend. The answer is either + `{:action :run :name ... :params ...}` (this poller owns the job) or + `{:action :skip ...}` (something else won it, or it is gone)." + [job-id scheduled-at] + (await (request :claim-job {:job-id job-id :scheduled-at scheduled-at}))) + +(defn ^:async report-job-progress + "The heartbeat of a running job: the stage hitos of the progress + contract plus the counters of the current work. A `:skip` says the + job can no longer hear the worker (cancelled, aborted, settled)." + [job-id progress] + (await (request :report-job-progress {:job-id job-id :progress progress}))) + +(defn ^:async create-job-session + "Mint an authentication session of the owner of a running job, so the + render views authenticate as the user, not as a machine." + [job-id] + (await (request :create-job-session {:job-id job-id}))) + +(defn ^:async complete-job + "Close a job whose result is the plain value `result`, with an + optional `:resource-id` this worker already owns (the internal path) + and an optional `:session-id` to close." + ([job-id result] + (complete-job job-id result nil)) + ([job-id result {:keys [resource-id session-id]}] + (await (request :complete-job + (cond-> {:job-id job-id :result result} + (some? resource-id) (assoc :resource-id resource-id) + (some? session-id) (assoc :session-id session-id)))))) + +(defn ^:async complete-job-with-artifact + "Close a job whose result is the artifact itself: the backend stores + the bytes and derives the result descriptor. `artifact` says what it + is (`:path`, `:filename`, `:mtype`); `ctx` names the job and, when + the worker rendered with a session, the one to close." + [{:keys [job-id session-id]} {:keys [path filename mtype]}] + (let [buffer (await (fsp/readFile (str path))) + blob (new js/Blob #js [buffer] #js {:type mtype}) + fdata (new http/FormData)] + (.append fdata "job-id" (str job-id)) + (.append fdata "content" blob filename) + (.append fdata "filename" filename) + (.append fdata "mtype" (str mtype)) + (when session-id + (.append fdata "session-id" (str session-id))) + (await (request-multipart :complete-job fdata)))) + +(defn ^:async fail-job + "Close a job as failed with the rich error the backend stores for + `job.error`. `:session-id` closes the render session the same way." + [job-id error & {:keys [session-id]}] + (await (request :fail-job + (cond-> {:job-id job-id :error error} + (some? session-id) (assoc :session-id session-id))))) diff --git a/exporter/src/exporter/consumer/config.cljs b/exporter/src/exporter/consumer/config.cljs new file mode 100644 index 0000000000..8d284e0c20 --- /dev/null +++ b/exporter/src/exporter/consumer/config.cljs @@ -0,0 +1,38 @@ +;; 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.config + "The knobs of the consumer: how many jobs run at once and which + queue they come from. Read live, so a long-lived process obeys a + configuration change on the next poll." + (:require + [app.config :as cf] + [cljs.core :as c])) + +(def queue-name + "The queue this worker consumes: the one the `:export-assets` job-def + routes its jobs to through `::jobs/queue-name`." + "exporter") + +(defn concurrency + "K pollers: each one owns its Redis connection, and the blocking + connection is the semaphore of the slot, so K is the whole concurrency + of the process. At least one: a worker role with zero pollers is a + misconfiguration, not a rest. + + Takes the configured value as an argument, so the tests do not + journey through the global config; the zero-arity reads it live." + ([] + (concurrency (c/get cf/config :exporter-worker-concurrency))) + ([value] + (max 1 (if (number? value) value 2)))) + +(defn queue-key + "The Redis list the dispatcher pushes this queue's payloads through: + `penpot.worker.queue::`." + ([] (queue-key (cf/get :tenant))) + ([tenant] + (str "penpot.worker.queue:" tenant ":" queue-name))) diff --git a/exporter/src/exporter/consumer/worker.cljs b/exporter/src/exporter/consumer/worker.cljs index 087979a966..20879acbb4 100644 --- a/exporter/src/exporter/consumer/worker.cljs +++ b/exporter/src/exporter/consumer/worker.cljs @@ -32,10 +32,10 @@ ["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.consumer.api :as api] + [exporter.consumer.config :as ccfg] [exporter.utils.system :as system])) (def ^:private poll-timeout-s 5) diff --git a/exporter/src/exporter/main.cljs b/exporter/src/exporter/main.cljs index 38d363b001..edeb14e6dc 100644 --- a/exporter/src/exporter/main.cljs +++ b/exporter/src/exporter/main.cljs @@ -17,10 +17,10 @@ ["node:worker_threads" :as wt] [app.common.logging :as l] [app.config :as cf] - [app.consumer.config :as ccfg] ;; Loaded for their init-key/halt-key methods, which is what makes ;; the pools and the queue consumer part of the system below. [exporter.browser] + [exporter.consumer.config :as ccfg] [exporter.consumer.worker] [exporter.utils.system :as system] [exporter.wasm.pool])) diff --git a/exporter/test/exporter_tests/consumer_api_test.cljs b/exporter/test/exporter_tests/consumer_api_test.cljs new file mode 100644 index 0000000000..06ac8c85a8 --- /dev/null +++ b/exporter/test/exporter_tests/consumer_api_test.cljs @@ -0,0 +1,181 @@ +;; 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-api-test + "The management client of the new tree against a fake http: what it + sends (method uri, shared key, shape of the body) is the whole + contract this client owns." + (:require + ["node:fs/promises" :as fsp] + ["node:path" :as path] + ["undici" :as http] + [app.common.transit :as transit] + [app.common.uuid :as uuid] + [app.config :as cf] + [app.util.shell :as sh] + [cljs.test :as t :include-macros true] + [cuerdas.core :as str] + [exporter.consumer.api :as api])) + +(def ^:private original-fetch api/fetch) + +(def ^:private calls + "Every fetch the client made, in order." + (atom [])) + +(defn- fake-fetch + [uri request] + (swap! calls conj [uri request]) + (js/Promise.resolve #js {:status 200 + :text (constantly + (js/Promise.resolve + (transit/encode-str {:action :run})))})) + +(defn- failure-fetch + "A fake that answers like a management endpoint that refuses the key." + [_uri _request] + (js/Promise.resolve #js {:status 403 + :text (constantly (js/Promise.resolve "{}"))})) + +(defn- install-fetch + ([] (install-fetch fake-fetch)) + ([the-fake] + (set! api/fetch the-fake) + (reset! calls []) + (fn [] (set! api/fetch original-fetch)))) + +(defn- last-request + [] + (second (last @calls))) + +(defn- fetch-header + "One header of the last request, exactly as it would travel." + [name] + (unchecked-get (.-headers (second (last @calls))) name)) + +(defn- body-value + "One field of the last FormData request." + [name] + (.get (.-body (last-request)) name)) + +(defn- decode-body + "The body of the last transit request, decoded." + [] + (transit/decode-str (.-body (last-request)))) + +(defn- last-uri + [] + (first (last @calls))) + +(t/deftest ^:async claim-job-posts-the-claim-of-one-poller + (let [restore (install-fetch)] + (try + (let [job-id (uuid/next)] + (await (api/claim-job job-id "2026-10-07T00:00:00Z")) + (t/is (str/ends-with? (last-uri) "/api/management/methods/claim-job")) + (t/is (= (str "exporter " cf/management-key) (fetch-header "X-Shared-Key"))) + (t/is (= {:job-id job-id :scheduled-at "2026-10-07T00:00:00Z"} + (decode-body)))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore))))) + +(t/deftest ^:async json-calls-declare-the-transit-content-type + ;; without it the backend never parses the body as transit and + ;; validates an empty map (every key a `missing-key`) + (let [restore (install-fetch)] + (try + (await (api/claim-job (uuid/next) "2026-10-07T00:00:00Z")) + (t/is (= "application/transit+json" + (fetch-header "Content-Type"))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore))))) + +(t/deftest ^:async report-job-progress-posts-the-beat + (let [restore (install-fetch)] + (try + (let [job-id (uuid/next) + progress {:stage :rendering + :counters {:objects {:current 1 :total 2}}}] + (await (api/report-job-progress job-id progress)) + (t/is (str/ends-with? (last-uri) "/api/management/methods/report-job-progress")) + (t/is (= {:job-id job-id :progress progress} (decode-body)))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore))))) + +(t/deftest ^:async fail-job-carries-the-error-and-the-session + (let [restore (install-fetch)] + (try + (let [job-id (uuid/next) + sid (uuid/next) + error {:type :internal :code :processing-error :hint "boom"}] + (await (api/fail-job job-id error)) + (await (api/fail-job job-id error :session-id sid)) + (t/is (= {:job-id job-id :error error :session-id sid} + (decode-body)))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore))))) + +(t/deftest ^:async complete-job-posts-the-plain-result + (let [restore (install-fetch)] + (try + (let [job-id (uuid/next) + sid (uuid/next)] + (await (api/complete-job job-id {:total 1})) + (await (api/complete-job job-id {:total 1} {:session-id sid})) + (t/is (= {:job-id job-id :result {:total 1} :session-id sid} + (decode-body)))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore))))) + +(t/deftest ^:async complete-job-with-artifact-rides-multipart + (let [restore (install-fetch)] + (try + (let [job-id (uuid/next) + sid (uuid/next) + tmpfile (path/join sh/tmpdir (str "test-artifact." (uuid/next)))] + (await (fsp/writeFile tmpfile "the export bytes")) + (await (api/complete-job-with-artifact + {:job-id job-id :session-id sid} + {:path tmpfile + :filename "export.zip" + :mtype "application/zip"})) + (t/is (instance? http/FormData (.-body (last-request)))) + (t/is (= (str job-id) (body-value "job-id"))) + (t/is (= "export.zip" (body-value "filename"))) + (t/is (= "application/zip" (body-value "mtype"))) + (t/is (= (str sid) (body-value "session-id"))) + (t/is (= (str "exporter " cf/management-key) + (fetch-header "X-Shared-Key")))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore))))) + +(t/deftest ^:async a-failed-management-answer-is-the-failure-of-the-call + (let [restore (install-fetch failure-fetch)] + (try + (try + (await (api/claim-job (uuid/next) "2026-10-07T00:00:00Z")) + (t/is false "the call should have rejected on a 403") + (catch :default cause + (let [data (ex-data cause)] + (t/is (= :internal (:type data))) + (t/is (= :failed-management-request (:code data))) + (t/is (= 403 (:status data)))))) + (catch :default cause + (t/is false (str "unexpected failure: " (ex-message cause)))) + (finally + (restore))))) diff --git a/exporter/test/exporter_tests/consumer_test.cljs b/exporter/test/exporter_tests/consumer_test.cljs index 1b03e0f2ba..7502b41f12 100644 --- a/exporter/test/exporter_tests/consumer_test.cljs +++ b/exporter/test/exporter_tests/consumer_test.cljs @@ -14,11 +14,11 @@ ["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.consumer.api :as api] [exporter.jobs :as jobs] [exporter.renderer :as renderer])) diff --git a/exporter/test/exporter_tests/consumer_worker_test.cljs b/exporter/test/exporter_tests/consumer_worker_test.cljs index 896b74aa57..af2c4b08a8 100644 --- a/exporter/test/exporter_tests/consumer_worker_test.cljs +++ b/exporter/test/exporter_tests/consumer_worker_test.cljs @@ -11,10 +11,10 @@ (: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.api :as api] [exporter.consumer.worker :as worker])) ;; ---- THE MANAGEMENT FAKE diff --git a/exporter/test/exporter_tests/exporter_consumer_config_test.cljs b/exporter/test/exporter_tests/exporter_consumer_config_test.cljs new file mode 100644 index 0000000000..9b7abed7bb --- /dev/null +++ b/exporter/test/exporter_tests/exporter_consumer_config_test.cljs @@ -0,0 +1,26 @@ +;; 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-config-test + "The consumer knobs of the new tree: how many jobs run at once and + which queue they come from." + (:require + [cljs.test :as t :include-macros true] + [exporter.consumer.config :as ccfg])) + +(t/deftest concurrency-is-at-least-one + (t/testing "the configured value is taken as given" + (t/is (= 2 (ccfg/concurrency 2))) + (t/is (= 4 (ccfg/concurrency 4)))) + (t/testing "zero or less is a misconfiguration, floored to one" + (t/is (= 1 (ccfg/concurrency 0))) + (t/is (= 1 (ccfg/concurrency -3)))) + (t/testing "an unset value falls back to the default K" + (t/is (= 2 (ccfg/concurrency nil))))) + +(t/deftest queue-key-names-tenant-and-queue + (t/is (= "penpot.worker.queue:main:exporter" + (ccfg/queue-key "main")))) diff --git a/exporter/test/exporter_tests/main_test.cljs b/exporter/test/exporter_tests/main_test.cljs index c225962410..d1d36dffdd 100644 --- a/exporter/test/exporter_tests/main_test.cljs +++ b/exporter/test/exporter_tests/main_test.cljs @@ -10,9 +10,9 @@ (:require ["node:process" :as proc] [app.common.logging :as l] - [app.consumer.config :as ccfg] [cljs.test :as t :include-macros true] [exporter.browser :as browser] + [exporter.consumer.config :as ccfg] [exporter.main :as main] [exporter.utils.system :as system])) diff --git a/exporter/test/exporter_tests/runner.cljs b/exporter/test/exporter_tests/runner.cljs index de67ff0ce9..a2d29a8773 100644 --- a/exporter/test/exporter_tests/runner.cljs +++ b/exporter/test/exporter_tests/runner.cljs @@ -13,10 +13,12 @@ [exporter-tests.api-test] [exporter-tests.browser-scope-test] [exporter-tests.browser-test] + [exporter-tests.consumer-api-test] [exporter-tests.consumer-config-test] [exporter-tests.consumer-plan-test] [exporter-tests.consumer-test] [exporter-tests.consumer-worker-test] + [exporter-tests.exporter-consumer-config-test] [exporter-tests.exporter-consumer-plan-test] [exporter-tests.exporter-jobs-test] [exporter-tests.jobs-test] @@ -45,9 +47,11 @@ (def test-namespaces ['exporter-tests.browser-scope-test 'exporter-tests.browser-test + 'exporter-tests.consumer-api-test 'exporter-tests.consumer-plan-test 'exporter-tests.consumer-test 'exporter-tests.consumer-worker-test + 'exporter-tests.exporter-consumer-config-test 'exporter-tests.exporter-consumer-plan-test 'exporter-tests.exporter-jobs-test 'exporter-tests.jobs-test