✨ Add export job model, store and scheduler to exporter

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))

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!))

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))

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))

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!))

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" [])))

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)))))