From cc71929deb3aec14db2c19afd0cfa3ecd7e153ee Mon Sep 17 00:00:00 2001 From: Elena Torro Date: Wed, 26 Aug 2026 11:22:34 +0200 Subject: [PATCH] :sparkles: Add export job model, store and scheduler to exporter --- exporter/src/app/config.cljs | 20 +- exporter/src/app/jobs.cljs | 284 ++++++++++++++++++++ exporter/src/app/jobs/scheduler.cljs | 133 +++++++++ exporter/src/app/jobs/store.cljs | 87 ++++++ exporter/src/app/jobs/utils.cljs | 86 ++++++ exporter/src/app/redis.cljs | 137 +++++++++- exporter/test/exporter_tests/jobs_test.cljs | 86 ++++++ 7 files changed, 820 insertions(+), 13 deletions(-) create mode 100644 exporter/src/app/jobs.cljs create mode 100644 exporter/src/app/jobs/scheduler.cljs create mode 100644 exporter/src/app/jobs/store.cljs create mode 100644 exporter/src/app/jobs/utils.cljs create mode 100644 exporter/test/exporter_tests/jobs_test.cljs diff --git a/exporter/src/app/config.cljs b/exporter/src/app/config.cljs index df88aae0e6..3dcb1e2062 100644 --- a/exporter/src/app/config.cljs +++ b/exporter/src/app/config.cljs @@ -28,7 +28,15 @@ :http-server-port 6061 :http-server-host "0.0.0.0" :tempdir "/tmp/penpot" - :redis-uri "redis://redis/0"}) + :redis-uri "redis://redis/0" + :export-max-concurrent-jobs 4 + :export-max-jobs-per-profile 2 + :export-queue-max 64 + :export-job-ttl 3600 + :wasm-worker-pool-max 2 + :wasm-worker-pool-min 1 + :wasm-render-idle-timeout 300 + :wasm-image-cache-mb 128}) (def ^:private schema:config [:map {:title "config"} @@ -42,7 +50,15 @@ [:redis-uri {:optional true} :string] [:tempdir {:optional true} :string] [:browser-pool-max {:optional true} ::sm/int] - [:browser-pool-min {:optional true} ::sm/int]]) + [:browser-pool-min {:optional true} ::sm/int] + [:export-max-concurrent-jobs {:optional true} ::sm/int] + [:export-max-jobs-per-profile {:optional true} ::sm/int] + [:export-queue-max {:optional true} ::sm/int] + [:export-job-ttl {:optional true} ::sm/int] + [:wasm-worker-pool-max {:optional true} ::sm/int] + [:wasm-worker-pool-min {:optional true} ::sm/int] + [:wasm-render-idle-timeout {:optional true} ::sm/int] + [:wasm-image-cache-mb {:optional true} ::sm/int]]) (def ^:private decode-config (sm/decoder schema:config sm/string-transformer)) diff --git a/exporter/src/app/jobs.cljs b/exporter/src/app/jobs.cljs new file mode 100644 index 0000000000..e711822402 --- /dev/null +++ b/exporter/src/app/jobs.cljs @@ -0,0 +1,284 @@ +;; 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 INC Sucursal en España SL + +(ns app.jobs + "Export job model and lifecycle. + + The record is persisted in redis (`app.jobs.store`); the runtime bits that + cannot be serialized -- cancel callbacks, the cancel signal shared with a + render worker, the throttling bookkeeping -- stay in this process, keyed by + job id. + + Every state change also publishes the same `:export-update` message the + exporter has always published, so websocket clients keep working unchanged." + (:require + [app.common.data :as d] + [app.common.exceptions :as ex] + [app.common.logging :as l] + [app.common.time :as ct] + [app.common.uuid :as uuid] + [app.jobs.store :as store] + [app.redis :as redis] + [promesa.core :as p])) + +(l/set-level! :debug) + +;; A large export reports progress per object; persisting each one would be +;; hundreds of writes for information nobody reads at that resolution. +(def ^:private progress-throttle-ms 250) + +(def ^:private terminal-states #{"ended" "error" "cancelled"}) + +(defonce ^:private registry (atom {})) + +(defn- now-ms + [] + (inst-ms (ct/now))) + +(defn- runtime + [job-id] + (get @registry (str job-id))) + +(defn lookup + "The live record of a job running in this process, or nil." + [job-id] + (:job (runtime job-id))) + +(defn fetch + "The job record from the shared store." + [job-id] + (store/fetch job-id)) + +(defn- publish! + [{:keys [id profile-id resource-id state done total name filename mtype + resource-uri error]}] + (redis/pub! (str profile-id) + (d/without-nils + {:type :export-update + :job-id id + :resource-id resource-id + :status state + :done done + :total total + :name name + :filename filename + :mtype mtype + :resource-uri resource-uri + :cause error}))) + +(defn- store-job! + [job] + (swap! registry update (str (:id job)) assoc :job job) + job) + +(defn create! + "Builds a queued job and persists it. `run-fn` is a 1-arg fn of the job that + performs the export and returns a promise; the caller decides when to run + it." + [{:keys [profile-id cmd backend total name resource-id]} run-fn] + (let [job {:id (uuid/next) + :profile-id profile-id + :cmd cmd + :backend backend + :state "queued" + :done 0 + :total total + :name name + :resource-id resource-id + :created-at (now-ms)}] + (swap! registry assoc (str (:id job)) {:job job :run-fn run-fn :cancelled? false}) + (->> (store/persist! job) + (p/fmap (constantly job))))) + +(defn run-fn + [job-id] + (:run-fn (runtime job-id))) + +(defn cancelled? + [job-id] + (boolean (:cancelled? (runtime job-id)))) + +(defn terminal? + [job] + (contains? terminal-states (:state job))) + +(defn check-cancelled! + "Raises when the job has been cancelled. Called as each render's turn comes + up, so a cancellation stops the ones that have not started yet." + [job] + (when (cancelled? (:id job)) + (ex/raise :type :internal + :code :job-cancelled + :hint "export job was cancelled"))) + +(defn cancel-signal + "Int32Array over a SharedArrayBuffer, readable from a worker thread: 0 while + the job is live, 1 once it has been cancelled. Nil once the job has been + released -- writing the signal back would leave an entry for a settled job in + the registry that nothing would ever remove." + [job-id] + (let [k (str job-id)] + (when-let [rt (get @registry k)] + (or (:cancel-signal rt) + (let [signal (js/Int32Array. (js/SharedArrayBuffer. 4))] + (swap! registry update k (fn [rt] (some-> rt (assoc :cancel-signal signal)))) + (when (cancelled? job-id) + (js/Atomics.store signal 0 1)) + signal))))) + +(defn on-cancel + "Registers a callback used to abort the job's in-flight work (terminating a + render worker). A job fans out over several renders, so callbacks + accumulate. One registered for a job that already settled is dropped: keeping + it would revive that job's registry entry for good." + [job-id f] + (swap! registry update (str job-id) + (fn [rt] (some-> rt (update :cancel-fns (fnil conj []) f))))) + +(defn release! + "Drops the runtime entry once the job settled. The persisted record stays + until its TTL." + [job-id] + (swap! registry dissoc (str job-id))) + +(defn- live + "The job as the lifecycle last left it, or nil once it settled. Callers hold + the snapshot handed to them when their work started; writing that back would + resurrect a failed export as running and drop the error with it." + [job] + (when-let [current (:job (runtime (:id job)))] + (when-not (terminal? current) + current))) + +(defn- persist-and-publish! + [job] + (store-job! job) + (publish! job) + (store/persist! job)) + +(defn transition! + "Moves the job on. The first terminal state wins: anything arriving after it + is dropped, so a late failure cannot overwrite a cancellation, nor a straggler + overwrite either." + [job data] + (if-let [job (live job)] + (persist-and-publish! (merge job data)) + (p/resolved job))) + +(defn start! + [job] + (transition! job {:state "running" :started-at (now-ms)})) + +(defn progress! + "Reports `done` objects completed. Writes are throttled, so the caller need + not care how often it calls this." + [{:keys [id] :as job} done] + (if-let [job (live job)] + (let [k (str id) + now (now-ms) + last (:last-progress-ms (runtime id) 0) + job (assoc job :done done) + write? (>= (- now last) progress-throttle-ms)] + (store-job! job) + (if write? + (do + (swap! registry update k assoc :last-progress-ms now) + (persist-and-publish! job)) + (p/resolved job))) + (p/resolved job))) + +(defn complete! + [job {:keys [uri filename mtype size] :as _resource}] + (transition! job {:state "ended" + :ended-at (now-ms) + :done (:total job) + :resource-uri uri + :filename filename + :mtype mtype + :size size})) + +(defn fail! + [job cause] + (l/error :hint "export job failed" :job-id (str (:id job)) :cause cause) + (transition! job {:state "error" + :ended-at (now-ms) + :error (ex-message cause)})) + +(defn- cancel-local! + [job-id] + (let [k (str job-id) + rt (get @registry k) + job (:job rt)] + (if (or (nil? job) (terminal? job)) + (p/resolved job) + (do + (swap! registry update k (fn [rt] (some-> rt (assoc :cancelled? true)))) + (when-let [signal (:cancel-signal rt)] + (js/Atomics.store signal 0 1)) + ;; Recorded before the callbacks run, not after: one of them + ;; (`scheduler/drop-queued!`) releases the job, and `transition!` on a + ;; released job is a no-op, so a queued job would keep claiming to be + ;; queued in the store and never publish its `cancelled` update. + (let [result (transition! job {:state "cancelled" :ended-at (now-ms)})] + (doseq [f (:cancel-fns rt)] + (try + (f) + (catch :default cause + (l/warn :hint "error on job cancel callback" :job-id k :cause cause)))) + result))))) + +(defn cancel! + "Cancels a job. One this process does not own is broadcast over the cancel + topic, so whoever runs it acts on it. Idempotent." + [job-id] + (if (some? (runtime job-id)) + (cancel-local! job-id) + (->> (fetch job-id) + (p/mcat (fn [job] + (cond + (nil? job) (p/resolved nil) + (terminal? job) (p/resolved job) + :else (p/do + (store/request-cancel! (:id job)) + job))))))) + +(defn- clean-abandoned! + "Marks every job left mid-flight by a previous process as cancelled. + + A queue and its running jobs live in the memory of the process that owns + them, so nothing in flight when it died can be resumed; without this the + record would keep claiming to be running until its TTL expires. + + NOTE: the store cannot tell whose jobs are whose, so with more than one + exporter behind a load balancer this would also cancel a sibling's running + jobs. Single-instance deployments only." + [] + (->> (store/fetch-all) + (p/mcat (fn [jobs] + (let [abandoned (remove terminal? jobs)] + (when (seq abandoned) + (l/warn :hint "cancelling jobs abandoned by a previous process" + :count (count abandoned))) + (->> abandoned + (map (fn [job] + (store/persist! (assoc job + :state "cancelled" + :interrupted true + :ended-at (now-ms))))) + (p/all))))) + (p/fmap (fn [result] (count result))) + (p/merr (fn [cause] + (l/warn :hint "unable to clean abandoned jobs" :cause cause) + (p/resolved 0))))) + +(defn init + [] + (store/on-cancel-request + (fn [job-id] + (when (some? (runtime job-id)) + (l/info :hint "remote cancel request" :job-id job-id) + (cancel-local! job-id)))) + (clean-abandoned!)) diff --git a/exporter/src/app/jobs/scheduler.cljs b/exporter/src/app/jobs/scheduler.cljs new file mode 100644 index 0000000000..c36b0ead74 --- /dev/null +++ b/exporter/src/app/jobs/scheduler.cljs @@ -0,0 +1,133 @@ +;; 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 INC Sucursal en España SL + +(ns app.jobs.scheduler + "Admission control for export jobs. + + Limits concurrent jobs and rejects work rather than allowing an unbounded backlog. + Queue order is FIFO, except jobs whose profile is already at its cap are skipped." + (:require + [app.common.exceptions :as ex] + [app.common.logging :as l] + [app.config :as cf] + [app.jobs :as jobs] + [app.jobs.utils :as job.utils] + [promesa.core :as p])) + +(l/set-level! :debug) + +(defonce ^:private state + (atom {:running {} ;; job-id -> profile-id + :queue []})) ;; vector of {:job :resolve :reject} + +(defn- max-concurrent [] (cf/get :export-max-concurrent-jobs 4)) +(defn- max-per-profile [] (cf/get :export-max-jobs-per-profile 2)) +(defn- max-queued [] (cf/get :export-queue-max 64)) + +(defn- running-for + [{:keys [running]} profile-id] + (count (filter #(= profile-id %) (vals running)))) + +(defn- eligible? + [state profile-id] + (and (< (count (:running state)) (max-concurrent)) + (< (running-for state profile-id) (max-per-profile)))) + +(declare ^:private pump!) + +(defn- finish! + [job-id] + (swap! state update :running dissoc (str job-id)) + (jobs/release! job-id) + (job.utils/release! job-id) + (pump!)) + +(defn- execute! + [{:keys [id profile-id] :as job}] + (swap! state update :running assoc (str id) profile-id) + (if (jobs/cancelled? id) + (do (finish! id) + (p/resolved job)) + (let [run-fn (jobs/run-fn id)] + (->> (p/do (jobs/start! job)) + (p/mcat (fn [job] (p/do (run-fn job)))) + (p/fnly (fn [_ _] (finish! id))))))) + +(defn- drop-queued! + "Removes a queued job and settles its promise, freeing its queue slot on cancellation." + [job-id] + (let [entry (volatile! nil)] + (swap! state (fn [state] + (let [queue (:queue state) + idx (->> (map-indexed vector queue) + (some (fn [[idx entry]] + (when (= (str job-id) (str (-> entry :job :id))) + idx))))] + (if idx + (do (vreset! entry (nth queue idx)) + (assoc state :queue (into (subvec queue 0 idx) (subvec queue (inc idx))))) + state)))) + (when-let [{:keys [resolve]} @entry] + (jobs/release! job-id) + (job.utils/release! job-id) + (resolve nil)))) + +(defn- take-eligible + "Pops the first queued entry that can run now, or nil." + [state] + (let [queue (:queue state) + idx (->> (map-indexed vector queue) + (some (fn [[idx entry]] + (when (eligible? state (-> entry :job :profile-id)) + idx))))] + (when idx + [(assoc state :queue (into (subvec queue 0 idx) (subvec queue (inc idx)))) + (nth queue idx)]))) + +(defn- pump! + [] + (loop [] + (let [entry (volatile! nil)] + (swap! state (fn [state] + (if-let [[next-state next-entry] (take-eligible state)] + (do (vreset! entry next-entry) next-state) + (do (vreset! entry nil) state)))) + (when-let [{:keys [job resolve reject]} @entry] + (-> (execute! job) + (p/then resolve) + (p/catch reject)) + (recur))))) + +(defn submit! + "Registers `job` for execution. Returns a promise of the job's result, which + resolves when the export actually finishes; callers that only need the handle + can ignore it. Raises when the exporter is saturated." + [job] + (let [resolve* (volatile! nil) + reject* (volatile! nil) + pending (p/create (fn [resolve reject] + (vreset! resolve* resolve) + (vreset! reject* reject)))] + (if (eligible? @state (:profile-id job)) + (-> (execute! job) + (p/then @resolve*) + (p/catch @reject*)) + + (let [queued (volatile! false)] + (swap! state (fn [state] + (if (< (count (:queue state)) (max-queued)) + (do (vreset! queued true) + (update state :queue conj {:job job + :resolve @resolve* + :reject @reject*})) + (do (vreset! queued false) state)))) + (when-not @queued + (ex/raise :type :validation + :code :queue-full + :hint "too many queued export jobs")) + (jobs/on-cancel (:id job) (fn [] (drop-queued! (:id job)))) + (l/dbg :hint "export job queued" :job-id (str (:id job))))) + pending)) diff --git a/exporter/src/app/jobs/store.cljs b/exporter/src/app/jobs/store.cljs new file mode 100644 index 0000000000..c4bb9e4672 --- /dev/null +++ b/exporter/src/app/jobs/store.cljs @@ -0,0 +1,87 @@ +;; 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 INC Sucursal en España SL + +(ns app.jobs.store + "Redis persistence for export jobs. + + Stores each job as a single blob with a TTL matching the exported file, + so records expire with their files. + + Reads use Redis. Cancellation requires the process running the job." + (:require + [app.common.logging :as l] + [app.common.transit :as t] + [app.config :as cf] + [app.redis :as redis] + [promesa.core :as p])) + +(def ^:private cancel-topic "export.job-cancel") + +(defn- job-key + [job-id] + (redis/->key "export.job." job-id)) + +(defn- ttl + [] + (cf/get :export-job-ttl 3600)) + +(defn persist! + "Writes the job record and refreshes its TTL. + + If the write fails, log it and continue. The export still runs + and publishes websocket updates, but the job can't be fetched + afterward (fetch returns nil, REST returns 404)." + [{:keys [id state] :as job}] + (let [jkey (job-key id)] + (->> (p/do + (redis/hset! jkey {:data (t/encode-str job)}) + (redis/expire! jkey (ttl)) + job) + (p/merr (fn [cause] + (if (= :redis-not-available (:code (ex-data cause))) + (l/warn :hint "job record not persisted, no redis connection" + :job-id (str id) :state state) + (l/error :hint "unable to persist job record" + :job-id (str id) :state state :cause cause)) + (p/resolved job)))))) + +(defn fetch + "The job record, or nil when unknown or expired." + [job-id] + (->> (redis/hgetall (job-key job-id)) + (p/fmap (fn [data] + (when-let [blob (get data "data")] + (try + (t/decode-str blob) + (catch :default cause + (l/warn :hint "unable to decode job record" :job-id (str job-id) :cause cause) + nil))))))) + +(defn fetch-all + [] + (->> (redis/scan (redis/->key "export.job.*")) + (p/mcat (fn [keys] + (->> (map (fn [k] + (->> (redis/hgetall k) + (p/fmap (fn [data] + (when-let [blob (get data "data")] + (try + (t/decode-str blob) + (catch :default _ nil))))))) + keys) + (p/all)))) + (p/fmap (fn [jobs] (vec (remove nil? jobs)))))) + +(defn request-cancel! + "Asks every exporter to cancel `job-id`. Only the one running it will act." + [job-id] + (redis/pub! cancel-topic (str job-id))) + +(defn on-cancel-request + "Registers `handler` (fn of the job-id string) for cancel requests. Returns an + unsubscribe fn." + [handler] + (redis/sub! cancel-topic handler)) diff --git a/exporter/src/app/jobs/utils.cljs b/exporter/src/app/jobs/utils.cljs new file mode 100644 index 0000000000..49814ee33b --- /dev/null +++ b/exporter/src/app/jobs/utils.cljs @@ -0,0 +1,86 @@ +;; 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 INC Sucursal en España SL + +(ns app.jobs.utils + "Temp file ownership for export jobs. + + Temp files used to be cleaned only by the per-file timer in `app.util.shell`, + an hour after creation and lost entirely on restart. Here each job owns the + paths it creates, so they are dropped as soon as it settles and whatever a + crash left behind is cleaned at boot." + (:require + ["node:fs/promises" :as fsp] + ["node:path" :as path] + [app.common.logging :as l] + [app.config :as cf] + [app.util.shell :as sh] + [cuerdas.core :as str] + [promesa.core :as p])) + +(def ^:private managed-prefix "penpot.") + +(defonce ^:private tracked (atom {})) + +(defn track! + "Registers `path` as owned by `job-id`, so it is removed when the job settles." + [job-id path] + (when (and job-id path) + (swap! tracked update (str job-id) (fnil conj #{}) path)) + path) + +(defn- remove-path! + [path] + (->> (p/do (fsp/rm path #js {:recursive true :force true})) + (p/merr (fn [cause] + (l/warn :hint "unable to remove job temp file" :path path :cause cause) + (p/resolved nil))))) + +(defn release! + "Removes every file the job owns. Called once the job reached a terminal + state and its result has already been uploaded, so nothing else reads them." + [job-id] + (let [k (str job-id) + paths (get @tracked k)] + (swap! tracked dissoc k) + (if (seq paths) + (->> (map remove-path! paths) + (p/all) + (p/fmap (fn [_] + (l/dbg :hint "released job temp files" :job-id k :count (count paths)) + nil))) + (p/resolved nil)))) + +(defn- clean! + "Removes managed temp files older than the job TTL. They can only be leftovers + of a previous process: every live one belongs to a job of this process." + [] + (let [max-age (* 1000 (cf/get :export-job-ttl 3600)) + now (js/Date.now)] + (->> (p/do (fsp/readdir sh/tmpdir)) + (p/mcat (fn [entries] + (->> (filter #(str/starts-with? % managed-prefix) entries) + (map (fn [entry] + (let [fpath (path/join sh/tmpdir entry)] + (->> (p/do (fsp/stat fpath)) + (p/mcat (fn [^js stat] + (if (> (- now (inst-ms (.-mtime stat))) max-age) + (->> (remove-path! fpath) + (p/fmap (constantly 1))) + (p/resolved 0)))) + (p/merr (fn [_] (p/resolved 0))))))) + (p/all)))) + (p/fmap (fn [results] + (let [removed (reduce + 0 results)] + (when (pos? removed) + (l/info :hint "removed orphaned export temp files" :count removed)) + removed))) + (p/merr (fn [cause] + (l/warn :hint "temp file cleanup failed" :cause cause) + (p/resolved 0)))))) + +(defn init + [] + (clean!)) diff --git a/exporter/src/app/redis.cljs b/exporter/src/app/redis.cljs index 8dfb144f39..b143ad78bd 100644 --- a/exporter/src/app/redis.cljs +++ b/exporter/src/app/redis.cljs @@ -8,47 +8,162 @@ (:require ["ioredis" :as redis] [app.common.data.macros :as dm] + [app.common.exceptions :as ex] [app.common.logging :as l] [app.common.transit :as t] - [app.config :as cf])) + [app.config :as cf] + [promesa.core :as p])) (l/set-level! :trace) (def client (atom nil)) +;; A connection in subscriber mode rejects every other command, so the +;; subscriptions need a connection of their own. +(def ^:private subscriber (atom nil)) + +(def ^:private subscriptions (atom {})) + (defn- create-client - [uri] + [uri role] (let [^js client (new redis/default uri)] (.on client "connect" - (fn [] (l/info :hint "redis connection established" :uri uri))) + (fn [] (l/info :hint "redis connection established" :uri uri :role role))) (.on client "error" - (fn [cause] (l/error :hint "error on redis connection" :cause cause))) + (fn [cause] (l/error :hint "error on redis connection" :role role :cause cause))) (.on client "close" - (fn [] (l/warn :hint "connection closed"))) + (fn [] (l/warn :hint "connection closed" :role role))) (.on client "reconnect" - (fn [ms] (l/warn :hint "reconnecting to redis" :ms ms))) + (fn [ms] (l/warn :hint "reconnecting to redis" :role role :ms ms))) (.on client "end" - (fn [] (l/warn :hint "client ended, no more connections will be attempted"))) + (fn [] (l/warn :hint "client ended, no more connections will be attempted" :role role))) client)) +(defn- dispatch-message + [topic payload] + (doseq [handler (get @subscriptions topic)] + (try + (handler payload) + (catch :default cause + (l/error :hint "error on redis subscription handler" :topic topic :cause cause))))) + (defn init [] - (swap! client (fn [prev] - (when prev (.disconnect ^js prev)) - (create-client (cf/get :redis-uri))))) - + (let [uri (cf/get :redis-uri)] + (swap! client (fn [prev] + (when prev (.disconnect ^js prev)) + (create-client uri "commands"))) + (swap! subscriber (fn [prev] + (when prev (.disconnect ^js prev)) + (let [^js conn (create-client uri "subscriber")] + (.on conn "message" (fn [topic payload] (dispatch-message topic payload))) + ;; Reinstate subscriptions after a reconnection. + (.on conn "connect" + (fn [] + (doseq [topic (keys @subscriptions)] + (.subscribe conn topic)))) + conn))))) (defn stop [] + (reset! subscriptions {}) + (swap! subscriber (fn [conn] + (when conn (.quit ^js conn)) + nil)) (swap! client (fn [client] (when client (.quit ^js client)) nil))) (def ^:private tenant (cf/get :tenant)) +(defn ->key + "Namespaces `parts` under the tenant, the same prefix `pub!` uses for topics." + [& parts] + (str tenant "." (apply str parts))) + (defn pub! [topic payload] (let [payload (if (map? payload) (t/encode-str payload) payload) topic (dm/str tenant "." topic)] (when-let [client @client] (.publish ^js client topic payload)))) + +(defn sub! + "Subscribes `handler` (fn of the raw payload string) to `topic`. Returns a + 0-arg fn that removes this handler." + [topic handler] + (let [topic (dm/str tenant "." topic)] + (swap! subscriptions update topic (fnil conj []) handler) + (when-let [conn @subscriber] + (.subscribe ^js conn topic)) + (fn [] + (swap! subscriptions update topic (fn [handlers] (vec (remove #(= % handler) handlers))))))) + +(defn- with-client + "Runs `f` against the command connection. Rejects when there is no connection + or the command fails: whether a failure is survivable depends on what the + caller was doing, and only the caller knows." + [f] + (if-let [client @client] + (p/do (f client)) + (p/rejected (ex/error :type :internal + :code :redis-not-available + :hint "no redis connection")))) + +(defn- with-client-lenient + "For reads, where an unreachable redis is reported as \"nothing there\"." + [f] + (->> (with-client f) + (p/merr (fn [cause] + (l/warn :hint "redis command failed" :cause cause) + (p/resolved nil))))) + +(defn hset! + "Writes `data` (a map of string/keyword -> value) as a hash. Nil values are + dropped, since redis has no null." + [k data] + (let [obj (reduce-kv (fn [obj field value] + (if (some? value) + (doto obj (unchecked-set (name field) (str value))) + obj)) + #js {} + data)] + (if (zero? (alength (js/Object.keys obj))) + (p/resolved nil) + (with-client (fn [^js client] (.hset client k obj)))))) + +(defn hgetall + "Returns the hash as a map of string keys, or nil when it does not exist." + [k] + (->> (with-client-lenient (fn [^js client] (.hgetall client k))) + (p/fmap (fn [result] + (when (and result (pos? (alength (js/Object.keys result)))) + (persistent! + (reduce (fn [res field] + (assoc! res field (unchecked-get result field))) + (transient {}) + (js/Object.keys result)))))))) + +(defn expire! + [k seconds] + (with-client (fn [^js client] (.expire client k seconds)))) + +(defn del! + [k] + (with-client (fn [^js client] (.del client k)))) + +(defn scan + "Every key matching `pattern`, walked in cursor batches so a large keyspace is + never blocked the way `KEYS` would block it." + [pattern] + (letfn [(step [cursor found] + (->> (with-client-lenient (fn [^js client] (.scan client cursor "MATCH" pattern "COUNT" 200))) + (p/mcat (fn [result] + (if (nil? result) + (p/resolved found) + (let [next-cursor (aget result 0) + found (into found (aget result 1))] + (if (= "0" next-cursor) + (p/resolved found) + (step next-cursor found))))))))] + (step "0" []))) diff --git a/exporter/test/exporter_tests/jobs_test.cljs b/exporter/test/exporter_tests/jobs_test.cljs new file mode 100644 index 0000000000..899c0f350f --- /dev/null +++ b/exporter/test/exporter_tests/jobs_test.cljs @@ -0,0 +1,86 @@ +;; 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 INC Sucursal en España SL + +(ns exporter-tests.jobs-test + "Job state machine. Runs without redis: a store write with no connection is + reported and swallowed, so only the in-process record is exercised." + (:require + [app.common.uuid :as uuid] + [app.jobs :as jobs] + [cljs.test :as t :include-macros true] + [promesa.core :as p])) + +(defn- create! + [] + (jobs/create! {:profile-id (uuid/next) + :cmd :export-shapes + :backend "wasm" + :total 10 + :name "test" + :resource-id (uuid/next)} + (constantly (p/resolved nil)))) + +(t/deftest progress-does-not-resurrect-a-finished-job + (t/testing "a render reporting after the export failed cannot undo the failure" + (t/async done + (p/let [job (create!) + _ (jobs/start! job) + _ (jobs/fail! (jobs/lookup (:id job)) (ex-info "boom" {})) + ;; `job` is the snapshot handed to the work when it started, which + ;; is what a straggling render still holds. + _ (jobs/progress! job 7)] + (let [current (jobs/lookup (:id job))] + (t/is (= "error" (:state current))) + (t/is (= "boom" (:error current))) + (t/is (not= 7 (:done current)))) + (jobs/release! (:id job)) + (done))))) + +(t/deftest first-terminal-state-wins + (t/testing "a failure arriving after a cancellation leaves the job cancelled" + (t/async done + (p/let [job (create!) + _ (jobs/start! job) + _ (jobs/cancel! (:id job)) + _ (jobs/fail! job (ex-info "too late" {}))] + (let [current (jobs/lookup (:id job))] + (t/is (= "cancelled" (:state current))) + (t/is (nil? (:error current)))) + (jobs/release! (:id job)) + (done))))) + +(t/deftest cancel-is-recorded-before-the-callbacks-run + (t/testing "a queued job dropped by its own cancel callback still ends cancelled" + (t/async done + (let [seen (atom ::not-called)] + (p/let [job (create!) + ;; What `scheduler/drop-queued!` does: it takes the job off the + ;; queue and releases it. Anything the lifecycle wrote after + ;; the callbacks ran would be dropped on the floor, so by the + ;; time one is called the record has to be terminal already. + _ (jobs/on-cancel (:id job) + (fn [] + (reset! seen (:state (jobs/lookup (:id job)))) + (jobs/release! (:id job)))) + _ (jobs/cancel! (:id job))] + (t/is (= "cancelled" @seen)) + (t/is (nil? (jobs/lookup (:id job)))) + (done)))))) + +(t/deftest writes-stop-once-the-job-is-released + (t/testing "a late write for a job the scheduler already settled is dropped" + (t/async done + (p/let [job (create!) + _ (jobs/start! job) + _ (jobs/complete! (jobs/lookup (:id job)) {:uri "http://example/x" + :filename "x.zip" + :mtype "application/zip"}) + ended (jobs/lookup (:id job)) + _ (jobs/release! (:id job)) + _ (jobs/progress! job 3)] + (t/is (= "ended" (:state ended))) + (t/is (nil? (jobs/lookup (:id job)))) + (done)))))