diff --git a/exporter/src/app/config.cljs b/exporter/src/app/config.cljs index 0fcc5a052d..48b73f3115 100644 --- a/exporter/src/app/config.cljs +++ b/exporter/src/app/config.cljs @@ -33,6 +33,10 @@ :exporter-max-jobs-per-profile 2 :exporter-queue-max 64 :exporter-job-ttl 3600 + ;; the roles this instance serves: by default only the http service; + ;; `worker` turns the process into a consumer of the backend jobs + :exporter-roles "http" + :exporter-worker-concurrency 2 :wasm-worker-pool-max 2 :wasm-worker-pool-min 1 :wasm-worker-idle-timeout 300 @@ -55,6 +59,8 @@ [:exporter-max-jobs-per-profile {:optional true} ::sm/int] [:exporter-queue-max {:optional true} ::sm/int] [:exporter-job-ttl {:optional true} ::sm/int] + [:exporter-roles {:optional true} :string] + [:exporter-worker-concurrency {:optional true} ::sm/int] [:wasm-worker-pool-max {:optional true} ::sm/int] [:wasm-worker-pool-min {:optional true} ::sm/int] [:wasm-worker-idle-timeout {:optional true} ::sm/int] diff --git a/exporter/src/app/consumer/api.cljs b/exporter/src/app/consumer/api.cljs new file mode 100644 index 0000000000..2c0f6ea654 --- /dev/null +++ b/exporter/src/app/consumer/api.cljs @@ -0,0 +1,153 @@ +;; 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 app.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." + (: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] + [promesa.core :as p])) + +(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"}) + +(def ^:private 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." + (fn [response] + (let [status (.-status response)] + (->> (.text response) + (p/fmap (fn [text] + (let [body (let [decoded (ex/try! (t/decode-str text))] + (if-not (ex/exception? decoded) decoded text))] + (if (= status 200) + body + (ex/raise :type :internal + :code :failed-management-request + :status status + :response body))))))))) + +(defn- request + "POST `params` to a management method as transit+json." + [method params] + (->> (fetch (method-uri method) + #js {:method "POST" + :headers (shared-key) + :body (t/encode-str params) + :dispatcher (dispatcher)}) + (p/mcat parse-response))) + +(defn- 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] + (->> (fetch (method-uri method) + #js {:method "POST" + :headers (shared-key) + :body form-data + :dispatcher (dispatcher)}) + (p/mcat parse-response))) + +;; ---- THE METHODS + +(defn 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] + (request :claim-job {:job-id job-id :scheduled-at scheduled-at})) + +(defn 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] + (request :report-job-progress {:job-id job-id :progress progress})) + +(defn 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] + (request :create-job-session {:job-id job-id})) + +(defn 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]}] + (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 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]}] + (->> (fsp/readFile (str path)) + (p/fmap (fn [buffer] + (let [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))) + fdata))) + (p/mcat (fn [fdata] + (request-multipart :complete-job fdata))))) + +(defn 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]}] + (request :fail-job + (cond-> {:job-id job-id :error error} + (some? session-id) (assoc :session-id session-id)))) diff --git a/exporter/src/app/consumer/config.cljs b/exporter/src/app/consumer/config.cljs new file mode 100644 index 0000000000..02504c75f5 --- /dev/null +++ b/exporter/src/app/consumer/config.cljs @@ -0,0 +1,68 @@ +;; 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 app.consumer.config + "The worker role of the exporter configuration. + + The exporter process may serve its http surfaces, consume the + backend `:exporter` queue, or both: `PENPOT_EXPORTER_ROLES` names the + roles (`http` by default, `http,worker` during the transition). The + worker pieces never read these values at import time: each accessor + reads the live configuration, so a long-lived process can be + restarted with another role without code changes." + (:require + [app.config :as cf] + [cljs.core :as c] + [cuerdas.core :as str])) + +(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- scan-roles + [spec] + (->> (if (string? spec) + (str/split spec ",") + spec) + (map str/trim) + (remove str/blank?) + (map keyword) + (into #{}))) + +(defn roles + "The set of roles this instance serves, as keywords." + ([] (roles (cf/get :exporter-roles))) + ([spec] + (if (nil? spec) + #{:http} + (scan-roles spec)))) + +(defn worker-enabled? + "True when this instance also consumes the exporter queue." + ([] (contains? (roles) :worker)) + ([roles] + (contains? roles :worker))) + +(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/test/exporter_tests/api_test.cljs b/exporter/test/exporter_tests/api_test.cljs new file mode 100644 index 0000000000..b2953659ba --- /dev/null +++ b/exporter/test/exporter_tests/api_test.cljs @@ -0,0 +1,180 @@ +;; 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.api-test + "The management client of the worker 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.exceptions :as ex] + [app.common.transit :as transit] + [app.common.uuid :as uuid] + [app.config :as cf] + [app.consumer.api :as api] + [app.util.shell :as sh] + [cljs.core :as c] + [cljs.test :as t :include-macros true] + [cuerdas.core :as str] + [promesa.core :as p])) + +(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]) + (p/resolved #js {:status 200 + :text (constantly (p/resolved (transit/encode-str {:action :run})))})) + +(defn- failure-fetch + "A fake that answers like a management endpoint that refuses the key." + [_uri _request] + (p/resolved #js {:status 403 + :text (constantly (p/resolved "{}"))})) + +(defn- install-fetch! + "Put a fake in place and return the thunk that puts the original + back. The thunk belongs at the end of the promise chain's `p/do`, so + the fake covers the whole async run." + ([] (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))) + +(defn- with-restore-and-done + "Run `thunk`, then put the original fetch back, then say the test is + finished: the rest never waits on the assertions, so a broken chain + reports a failure, never a hang." + [thunk restore! done] + (p/do thunk (restore!) (done))) + +(t/deftest claim-job-posts-the-claim-of-one-poller + (t/async done + (let [job-id (uuid/next) + restore! (install-fetch!)] + (with-restore-and-done + (p/fmap (fn [_] + (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)))) + (api/claim-job job-id "2026-10-07T00:00:00Z")) + restore! + done)))) + +(t/deftest report-job-progress-posts-the-beat + (t/async done + (let [job-id (uuid/next) + progress {:stage :rendering + :counters {:objects {:current 1 :total 2}}} + restore! (install-fetch!)] + (with-restore-and-done + (p/fmap (fn [_] + (t/is (str/ends-with? (last-uri) "/api/management/methods/report-job-progress")) + (t/is (= {:job-id job-id :progress progress} (decode-body)))) + (api/report-job-progress job-id progress)) + restore! + done)))) + +(t/deftest fail-job-carries-the-error-and-the-session + (t/async done + (let [job-id (uuid/next) + sid (uuid/next) + error {:type :internal :code :processing-error :hint "boom"} + restore! (install-fetch!)] + (with-restore-and-done + (p/mcat (fn [_] + (p/fmap (fn [_] + (t/is (= {:job-id job-id :error error :session-id sid} + (decode-body)))) + (api/fail-job job-id error :session-id sid))) + (api/fail-job job-id error)) + restore! + done)))) + +(t/deftest complete-job-posts-the-plain-result + (t/async done + (let [job-id (uuid/next) + sid (uuid/next) + restore! (install-fetch!)] + (with-restore-and-done + (p/mcat (fn [_] + (p/fmap (fn [_] + (t/is (= {:job-id job-id :result {:total 1} :session-id sid} + (decode-body)))) + (api/complete-job job-id {:total 1} {:session-id sid}))) + (api/complete-job job-id {:total 1})) + restore! + done)))) + +(t/deftest complete-job-with-artifact-rides-multipart + (t/async done + (let [job-id (uuid/next) + sid (uuid/next) + tmpfile (path/join sh/tmpdir (str "test-artifact." (uuid/next))) + restore! (install-fetch!)] + (with-restore-and-done + (p/mcat (fn [_] + (p/fmap (fn [_] + (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")))) + (api/complete-job-with-artifact + {:job-id job-id :session-id sid} + {:path tmpfile + :filename "export.zip" + :mtype "application/zip"}))) + (->> (fsp/writeFile tmpfile "the export bytes") + (p/fmap (fn [_] nil)))) + restore! + done)))) + +(t/deftest a-failed-management-answer-is-the-failure-of-the-call + (t/async done + (let [restore! (install-fetch! failure-fetch)] + (with-restore-and-done + (p/catch (fn [error] + (let [data (ex-data error)] + (t/is (= :internal (:type data))) + (t/is (= :failed-management-request (:code data))) + (t/is (= 403 (:status data))))) + (api/claim-job (uuid/next) "2026-10-07T00:00:00Z")) + restore! + done)))) diff --git a/exporter/test/exporter_tests/consumer_config_test.cljs b/exporter/test/exporter_tests/consumer_config_test.cljs new file mode 100644 index 0000000000..e3f95b6476 --- /dev/null +++ b/exporter/test/exporter_tests/consumer_config_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.consumer-config-test + "The worker side of the config: the roles a process serves and how + many pollers a worker owns." + (:require + [app.consumer.config :as ccfg] + [cljs.test :as t :include-macros true] + [cuerdas.core :as str])) + +(t/deftest roles-parse-the-env-list + (t/testing "the default keeps the http surfaces only" + (t/is (= #{:http} (ccfg/roles "http")))) + (t/is (= #{:http :worker} (ccfg/roles "http,worker"))) + (t/is (= #{:http :worker} (ccfg/roles "http, worker"))) + (t/testing "the parser does not invent roles from noise" + (t/is (= #{} (ccfg/roles ""))) + (t/is (= #{:worker} (ccfg/roles ",worker"))))) + +(t/deftest worker-enabled-reads-the-roles + (t/is (true? (ccfg/worker-enabled? #{:http :worker}))) + (t/is (true? (ccfg/worker-enabled? #{:worker}))) + (t/is (false? (ccfg/worker-enabled? #{:http})))) + +(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/runner.cljs b/exporter/test/exporter_tests/runner.cljs index 82c3e4c46f..819ced15c7 100644 --- a/exporter/test/exporter_tests/runner.cljs +++ b/exporter/test/exporter_tests/runner.cljs @@ -10,6 +10,8 @@ [cljs.test :as t] [clojure.string :as str] [clojure.tools.cli :refer [parse-opts]] + [exporter-tests.api-test] + [exporter-tests.consumer-config-test] [exporter-tests.export-shapes-test] [exporter-tests.jobs-test] [exporter-tests.renderer-svg-test] @@ -23,6 +25,8 @@ (def test-namespaces ['exporter-tests.export-shapes-test 'exporter-tests.jobs-test + 'exporter-tests.api-test + 'exporter-tests.consumer-config-test 'exporter-tests.renderer-svg-test 'exporter-tests.scheduler-test 'exporter-tests.shell-test