diff --git a/.serena/memories/exporter/core.md b/.serena/memories/exporter/core.md index ee8841e92c..b1f5c0de39 100644 --- a/.serena/memories/exporter/core.md +++ b/.serena/memories/exporter/core.md @@ -1,47 +1,49 @@ # Exporter Architecture and Workflow -`exporter/`: CLJS/Node headless export service. Depends on `common/`; uses Playwright plus export JS/CLJS deps for SVG/PDF/assets. +`exporter/`: CLJS/Node **worker process** that consumes the backend's +`:exporter` job queue and renders shapes, frames and files to bitmap, SVG and +PDF. Depends on `common/`; uses Playwright plus export JS/CLJS deps for +SVG/PDF/assets. There is no HTTP surface of its own and nothing is persisted +on this side: the row of the job, its cancel and its result are the backend's. ## Layout and commands - Source: `exporter/src/`; config: `deps.edn`, `shadow-cljs.edn`, `package.json`; runtime helpers/assets: `vendor/`, `scripts/`. -- From `exporter/`: setup `./scripts/setup`; watch `pnpm run watch` or `pnpm run watch:app`; production build `pnpm run build`; test bundle `pnpm run build:test`; tests `pnpm run test` or `pnpm run test:quiet`; lint `pnpm run lint:clj`; format check/fix `pnpm run check-fmt:clj` / `pnpm run fmt:clj`. +- From `exporter/`: setup `./scripts/setup`; watch `pnpm run watch` or `pnpm run watch:app`; production build `pnpm run build`; test bundle `pnpm run build:test`; tests `pnpm run test` or `pnpm run test:quiet`; lint `pnpm run lint:clj`; format check/fix `pnpm run check-fmt:clj` / `pnpm run fmt:clj`. `./scripts/ci exporter` in the repo root is green (test-quiet defaults `PENPOT_SECRET_KEY` like `test` does). - Because exporter consumes `common/`, shared file/shape/model changes may need exporter verification even when the immediate change is not under `exporter/`. - Cross-cutting testing principles and anti-patterns: `mem:testing`. - Exporter test conventions and CI: `mem:exporter/testing`. -- Promise chains in the exporter (management client, auth) follow promesa's two API families; arg orders, what each fn must return, the `->>` preference and the async-test silent-green trap: `mem:clojure/promesa` — read before touching promise code. +- Promise chains in the exporter (management client, runner) follow promesa's two API families; arg orders, what each fn must return, the `->>` preference and the async-test silent-green trap: `mem:clojure/promesa` — read before touching promise code. -## HTTP and browser pool (legacy surface, doomed) +## The worker anatomy -- POST body limit is about 60 MB. Exporter supports `application/transit+json`; request params merge query params and body params. -- Map response bodies are Transit JSON and force HTTP 200; nil 200 bodies become 204. -- Auth token comes from cookie `auth-token`, then uploads use Bearer auth plus the management shared key. -- Each export job gets a fresh Playwright browser context. On success, the context closes and the browser returns to the pool; on error, the browser is destroyed instead of reused. -- Borrow validates browser connection. Pool acquire timeout is about 10s; font loading timeout logs a warning and continues after about 15s. -- The `http` role needs `PENPOT_EXPORTER_ROLES=http`. Default is gone once the HTTP deletion task lands; the scheduler config keys (`PENPOT_EXPORTER_MAX_CONCURRENT_JOBS` and friends) belong to this surface. - -## Worker (consumer of the jobs substrate) - -- Roles: `PENPOT_EXPORTER_ROLES` selects `http` and/or `worker`; **default is `worker`** (consumer-only process, no HTTP server to preserve): `app.consumer.config`, all accessors read live config. -- `worker` boots K pollers (`app.consumer.worker`): one Redis connection per poller (connection = slot + backpressure), BLPOP on `penpot.worker.queue::exporter`, payload `[job-id scheduled-at]` JSON; corrupt payload warns and drops, dead Redis delays 1s and retries. `K = PENPOT_EXPORTER_WORKER_CONCURRENCY` default 2, floor 1; replicas × K is the total concurrency. Pollers start after pools in `core.cljs` and stop before redis in the shutdown. +- Main thread boots pools + K pollers (`core.cljs`): browser pool, wasm pool, temp-file cleaner (`jobs.utils/init`), then `consumer.worker/start!`. Shutdown: browser → wasm pool → pollers → done (the pollers own their connections; the pools unwind while an export is still in flight). +- K pollers (`app.consumer.worker`): one Redis connection per poller (connection = slot + backpressure), BLPOP on `penpot.worker.queue::exporter`, payload `[job-id scheduled-at]` JSON; corrupt payload warns and drops, dead Redis delays 1s and retries. `K = PENPOT_EXPORTER_WORKER_CONCURRENCY` default 2, floor 1; replicas × K is the total concurrency. Pools size themselves to K (`browser` and `wasm.pool` read `ccfg/concurrency`); there is no independent pool-max config anymore, only `:wasm-worker-pool-min`. - Management client (`app.consumer.api`): claim/progress/create-job-session/complete-job (JSON and multipart via undici FormData)/fail-job, POST to `/api/management/methods/` with `X-Shared-Key: exporter ` (key = `PENPOT_EXPORTER_SHARED_KEY` or HKDF-derived from the secret); transit bodies, non-2xx errors carry `:status` and the error body. -- Runner (`app.consumer.exports`, `run-export!`): mints a render session (`create-job-session`), plans via `make-plan` (single → artifact of its type; multi → zip; frames → pdf via pdfunite), settles via `complete-job-with-artifact` multipart or `fail-job`; the settle never rejects. +- Runner (`app.consumer.exports`, `run-export!`): mints a render session (`create-job-session`), plans via `make-plan` (single → artifact of its type; multi → zip; frames → pdf via pdfunite), settles via `complete-job-with-artifact` multipart or `fail-job`; the settle never rejects. `app.consumer.plan` holds the render-plan pieces (name transducers, partition of 50, grouping by `[scale type]`). - Beats: per-object milestones `:preparing/:rendering/:packaging` with `objects`/`pages` counters, throttled at 250ms; a watchdog repeats the last milestone every 1s. -- Cancellation: a `skip` answer on any beat means the row is terminal (cancelled, aborted, settled). The runner raises `:job-cancelled` between units of work; mid-Skia the watchdog fires the local port once (`jobs/mark-cancelled`: flag + SharedArrayBuffer + terminate callbacks registered by `renderer.wasm/with-scope`), killing the leased worker thread. The runner owns the local registry entry via `jobs/register!` and releases it on settle. -- The legacy job machinery (`app.jobs`, scheduler, store) only serves the `http` surfaces plus the local cancel arm; the worker reuses pools and `job.utils/track!/release!` for temp files. - -## Export batching and async behavior - -- `prepare-exports` groups entries by `[scale type]` and partitions groups into chunks of 50. Each partition uses file/page/share/name from its first item, so be careful if entries might cross those boundaries. -- Single-export response is used only when multiple export is not forced and there is exactly one prepared export containing exactly one object. -- Multi-object export can run async: when `wait` is false it returns a resource immediately and publishes progress/end/error to Redis by profile topic; when `wait` is true it waits for upload and returns the uploaded resource. -- Frame export returns a resource immediately and publishes Redis updates; it does not follow the same `wait` option path. -- ZIP entry names are sanitized and duplicates receive numeric suffixes. +- Cancellation: a `skip` answer on any beat means the row is terminal (cancelled, aborted, settled). The runner raises `:job-cancelled` between units of work; mid-Skia the watchdog fires the local port once (`jobs/mark-cancelled`: flag + terminal local record + SharedArrayBuffer + terminate callbacks registered by `renderer.wasm/with-scope`), killing the leased worker thread. The runner owns the local registry entry via `jobs/register!` and releases it on settle. +- `app.jobs` is the **local cancel arm only**: a runtime registry keyed by job-id (`register!`, `mark-cancelled`, `cancel-signal`, `on-cancel`, `cancelled?`, `release!`). `mark-cancelled` marks the local record terminal BEFORE the callbacks run and returns true/nil (marked-now / no-op for unowned or settled). No redis writes. +- Temp files: `app.jobs.utils/track!`/`release!` per job id; a boot-time clean drops what a previous process left (aged `:exporter-job-ttl`). After the deletion there is no redis-side store, cancel topic or abandoned-job sweep: an unclaimed-in-time row goes `aborted` by the backend's own lease GC. ## Render details +- Headless engines (wasm/Skia) lease one render worker for the whole run (`rd/with-scope`); browser renders go one DOM page per partition of 50. +- Each export gets a fresh Playwright browser context. On success the context closes and the browser returns to the pool; on error the browser is destroyed instead of reused. Borrow validates the connection; pool acquire timeout about 10s; font loading timeout logs a warning and continues after about 15s. - Bitmap export differs for WASM vs non-WASM render paths: WASM forces Playwright `deviceScaleFactor` to 1 and passes scale through the render URL; non-WASM uses `deviceScaleFactor = scale`. - WebP is produced by taking a PNG screenshot and converting it with ImageMagick. - SVG export rasterizes text foreignObjects to PNG, converts through PPM/color masks/potrace, and reassembles SVG paths. It also replaces non-breaking spaces for SVG compatibility and drops empty defs/paths. - PDF export injects `@page` sizing through raw browser `evaluate` JavaScript; that code cannot rely on CLJS runtime helpers. -- Temporary resources schedule local deletion, then uploads POST to `/api/management/methods/upload-tempfile` with `X-Shared-Key: exporter ` and Bearer auth. +- ZIP entry names are sanitized (`plan/sanitize-file-regex`) and duplicates receive numeric suffixes. + +## On the render engines (do not lose in refactors) + +- The wasm engines run on worker threads; the Skia wasm bundle needs + `../render-wasm/build export` to generate `src/app/wasm/shared.js` before + compiling. +- A render thread cannot read a flag: the cancel reaches it through the + SharedArrayBuffer signal, and a render that never answers is killed by the + terminate watchdog of `app.wasm.pool` (the same mechanism a hard cancel + rides on). +- `core.cljs` decides what boots by thread: the main thread consumes the + queue, a render worker (`wasm.worker/main`) renders. diff --git a/.serena/memories/exporter/testing.md b/.serena/memories/exporter/testing.md index 189c1e852c..ab066cca6d 100644 --- a/.serena/memories/exporter/testing.md +++ b/.serena/memories/exporter/testing.md @@ -12,5 +12,6 @@ - Focus a test var with `node target/tests/test.js --focus exporter-tests.renderer-svg-test/creates-the-correct-gradient-element`. - Set app log level by appending `--log-level warn` (or `trace|debug|info|warn|error`). - `test:quiet` accepts forwarded options but rebuilds the bundle; prefer the direct runner after `build:test` for focused runs. +- The direct runner and `test`/`test:quiet` default `PENPOT_SECRET_KEY=test-secret-key` when the environment says nothing: the config needs a secret to boot. - From `exporter/`: `pnpm run check-fmt:clj` checks ClojureScript formatting. - From `exporter/`: `pnpm run lint:clj` runs ClojureScript linting. diff --git a/docker/images/docker-compose.yaml b/docker/images/docker-compose.yaml index e667bc3049..02fd2af3cb 100644 --- a/docker/images/docker-compose.yaml +++ b/docker/images/docker-compose.yaml @@ -211,6 +211,9 @@ services: - penpot penpot-exporter: + # The exporter is a consumer's process: it polls the backend's + # `:exporter` job queue through valkey and calls the backend's + # management API to claim, report and settle every export job. image: "penpotapp/exporter:${PENPOT_VERSION:-2.18}" restart: always @@ -225,17 +228,17 @@ services: << : [*penpot-flags, *penpot-secret-key, *penpot-public-uri] # Don't touch it; this uses an internal docker network to - # communicate with the frontend. + # reach the backend's management API and the frontend of the + # assets the exports render. PENPOT_INTERNAL_URI: http://penpot-frontend:8080 - ## Valkey (or previously Redis) is used for the websockets notifications - ## and for storing the state export jobs + ## Valkey (or previously Redis) carries the export jobs queue the + ## backend dispatcher pushes to, plus the websockets + ## notifications PENPOT_REDIS_URI: redis://penpot-valkey/0 - # PENPOT_EXPORTER_MAX_CONCURRENT_JOBS: 4 - # PENPOT_EXPORTER_MAX_JOBS_PER_PROFILE: 2 - # PENPOT_EXPORTER_QUEUE_MAX: 64 # PENPOT_EXPORTER_JOB_TTL: 3600 + # PENPOT_EXPORTER_WORKER_CONCURRENCY: 2 # PENPOT_WASM_WORKER_POOL_MAX: 2 # PENPOT_WASM_WORKER_POOL_MIN: 1 # PENPOT_WASM_WORKER_IDLE_TIMEOUT: 300 diff --git a/docker/images/files/nginx-entrypoint.sh b/docker/images/files/nginx-entrypoint.sh index 3ec97403a8..3d0d2a5124 100644 --- a/docker/images/files/nginx-entrypoint.sh +++ b/docker/images/files/nginx-entrypoint.sh @@ -52,13 +52,12 @@ update_oidc_name /var/www/app/js/config.js ######################################### export PENPOT_BACKEND_URI=${PENPOT_BACKEND_URI:-http://penpot-backend:6060} -export PENPOT_EXPORTER_URI=${PENPOT_EXPORTER_URI:-http://penpot-exporter:6061} export PENPOT_HTTP_SERVER_MAX_BODY_SIZE=${PENPOT_HTTP_SERVER_MAX_BODY_SIZE:-367001600} # Default to 350MiB export PENPOT_IPV6_LISTEN_DIRECTIVE=${PENPOT_IPV6_LISTEN_DIRECTIVE:-"listen [::]:8080 default_server reuseport backlog=16384;"} if is_truthy "${PENPOT_DISABLE_IPV6_LISTEN:-}"; then export PENPOT_IPV6_LISTEN_DIRECTIVE="" fi -envsubst "\$PENPOT_BACKEND_URI,\$PENPOT_EXPORTER_URI,\$PENPOT_HTTP_SERVER_MAX_BODY_SIZE,\$PENPOT_IPV6_LISTEN_DIRECTIVE" \ +envsubst "\$PENPOT_BACKEND_URI,\$PENPOT_HTTP_SERVER_MAX_BODY_SIZE,\$PENPOT_IPV6_LISTEN_DIRECTIVE" \ < /tmp/nginx.conf.template > /etc/nginx/nginx.conf if [[ $PENPOT_FLAGS == *"enable-admin-console"* ]]; then diff --git a/docker/images/files/nginx.conf.template b/docker/images/files/nginx.conf.template index 0451e7a24a..deed4e1846 100644 --- a/docker/images/files/nginx.conf.template +++ b/docker/images/files/nginx.conf.template @@ -148,11 +148,6 @@ http { proxy_pass $PENPOT_BACKEND_URI/link-preview$is_args$args; } - location /api/export { - proxy_set_header Host $proxy_host; - proxy_pass $PENPOT_EXPORTER_URI; - } - location /api { proxy_set_header Host $proxy_host; proxy_pass $PENPOT_BACKEND_URI/api; diff --git a/exporter/README.md b/exporter/README.md index df4d4412db..b954b643b8 100644 --- a/exporter/README.md +++ b/exporter/README.md @@ -1,99 +1,55 @@ # Exporter -Node service that renders shapes and files to bitmap, SVG and PDF. Wasm exports -are **jobs**: created over HTTP, admitted by a scheduler with bounded -concurrency, and persisted in Redis so their state can be queried and cancelled. -The legacy entry point, which is what the browser backend still goes through, -runs the export as soon as it is asked for, with no admission control. +Node process that renders shapes, frames and files to bitmap, SVG and PDF. It +is a **consumer**: it polls the backend's `:exporter` job queue on Redis and +drives each export through the backend's management API — claim, report the +milestones, and settle (`complete-job` multipart with the artifact, or +`fail-job` on error). There is no HTTP surface of its own and nothing is +persisted on this side: the row of the job, its cancel button and its result +all live in the backend. -## HTTP API +## How a job runs -Mounted under `/api/export` (the router matches on the path *after* that prefix, -so it also works when the process is hit directly on `/`). +1. The backend dispatcher pushes `[job-id, scheduled-at]` (JSON) into + `penpot.worker.queue::exporter`. +2. One of the `K` pollers pops it and owns that connection for the whole run: + the blocking pop is both the semaphore of the slot and the backpressure, so + the process never runs more exports than pollers. `K` is + `PENPOT_EXPORTER_WORKER_CONCURRENCY` (default 2, and the browser and wasm + pools size themselves to the same figure). +3. The poller claims the job over the management API and, on `:run`, mints a + render session (`create-job-session`) through which every asset fetch + happens as the owner of the job. +4. The milestones are reported as beats (`:preparing`, `:rendering`, + `:packaging`, with the `objects` and `pages` counters). A watchdog re-sends + the last one every second, whatever the run is doing. +5. It settles with `complete-job` (the artifact rides multipart, the session + is closed in the same step) or `fail-job`. The settle never rejects: a + worker that cannot talk leaves the row to the backend's lease GC. -| Method | Path | Description | -|----------|-----------------|----------------------------------------------------| -| `POST` | `/` | Legacy command multiplex; runs unscheduled | -| `POST` | `/jobs` | Create an export job | -| `GET` | `/jobs/{id}` | Job record | -| `DELETE` | `/jobs/{id}` | Request cancellation | +## Cancellation -Job states: `queued` -> `running` -> `ended` | `error` | `cancelled`. The last -three are terminal. - -## Redis layout - -Every key is namespaced with `penpot.exporter.` plus the tenant -(`PENPOT_TENANT`, `default` in code but set to the workspace name in devenv, -e.g. `devenv-ws0`). - -``` -penpot.exporter.{tenant}.job.{job-id} hash field: data (transit blob of the - whole record) -penpot.exporter.{tenant}.job-cancel pubsub payload: the job id, one line -``` - -There is no index: the keyspace is one self-expiring hash per job and nothing -else. Each hash carries the same TTL as the exported file -(`PENPOT_EXPORTER_JOB_TTL`, default 3600s), refreshed on every write and never -after the job settles. - -## Inspecting Redis - -Redis is not published on the host, so `redis-cli` from your machine gets -connection refused. Run it **inside the devenv container**, against the `valkey` -host on database 0: - -```bash -redis-cli -h valkey -n 0 -``` - -`redis-cli -u "$PENPOT_REDIS_URI"` does the same and follows whatever the env is -set to (`redis://valkey/0` in devenv). - -Keys carry the tenant, which in devenv is the **workspace name** -(`$PENPOT_TENANT`, e.g. `devenv-ws0`), not `default`. From the prompt: - -``` -# every job record -KEYS penpot.exporter.devenv-ws0.job.* - -# the whole record, transit-json in the `data` field -HGET penpot.exporter.devenv-ws0.job. data - -# seconds left before the record expires -TTL penpot.exporter.devenv-ws0.job. - -# watch cancellations as they are published (blocks the connection) -SUBSCRIBE penpot.exporter.devenv-ws0.job-cancel - -# drop one record -DEL penpot.exporter.devenv-ws0.job. -``` - -`KEYS` is fine here -- the keyspace is a handful of job hashes. On a real -deployment use `SCAN 0 MATCH penpot.exporter..job.* COUNT 100` instead. -Do not `FLUSHDB`: the backend shares this database. - -The backend debug UI also renders these records: `/dbg` has an *Export jobs* -section, with a `?job-id=` filter. +A cancellation reaches the worker through the beats: a `skip` answer means the +backend no longer listens (the row is terminal), so the runner stops between +units of work; mid-Skia the watchdog terminates the leased render worker. The +row was already `cancelled` by the backend's `cancel-job`. ## Configuration -| Variable | Default | Description | -|---------------------------------------|---------|--------------------------------------| -| `PENPOT_REDIS_URI` | `redis://redis/0` | Job store and cancel topic | -| `PENPOT_TENANT` | `default` | Key and topic prefix | -| `PENPOT_EXPORTER_JOB_TTL` | `3600` | Lifetime of a job record, in seconds | -| `PENPOT_EXPORTER_MAX_CONCURRENT_JOBS` | `4` | Admission limit | -| `PENPOT_EXPORTER_MAX_JOBS_PER_PROFILE`| `2` | Per-profile admission limit | -| `PENPOT_EXPORTER_QUEUE_MAX` | `64` | Queue cap; over it, `429 :queue-full` | -| `PENPOT_WASM_WORKER_POOL_MAX` | `2` | Headless render worker threads; min 1 | -| `PENPOT_WASM_WORKER_POOL_MIN` | `1` | Workers kept warm; clamped to the max | -| `PENPOT_WASM_WORKER_IDLE_TIMEOUT` | `300` | Silence before a worker is terminated, in seconds | -| `PENPOT_WASM_WORKER_IMAGE_CACHE_SIZE` | `134217728` | Per-worker image cache budget, in bytes | +| Variable | Default | Description | +|------------------------------------------|---------|----------------------------------------------------| +| `PENPOT_PUBLIC_URI` | — | The frontend the assets render through | +| `PENPOT_INTERNAL_URI` | `public-uri` | Same network's frontend; the management API too | +| `PENPOT_REDIS_URI` | `redis://redis/0` | The queue and the rest of redis | +| `PENPOT_TENANT` | `default` | Queue prefix (workspace name in devenv) | +| `PENPOT_EXPORTER_SHARED_KEY` | — | The shared key; derived from the secret otherwise | +| `PENPOT_EXPORTER_JOB_TTL` | `3600` | Lifetime of the temp files a job owns, in seconds | +| `PENPOT_EXPORTER_WORKER_CONCURRENCY` | `2` | `K`: pollers, running exports, pool sizes | +| `PENPOT_WASM_WORKER_POOL_MIN` | `1` | Render workers kept warm; clamped to the max | +| `PENPOT_WASM_WORKER_IDLE_TIMEOUT` | `300` | Silence before a worker is terminated, in seconds | +| `PENPOT_WASM_WORKER_IMAGE_CACHE_SIZE` | `134217728` | Per-worker image cache budget, in bytes | -A headless job leases one render worker for its whole run, so it is admitted -only when a worker is free: `PENPOT_WASM_WORKER_POOL_MAX` is the real limit for -them, and `PENPOT_EXPORTER_MAX_CONCURRENT_JOBS` bounds the browser ones -alongside. +## Inspecting the backend + +Nothing about a job is stored here: `get-job` in the backend answers the full +row, and its `app.debug`/admin surfaces render the same records. diff --git a/exporter/scripts/test-quiet.js b/exporter/scripts/test-quiet.js index b1be0dd682..c7f7c610aa 100644 --- a/exporter/scripts/test-quiet.js +++ b/exporter/scripts/test-quiet.js @@ -24,6 +24,13 @@ progress("Running tests..."); const result = spawnSync( "node", ["target/tests/test.js", ...process.argv.slice(2)], - { stdio: "inherit" }, -); + { + stdio: "inherit", + // the config of the process needs a secret to even boot: same + // default the `test` script uses when the environment says nothing + env: { + ...process.env, + PENPOT_SECRET_KEY: process.env.PENPOT_SECRET_KEY || "test-secret-key", + }, + }); process.exit(result.status ?? 1); diff --git a/exporter/src/app/auth.cljs b/exporter/src/app/auth.cljs deleted file mode 100644 index 0bcfb27199..0000000000 --- a/exporter/src/app/auth.cljs +++ /dev/null @@ -1,98 +0,0 @@ -;; 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.auth - "Resolves the caller's session cookie to a real profile id. - - The export commands take `:profile-id` from the request body, which was - harmless while it only picked a pub/sub topic. The job API can read and - cancel other people's work, so its ownership comes from the session: the - token goes to the backend's `get-profile` command, which answers with the - anonymous profile (`uuid/zero`) when it is not a valid session. - - Results are memoized briefly, so a burst of export calls from one client is - one round trip rather than one per request." - (:require - ["undici" :as http] - [app.common.exceptions :as ex] - [app.common.logging :as l] - [app.common.transit :as t] - [app.common.uri :as u] - [app.common.uuid :as uuid] - [app.config :as cf] - [promesa.core :as p])) - -(def ^:private cache-ttl-ms 60000) - -(defonce ^:private cache (atom {})) - -(defn- put-in-cache - "Stores the resolution by session token and removes expired entries. - Without cleanup, long-lived exporters accumulate stale entries" - [cache token profile-id now] - (-> (into {} (remove (fn [[_ {:keys [expires-at]}]] (<= expires-at now))) cache) - (assoc token {:profile-id profile-id - :expires-at (+ now cache-ttl-ms)}))) - -(defn- rpc-uri - [] - (-> (cf/get-internal-uri) - (u/ensure-path-slash) - (u/join "api/rpc/command/get-profile") - (str))) - -(defn- fetch-profile-id - [token] - (let [uri (rpc-uri) - headers #js {"Content-Type" "application/transit+json" - "X-Shared-Key" (str "exporter " cf/management-key) - "Cookie" (str "auth-token=" token)}] - (->> (p/do (http/fetch uri #js {:method "POST" :headers headers :body (t/encode-str {})})) - (p/mcat (fn [^js resp] - (if (= 200 (.-status resp)) - (.text resp) - (p/resolved nil)))) - (p/fmap (fn [body] - (some-> body t/decode-str :id))) - (p/merr (fn [cause] - (l/warn :hint "unable to resolve session profile" :uri uri :cause cause) - (p/resolved nil)))))) - -(defn resolve-profile-id - "Promise of the authenticated profile id, or nil for an anonymous or absent - session." - [token] - (if (nil? token) - (p/resolved nil) - (let [{:keys [profile-id expires-at]} (get @cache token)] - (if (and expires-at (> expires-at (js/Date.now))) - (p/resolved profile-id) - (->> (fetch-profile-id token) - (p/fmap (fn [profile-id] - (let [profile-id (when (and profile-id (not= uuid/zero profile-id)) profile-id)] - (swap! cache put-in-cache token profile-id (js/Date.now)) - profile-id)))))))) - -(defn require-profile-id - "Like `resolve-profile-id`, but rejects anonymous callers." - [token] - (->> (resolve-profile-id token) - (p/mcat (fn [profile-id] - (if profile-id - (p/resolved profile-id) - (ex/raise :type :authentication - :code :authentication-required - :hint "no valid session for this request")))))) - -(defn check-owner! - "Raises unless `profile-id` owns `job`." - [job profile-id] - (when (or (nil? job) - (not= (str (:profile-id job)) (str profile-id))) - (ex/raise :type :not-found - :code :object-not-found - :hint "job does not exist")) - job) diff --git a/exporter/src/app/browser.cljs b/exporter/src/app/browser.cljs index 9a02e0a5ae..7d1c88e847 100644 --- a/exporter/src/app/browser.cljs +++ b/exporter/src/app/browser.cljs @@ -11,7 +11,7 @@ ["playwright" :as pw] [app.common.exceptions :as ex] [app.common.logging :as l] - [app.config :as cf] + [app.consumer.config :as ccfg] [app.util.object :as obj] [promesa.core :as p])) @@ -155,8 +155,8 @@ (defn init [] - (let [opts #js {:max (cf/get :browser-pool-max 5) - :min (cf/get :browser-pool-min 0) + (let [opts #js {:max (max 1 (ccfg/concurrency)) + :min 0 :testOnBorrow true :evictionRunIntervalMillis 5000 :numTestsPerEvictionRun 5 diff --git a/exporter/src/app/config.cljs b/exporter/src/app/config.cljs index d9829b7ec9..324d4153d8 100644 --- a/exporter/src/app/config.cljs +++ b/exporter/src/app/config.cljs @@ -24,21 +24,12 @@ {:public-uri "http://localhost:3449" ;; :internal-uri nil ;; internal-uri cannot be nil :tenant "default" - :host "localhost" - :http-server-port 6061 - :http-server-host "0.0.0.0" :tempdir "/tmp/penpot" :redis-uri "redis://redis/0" - :exporter-max-concurrent-jobs 4 - :exporter-max-jobs-per-profile 2 - :exporter-queue-max 64 :exporter-job-ttl 3600 - ;; the roles this instance serves: by default the consumer of the - ;; backend jobs queue; `http` keeps the legacy surfaces alive until - ;; the deletion task takes them away - :exporter-roles "worker" + ;; how many export jobs run at once: every poller owns one Redis + ;; connection, and the pools size themselves to the same number :exporter-worker-concurrency 2 - :wasm-worker-pool-max 2 :wasm-worker-pool-min 1 :wasm-worker-idle-timeout 300 :wasm-worker-image-cache-size (* 128 1024 1024)}) @@ -49,7 +40,6 @@ [:public-uri {:optional true} ::sm/uri] [:internal-uri {:optional true} ::sm/uri] [:exporter-shared-key {:optional true} :string] - [:host {:optional true} :string] [:tenant {:optional true} :string] [:flags {:optional true} [::sm/set :keyword]] [:redis-uri {:optional true} :string] diff --git a/exporter/src/app/consumer/config.cljs b/exporter/src/app/consumer/config.cljs index 460f634106..7d524f379d 100644 --- a/exporter/src/app/consumer/config.cljs +++ b/exporter/src/app/consumer/config.cljs @@ -5,50 +5,18 @@ ;; Copyright (c) KALEIDOS SUBSIDIARY SL (ns app.consumer.config - "The roles of the exporter process. - - `PENPOT_EXPORTER_ROLES` names the roles the process serves: `http` - serves the surfaces the exporter used to serve (gone once the - deletion task lands), `worker` consumes the backend `:exporter` - queue. The default is `worker`: the exporter is a consumer's process, - and no surface of it needs to be preserved. 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." + "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] - [cuerdas.core :as str])) + [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- 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) - #{:worker} - (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 diff --git a/exporter/src/app/consumer/exports.cljs b/exporter/src/app/consumer/exports.cljs index 6c34f59a25..f8e4e63aec 100644 --- a/exporter/src/app/consumer/exports.cljs +++ b/exporter/src/app/consumer/exports.cljs @@ -7,17 +7,15 @@ (ns app.consumer.exports "The runner of the `:export-assets` job. - Takes the params the job-def froze and transplant the legacy export - machinery into the jobs substrate: the name transducers and - partitioning of `app.handlers.export-shapes`, and the pdf assembly - its `export-frames` twin does, dress the same run the old `/api/export` - served. What changes around them is the frame: the session minted - through `create-job-session` renders as the user, the milestone - vocabulary of the progress contract dresses the beats (`:preparing`, + Runs one export end to end: the render plan of + `app.consumer.plan` (the name transducers and the partitioning that + the export surfaces always kept), the pdf assembly of the frames + run, the beats shaped by the progress contract (`:preparing`, `:rendering`, `:packaging`, with the `objects` and `pages` counters - of every partition), and the settle is one `complete-job` multipart - call that stores the artifact and closes the session in the same - step. + of every partition) and one settle: a `complete-job` multipart call + that stores the artifact and closes the session the backend minted + in the same step. The render runs as the job's owner through the + session `create-job-session` opened. Cancellation arrives through the beats: a `skip` answer from `report-job-progress` says the job can no longer hear the worker @@ -34,7 +32,7 @@ [app.common.exceptions :as ex] [app.common.logging :as l] [app.consumer.api :as api] - [app.handlers.export-shapes :as shapes] + [app.consumer.plan :as plan] [app.handlers.resources :as rsc] [app.jobs :as jobs] [app.jobs.utils :as job.utils] @@ -143,7 +141,7 @@ [token {:keys [exports force-multiple name skip-children is-wasm]}] (let [items (normalize-items exports) frames? (boolean (every? frame-item? items)) - prepared (shapes/prepare-exports items token is-wasm) + prepared (plan/prepare-exports items token is-wasm) single? (and (not frames?) (not (true? force-multiple)) (= 1 (count prepared)) @@ -155,7 +153,7 @@ {:frames? frames? :single? single? :prepared prepared - :total (shapes/count-objects prepared) + :total (plan/count-objects prepared) :skip-children skip-children :is-wasm (boolean is-wasm) :counter-kind (if frames? :pages :objects) @@ -230,7 +228,7 @@ (counter counter-kind @rendered total)) (rsc/add-to-zip zip path (str/replace filename - shapes/sanitize-file-regex + plan/sanitize-file-regex "_")))] (p/let [beat-one (beats! :rendering (counter counter-kind 0 total) :force? true) _ (check-beat! beat-one) diff --git a/exporter/src/app/consumer/plan.cljs b/exporter/src/app/consumer/plan.cljs new file mode 100644 index 0000000000..dec8b7e714 --- /dev/null +++ b/exporter/src/app/consumer/plan.cljs @@ -0,0 +1,89 @@ +;; 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.plan + "The render plan of an assets run. + + The items the backend froze in the job become the renders this + process runs: file names built once and kept free of duplicates, then + grouped by scale and type, one render per group on the headless + engines and one DOM page per partition of fifty on the browser + engines. `app.consumer.exports` runs whatever this namespace plans, + and this namespace only plans." + (:require + [app.common.data :as d] + [app.renderer :as rd] + [app.util.mime :as mime] + [cuerdas.core :as str])) + +;; Regex to clean namefiles +(def sanitize-file-regex #"[\\/:*?\"<>|]") + +(defn count-objects + [exports] + (reduce + 0 (map (comp count :objects) exports))) + +(defn- assoc-file-name + "A transducer that assocs a candidate filename and avoid duplicates" + [] + (letfn [(find-candidate [params used] + (loop [index 0] + (let [candidate (str (:name params) + (:suffix params "") + (when (pos? index) + (str/concat "-" (inc index))) + (mime/get-extension (:type params)))] + (if (contains? used candidate) + (recur (inc index)) + candidate))))] + (fn [rf] + (let [used (volatile! #{})] + (fn + ([] (rf)) + ([result] (rf result)) + ([result params] + (let [candidate (find-candidate params @used) + params (assoc params :filename candidate)] + (vswap! used conj candidate) + (rf result params)))))))) + +(def ^:const ^:private + default-partition-size 50) + +(defn prepare-exports + [exports token is-wasm] + (letfn [(process-group [[part1 :as group]] + ;; Browser path renders one DOM page per partition; headless WASM + ;; takes the whole group in one render. + (if (rd/headless? {:is-wasm is-wasm :type (:type part1)}) + [(build-render group)] + (sequence (comp (partition-all default-partition-size) + (map build-render)) + group))) + + (build-render [[part1 :as part]] + {:file-id (:file-id part1) + :page-id (:page-id part1) + :share-id (:share-id part1) + :name (:name part1) + :token token + :type (:type part1) + :scale (:scale part1) + :objects (mapv part-entry->object part)}) + + (part-entry->object [entry] + {:id (:object-id entry) + :filename (:filename entry) + :name (:name entry) + :suffix (:suffix entry)})] + + (let [xform (comp + (map #(assoc % :token token)) + (assoc-file-name))] + (->> (sequence xform exports) + (d/group-by (juxt :scale :type)) + (map second) + (into [] (mapcat process-group)))))) diff --git a/exporter/src/app/core.cljs b/exporter/src/app/core.cljs index 05a2ee5a07..3d0bfe37a2 100644 --- a/exporter/src/app/core.cljs +++ b/exporter/src/app/core.cljs @@ -13,33 +13,19 @@ [app.config :as cf] [app.consumer.config :as ccfg] [app.consumer.worker :as worker] - [app.http :as http] - [app.jobs :as jobs] [app.jobs.utils :as job.utils] - [app.redis :as redis] [app.wasm :as wasm] [app.wasm.pool :as wasm.pool] [app.wasm.worker :as wasm.worker] - [cuerdas.core :as str] [promesa.core :as p])) (enable-console-print!) (l/setup! {:app :info}) -(defn- roles-log - [] - (->> (ccfg/roles) - (map name) - (str/join ","))) - (defn start "Render workers run this same bundle, so the thread decides what gets booted: - the http server and its pools, or one render worker. - - In the main thread the roles of the process decide the rest: `http` - boots the server and its pools exactly as always, `worker` boots the - consumer of the backend queue, and both can coexist during the - transition (the engines and the pools are shared)." + one render worker, or the single role of the main thread: consuming the + backend's export queue." [& _] (if-not ^boolean wt/isMainThread (wasm.worker/main) @@ -47,25 +33,16 @@ (l/info :msg "initializing" :public-uri (str (cf/get :public-uri)) :internal-uri (str (cf/get-internal-uri)) - :roles (roles-log) + :workers (ccfg/concurrency) :version (:full cf/version)) (l/info :msg "headless wasm export available" :wasm-dir wasm/artifact-dir - :workers (cf/get :wasm-worker-pool-max) :image-cache-size (cf/get :wasm-worker-image-cache-size)) (p/do (bwr/init) - (redis/init) - (jobs/init) (job.utils/init) (wasm.pool/init) - (when (ccfg/worker-enabled?) - (worker/start!)) - (when (contains? (ccfg/roles) :http) - (http/init)) - ;; With no role at all, the process boots into a no-op: the log - ;; above is the only surface of that mistake, and honest. - nil)))) + (worker/start!))))) (def main start) @@ -99,13 +76,10 @@ (p/do (shutdown-step "browser-pool" bwr/stop) (shutdown-step "wasm-worker-pool" wasm.pool/stop) - ;; the worker's pollers are inside their redis connections; ending - ;; redis first would leave the pops on broken connections - (when (ccfg/worker-enabled?) - (shutdown-step "consumer-worker" worker/stop!)) - (shutdown-step "redis" redis/stop) - (when (contains? (ccfg/roles) :http) - (shutdown-step "http" http/stop)) + ;; the pollers own their connections and may sit in a blocking + ;; pop for the whole poll timeout; draining the pools first lets + ;; the exports in flight unwind against their own resources + (shutdown-step "consumer-worker" worker/stop!) (done))))) (.on proc/default "uncaughtException" diff --git a/exporter/src/app/handlers.cljs b/exporter/src/app/handlers.cljs deleted file mode 100644 index bd2b453951..0000000000 --- a/exporter/src/app/handlers.cljs +++ /dev/null @@ -1,110 +0,0 @@ -;; 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.handlers - (:require - [app.auth :as auth] - [app.common.data :as d] - [app.common.logging :as l] - [app.common.spec :as us] - [app.handlers.export :as export] - [app.util.transit :as t] - [clojure.spec.alpha :as s] - [promesa.core :as p])) - -(l/set-level! :debug) - -(def ^:private error-codes - #{:queue-full}) - -(defn on-error - [error exchange] - (let [{:keys [type code] :as data} (ex-data error)] - (cond - (and (= :validation type) - (contains? error-codes code)) - (let [data {:type :validation - :code code - :hint (ex-message error)}] - (l/warn :hint "rejecting export request" :code code) - (-> exchange - (assoc :response/status 429) - (assoc :response/body (t/encode data)) - (assoc :response/headers {"content-type" "application/transit+json"}))) - - (= :authentication type) - (let [data {:type :authentication - :code code - :hint (ex-message error)}] - (-> exchange - (assoc :response/status 401) - (assoc :response/body (t/encode data)) - (assoc :response/headers {"content-type" "application/transit+json"}))) - - (or (= :validation type) - (= :assertion type)) - (let [explain (us/pretty-explain data) - data (-> data - (assoc :explain explain) - (assoc :type :validation) - (dissoc ::s/problems ::s/value ::s/spec))] - (-> exchange - (assoc :response/status 400) - (assoc :response/body (t/encode data)) - (assoc :response/headers {"content-type" "application/transit+json"}))) - - (= :not-found type) - (-> exchange - (assoc :response/status 404) - (assoc :response/body (t/encode data)) - (assoc :response/headers {"content-type" "application/transit+json"})) - - (and (= :internal type) - (= :browser-not-ready code)) - (let [data {:type :server-error - :code :internal - :hint (ex-message error) - :data data}] - (-> exchange - (assoc :response/status 503) - (assoc :response/body (t/encode data)) - (assoc :response/headers {"content-type" "application/transit+json"}))) - - :else - (let [data {:type :server-error - :code code - :hint (ex-message error) - :data data}] - (l/error :hint "unexpected internal error" :cause error) - (-> exchange - (assoc :response/status 500) - (assoc :response/body (t/encode (d/without-nils data))) - (assoc :response/headers {"content-type" "application/transit+json"})))))) - -(defn handler - "The original `POST /api/export` entry point, and the one the browser backend - still goes through. The export runs as soon as it is asked for, and the - contract is unchanged: `:wait` answers with the finished resource, otherwise - with the resource handle while the work runs." - [{:keys [:request/params :request/auth-token] :as exchange}] - (let [{:keys [cmd wait] :as params} (export/conform-params params)] - (l/debug :hint "process-request" :cmd cmd) - (->> (auth/resolve-profile-id auth-token) - (p/mcat (fn [profile-id] - ;; The session wins when there is one; the body value stays - ;; the fallback so nothing that used to work stops working. - (export/export! auth-token (cond-> params - (some? profile-id) - (assoc :profile-id profile-id))))) - (p/mcat (fn [{:keys [resource pending]}] - (if wait - (p/fmap (fn [resource] - (assoc exchange :response/body resource)) - pending) - (do - (p/merr (constantly nil) pending) - (p/resolved - (assoc exchange :response/body (dissoc resource :path)))))))))) diff --git a/exporter/src/app/handlers/export.cljs b/exporter/src/app/handlers/export.cljs deleted file mode 100644 index 8ba9a9c322..0000000000 --- a/exporter/src/app/handlers/export.cljs +++ /dev/null @@ -1,100 +0,0 @@ -;; 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.handlers.export - "Handle export jobs" - (:require - [app.common.spec :as us] - [app.handlers.export-frames :as export-frames] - [app.handlers.export-shapes :as export-shapes] - [app.jobs :as jobs] - [app.jobs.scheduler :as scheduler] - [app.jobs.utils :as job.utils] - [clojure.spec.alpha :as s] - [promesa.core :as p])) - -;; --- PARAMS - -(defmulti command-spec :cmd) - -(s/def ::cmd ::us/keyword) -(s/def ::wait ::us/boolean) - -(defmethod command-spec :export-shapes [_] ::export-shapes/params) -(defmethod command-spec :export-frames [_] ::export-frames/params) - -(s/def ::params - (s/and (s/keys :req-un [::cmd] - :opt-un [::wait]) - (s/multi-spec command-spec :cmd))) - -(defn conform-params - [params] - (us/conform ::params params)) - -(defn- prepare - [cmd auth-token params] - (case cmd - :export-shapes (export-shapes/prepare auth-token params) - :export-frames (export-frames/prepare auth-token params))) - -(defn- current - [job] - (or (jobs/lookup (:id job)) job)) - -(defn- run-and-track - [job run] - (->> (p/do (run job)) - (p/mcat (fn [resource] - (->> (jobs/complete! (current job) resource) - (p/fmap (constantly resource))))) - (p/merr (fn [cause] - (if (jobs/cancelled? (:id job)) - (p/rejected cause) - (->> (jobs/fail! (current job) cause) - (p/mcat (fn [_] (p/rejected cause))))))))) - -(defn- run-now! - "Runs the job as soon as it is created, outside the scheduler." - [job] - (->> (p/do (jobs/start! job)) - (p/mcat (fn [job] (p/do ((jobs/run-fn (:id job)) job)))) - (p/fnly (fn [_ _] - (jobs/release! (:id job)) - (job.utils/release! (:id job)))))) - -(defn- create! - [auth-token {:keys [cmd profile-id] :as params} start] - (let [{:keys [resource total headless run]} (prepare cmd auth-token params)] - (->> (jobs/create! {:profile-id profile-id - :cmd cmd - ;; Backend used for admission and cancel UI: headless - ;; WASM when `:is-wasm` is set, otherwise browser. - :backend (if headless "wasm" "browser") - :total total - :name (:name resource) - :resource-id (:id resource)} - (fn [job] (run-and-track job run))) - (p/fmap (fn [job] - (try - {:job job - :resource resource - :pending (start job)} - (catch :default cause - (jobs/fail! job cause) - (jobs/release! (:id job)) - (throw cause)))))))) - -(defn create-job! - "Returns a promise of `{:job :resource :pending}`." - [auth-token params] - (create! auth-token params scheduler/submit!)) - -(defn export! - "Like `create-job!`, but the work starts right away: This is what - keeps browser exports behaving exactly as they did before there were jobs." - [auth-token params] - (create! auth-token params run-now!)) diff --git a/exporter/src/app/handlers/export_frames.cljs b/exporter/src/app/handlers/export_frames.cljs deleted file mode 100644 index ed290d4b50..0000000000 --- a/exporter/src/app/handlers/export_frames.cljs +++ /dev/null @@ -1,93 +0,0 @@ -;; 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.handlers.export-frames - (:require - [app.common.spec :as us] - [app.handlers.export-shapes :refer [count-objects headless-exports? prepare-exports]] - [app.handlers.resources :as rsc] - [app.jobs :as jobs] - [app.jobs.utils :as job.utils] - [app.renderer :as rd] - [app.util.shell :as sh] - [cljs.spec.alpha :as s] - [cuerdas.core :as str] - [promesa.core :as p])) - -(declare ^:private join-pdf) -(declare ^:private move-file) - -(s/def ::name ::us/string) -(s/def ::file-id ::us/uuid) -(s/def ::page-id ::us/uuid) -(s/def ::object-id ::us/uuid) -(s/def ::is-wasm ::us/boolean) - -(s/def ::export - (s/keys :req-un [::file-id ::page-id ::object-id ::name])) - -(s/def ::exports - (s/every ::export :kind vector? :min-count 1)) - -(s/def ::params - (s/keys :req-un [::exports] - :opt-un [::name ::is-wasm])) - -(defn- run-export - [job auth-token resource {:keys [exports is-wasm file-id]}] - (let [rendered (atom []) - - on-object - (fn [{:keys [path] :as _object}] - (job.utils/track! (:id job) path) - (jobs/progress! job (count (swap! rendered conj path)))) - - exports - (map #(assoc % :is-wasm is-wasm :job-id (:id job)) exports)] - - (job.utils/track! (:id job) (:path resource)) - (->> (rd/with-scope exports - (fn [render] - (jobs/check-cancelled! job) - (->> exports - (map (fn [export] (render export on-object))) - (p/all)))) - (p/fmap (fn [_] @rendered)) - (p/mcat (partial join-pdf job file-id)) - (p/mcat (partial move-file resource)) - (p/fmap (constantly resource)) - (p/mcat (partial rsc/upload-resource auth-token)) - (p/mcat (fn [resource] - (->> (sh/stat (:path resource)) - (p/fmap #(merge resource %))))) - (p/fmap (fn [resource] (dissoc resource :path)))))) - -(defn prepare - [auth-token {:keys [exports name is-wasm] :as _params}] - (let [exports (-> (map #(assoc % :type :pdf :scale 1 :suffix "") exports) - (prepare-exports auth-token is-wasm)) - resource (rsc/create :pdf (or name (-> exports first :name))) - file-id (-> exports first :file-id)] - {:resource resource - :total (count-objects exports) - :headless (headless-exports? exports is-wasm) - :run (fn [job] (run-export job auth-token resource - {:exports exports - :is-wasm is-wasm - :file-id file-id}))})) - -(defn- join-pdf - [job file-id paths] - (p/let [prefix (str/concat "penpot.pdfunite." file-id ".") - path (job.utils/track! (:id job) (sh/tempfile :prefix prefix :suffix ".pdf"))] - (apply sh/run-cmd! "pdfunite" (conj (vec paths) path)) - path)) - -(defn- move-file - [{:keys [path] :as resource} output-path] - (p/do - (sh/move! output-path path) - resource)) diff --git a/exporter/src/app/handlers/export_shapes.cljs b/exporter/src/app/handlers/export_shapes.cljs deleted file mode 100644 index 27f5844e98..0000000000 --- a/exporter/src/app/handlers/export_shapes.cljs +++ /dev/null @@ -1,197 +0,0 @@ -;; 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.handlers.export-shapes - (:require - [app.common.data :as d] - [app.common.spec :as us] - [app.handlers.resources :as rsc] - [app.jobs :as jobs] - [app.jobs.utils :as job.utils] - [app.renderer :as rd] - [app.util.mime :as mime] - [app.util.shell :as sh] - [cljs.spec.alpha :as s] - [cuerdas.core :as str] - [promesa.core :as p])) - -(declare prepare-exports) - -;; Regex to clean namefiles -(def sanitize-file-regex #"[\\/:*?\"<>|]") - -(s/def ::file-id ::us/uuid) -(s/def ::filename ::us/string) -(s/def ::name ::us/string) -(s/def ::object-id ::us/uuid) -(s/def ::page-id ::us/uuid) -(s/def ::share-id ::us/uuid) -(s/def ::profile-id ::us/uuid) -(s/def ::scale ::us/number) -(s/def ::suffix ::us/string) -(s/def ::type ::us/keyword) -(s/def ::wait ::us/boolean) -(s/def ::is-wasm ::us/boolean) - -(s/def ::export - (s/keys :req-un [::page-id ::file-id ::object-id ::type ::suffix ::scale ::name] - :opt-un [::share-id])) - -(s/def ::exports - (s/coll-of ::export :kind vector? :min-count 1)) - -(s/def ::params - (s/keys :req-un [::exports ::profile-id] - :opt-un [::wait ::name ::skip-children ::force-multiple ::is-wasm])) - -(defn count-objects - [exports] - (reduce + 0 (map (comp count :objects) exports))) - -(defn- render! - [job export on-object] - (jobs/check-cancelled! job) - (rd/render (assoc export :job-id (:id job)) on-object)) - -(defn- scoped-renders - "Renders every export, the headless ones sharing a single worker." - [job exports on-object] - (rd/with-scope exports - (fn [render] - (jobs/check-cancelled! job) - (->> exports - (map (fn [export] (render export on-object))) - (p/all))))) - -(defn- run-single - [job auth-token resource {:keys [export is-wasm skip-children]}] - (job.utils/track! (:id job) (:path resource)) - (->> (render! job - (assoc export :skip-children skip-children :is-wasm (boolean is-wasm)) - (fn [{:keys [path] :as _object}] - (job.utils/track! (:id job) path) - (sh/move! path (:path resource)))) - (p/fmap (constantly resource)) - (p/mcat (partial rsc/upload-resource auth-token)) - (p/fmap (fn [resource] (dissoc resource :path))))) - -(defn- run-multiple - [job auth-token resource {:keys [exports is-wasm]}] - (let [failure (volatile! nil) - - zip (rsc/create-zip :resource resource - :on-error (fn [cause] (vreset! failure cause)) - :on-progress (fn [{:keys [done]}] - (jobs/progress! job done))) - - append (fn [{:keys [filename path] :as _object}] - (job.utils/track! (:id job) path) - (rsc/add-to-zip zip path (str/replace filename sanitize-file-regex "_")))] - - (job.utils/track! (:id job) (:path resource)) - (->> (scoped-renders job - (map #(assoc % :is-wasm (boolean is-wasm) :job-id (:id job)) exports) - append) - (p/mcat (fn [_] - (if-let [cause @failure] - (p/rejected cause) - (rsc/close-zip zip)))) - (p/fmap (constantly resource)) - (p/mcat (partial rsc/upload-resource auth-token)) - (p/fmap (fn [resource] (dissoc resource :path)))))) - -(defn headless-exports? - "Whether any of `exports` renders headless, and so whether the job leases a - render worker. Mirrors what `rd/with-scope` decides at run time." - [exports is-wasm] - (boolean (some #(rd/headless? {:is-wasm is-wasm :type (:type %)}) exports))) - -(defn prepare - [auth-token {:keys [exports force-multiple name skip-children is-wasm] :as _params}] - (let [exports (prepare-exports exports auth-token is-wasm) - headless? (headless-exports? exports is-wasm) - single? (and (not force-multiple) - (= 1 (count exports)) - (= 1 (count (-> exports first :objects))))] - (if single? - (let [export (first exports) - resource (rsc/create (:type export) (or name (:name export)))] - {:resource resource - :total 1 - :headless headless? - :run (fn [job] (run-single job auth-token resource - {:export export - :is-wasm is-wasm - :skip-children skip-children}))}) - - (let [resource (rsc/create :zip (or name (-> exports first :name)))] - {:resource resource - :total (count-objects exports) - :headless headless? - :run (fn [job] (run-multiple job auth-token resource - {:exports exports :is-wasm is-wasm}))})))) - -(defn- assoc-file-name - "A transducer that assocs a candidate filename and avoid duplicates" - [] - (letfn [(find-candidate [params used] - (loop [index 0] - (let [candidate (str (:name params) - (:suffix params "") - (when (pos? index) - (str/concat "-" (inc index))) - (mime/get-extension (:type params)))] - (if (contains? used candidate) - (recur (inc index)) - candidate))))] - (fn [rf] - (let [used (volatile! #{})] - (fn - ([] (rf)) - ([result] (rf result)) - ([result params] - (let [candidate (find-candidate params @used) - params (assoc params :filename candidate)] - (vswap! used conj candidate) - (rf result params)))))))) - -(def ^:const ^:private - default-partition-size 50) - -(defn prepare-exports - [exports token is-wasm] - (letfn [(process-group [[part1 :as group]] - ;; Browser path renders one DOM page per partition; headless WASM - ;; takes the whole group in one render. - (if (rd/headless? {:is-wasm is-wasm :type (:type part1)}) - [(build-render group)] - (sequence (comp (partition-all default-partition-size) - (map build-render)) - group))) - - (build-render [[part1 :as part]] - {:file-id (:file-id part1) - :page-id (:page-id part1) - :share-id (:share-id part1) - :name (:name part1) - :token token - :type (:type part1) - :scale (:scale part1) - :objects (mapv part-entry->object part)}) - - (part-entry->object [entry] - {:id (:object-id entry) - :filename (:filename entry) - :name (:name entry) - :suffix (:suffix entry)})] - - (let [xform (comp - (map #(assoc % :token token)) - (assoc-file-name))] - (->> (sequence xform exports) - (d/group-by (juxt :scale :type)) - (map second) - (into [] (mapcat process-group)))))) diff --git a/exporter/src/app/handlers/jobs.cljs b/exporter/src/app/handlers/jobs.cljs deleted file mode 100644 index 08c687a3d7..0000000000 --- a/exporter/src/app/handlers/jobs.cljs +++ /dev/null @@ -1,60 +0,0 @@ -;; 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.handlers.jobs - "REST surface for export jobs, under `/api/export/jobs`. - - Ownership always comes from the session (see `app.auth`), never from the - request body." - (:require - [app.auth :as auth] - [app.common.exceptions :as ex] - [app.common.logging :as l] - [app.handlers.export :as export] - [app.jobs :as jobs] - [promesa.core :as p])) - -(defn create - [{:keys [:request/auth-token :request/params] :as exchange}] - (->> (auth/require-profile-id auth-token) - (p/mcat (fn [profile-id] - (let [params (-> params - (assoc :profile-id profile-id) - (export/conform-params))] - (l/dbg :hint "create export job" :cmd (:cmd params) :profile-id (str profile-id)) - (export/create-job! auth-token params)))) - (p/fmap (fn [{:keys [job resource pending]}] - ;; A failure is reported through the job record, so the - ;; promise must not surface as an unhandled rejection. - (p/merr (constantly nil) pending) - (-> exchange - (assoc :response/body (-> (or (jobs/lookup (:id job)) job) - (assoc :filename (:filename resource)) - (assoc :mtype (:mtype resource))))))))) - -(defn fetch - [{:keys [:request/auth-token] :as exchange} job-id] - (->> (auth/require-profile-id auth-token) - (p/mcat (fn [profile-id] - (->> (jobs/fetch job-id) - (p/fmap #(auth/check-owner! % profile-id))))) - (p/fmap (fn [job] - (assoc exchange :response/body job))))) - -(defn cancel - [{:keys [:request/auth-token] :as exchange} job-id] - (->> (auth/require-profile-id auth-token) - (p/mcat (fn [profile-id] - (->> (jobs/fetch job-id) - (p/fmap #(auth/check-owner! % profile-id))))) - (p/mcat (fn [job] (jobs/cancel! (:id job)))) - (p/mcat (fn [_] (jobs/fetch job-id))) - (p/fmap (fn [job] - (if job - (assoc exchange :response/body job) - (ex/raise :type :not-found - :code :object-not-found - :hint "job does not exist")))))) diff --git a/exporter/src/app/handlers/resources.cljs b/exporter/src/app/handlers/resources.cljs index d703821f00..9e719e2701 100644 --- a/exporter/src/app/handlers/resources.cljs +++ b/exporter/src/app/handlers/resources.cljs @@ -5,18 +5,14 @@ ;; Copyright (c) KALEIDOS SUBSIDIARY SL (ns app.handlers.resources - "Temporal resources management." + "The run's local resources: the artifact files the runner parks in + the temp area until the settle moves them into the job's completion, + zipped once a run packs more than one." (:require ["archiver" :as arc] ["node:fs" :as fs] - ["node:fs/promises" :as fsp] ["node:path" :as path] - ["undici" :as http] - [app.common.exceptions :as ex] - [app.common.transit :as t] - [app.common.uri :as u] [app.common.uuid :as uuid] - [app.config :as cf] [app.util.mime :as mime] [app.util.shell :as sh] [cljs.core :as c] @@ -63,36 +59,3 @@ (p/create (fn [resolve] (.on ^js zip "close" resolve) (.finalize ^js zip)))) - -(defn upload-resource - [auth-token resource] - (->> (fsp/readFile (:path resource)) - (p/fmap (fn [buffer] - (new js/Blob #js [buffer] #js {:type (:mtype resource)}))) - (p/mcat (fn [blob] - (let [fdata (new http/FormData) - agent (new http/Agent #js {:connect #js {:rejectUnauthorized false}}) - headers #js {"X-Shared-Key" (str "exporter " cf/management-key) - "Authorization" (str "Bearer " auth-token)} - - request #js {:headers headers - :method "POST" - :body fdata - :dispatcher agent} - uri (-> (cf/get-internal-uri) - (u/ensure-path-slash) - (u/join "api/management/methods/upload-tempfile") - (str))] - - (.append fdata "content" blob (:filename resource)) - (http/fetch uri request)))) - - (p/mcat (fn [response] - (if (not= (.-status response) 200) - (ex/raise :type :internal - :code :unable-to-upload-resource - :response-status (.-status response)) - (.text response)))) - (p/fmap t/decode-str) - (p/fmap (fn [result] - (merge resource (dissoc result :id)))))) diff --git a/exporter/src/app/http.cljs b/exporter/src/app/http.cljs deleted file mode 100644 index 93100885f0..0000000000 --- a/exporter/src/app/http.cljs +++ /dev/null @@ -1,189 +0,0 @@ -;; 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.http - (:require - ["cookies$default" :as Cookies] - ["inflation$default" :as inflate] - ["node:http" :as http] - ["node:stream$default" :as stream] - ["raw-body$default" :as raw-body] - [app.common.logging :as l] - [app.common.transit :as t] - [app.config :as cf] - [app.handlers :as handlers] - [app.router :as router] - [cuerdas.core :as str] - [lambdaisland.uri :as u] - [promesa.core :as p])) - -(l/set-level! :info) - -(defprotocol IStreamableResponseBody - (write-body! [_ response])) - -(extend-protocol IStreamableResponseBody - string - (write-body! [data response] - (.write ^js response data) - (.end ^js response)) - - js/Buffer - (write-body! [data response] - (.write ^js response data) - (.end ^js response)) - - stream/Stream - (write-body! [data response] - (.pipe ^js data response) - (.on ^js data "error" (fn [cause] - (js/console.error cause) - (.end response))))) - -(defn- handle-response - [{:keys [:response/body - :response/headers - :response/status - response] - :as exchange}] - (let [status (or status 200) - headers (clj->js headers) - body (or body "")] - (.writeHead ^js response status headers) - (write-body! body response))) - -(defn- parse-headers - [req] - (let [orig (unchecked-get req "headers")] - (persistent! - (reduce #(assoc! %1 (str/lower %2) (unchecked-get orig %2)) - (transient {}) - (js/Object.keys orig))))) - -(defn- wrap-body-params - [handler] - (let [opts #js {:limit "60mb" :encoding "utf8"}] - (fn [{:keys [:request/method :request/headers request] :as exchange}] - (let [ctype (get headers "content-type")] - (if (= method "post") - (-> (raw-body (inflate request) opts) - (p/then (fn [data] - (cond-> data - (= ctype "application/transit+json") - (t/decode-str)))) - (p/then (fn [data] - (handler (assoc exchange :request/body-params data))))) - (handler exchange)))))) - -(defn- wrap-params - [handler] - (fn [{:keys [:request/body-params :request/query-params] :as exchange}] - (handler (assoc exchange :request/params (merge query-params body-params))))) - -(defn- wrap-response-format - [handler] - (fn [exchange] - (p/then - (handler exchange) - (fn [{:keys [:response/body :response/status] :as exchange}] - (cond - (map? body) - (let [data (t/encode-str body {:type :json-verbose}) - size (js/Buffer.byteLength data "utf-8")] - (-> exchange - (assoc :response/body data) - (assoc :response/status (or status 200)) - (update :response/headers assoc "content-type" "application/transit+json") - (update :response/headers assoc "content-length" size))) - - (and (nil? body) - (= 200 status)) - (-> exchange - (assoc :response/body "") - (assoc :response/status 204) - (assoc :response/headers {"content-length" 0})) - - :else - exchange))))) - -(defn- wrap-query-params - [handler] - (fn [{:keys [:request/uri] :as exchange}] - (handler (assoc exchange :request/query-params (u/query-string->map (:query uri)))))) - -(defn- wrap-error - [handler on-error] - (fn [exchange] - (-> (p/do (handler exchange)) - (p/catch (fn [cause] (on-error cause exchange)))))) - -(defn- wrap-auth - [handler cookie-name] - (fn [{:keys [:request/cookies] :as exchange}] - (let [token (.get ^js cookies cookie-name)] - (handler (cond-> exchange token (assoc :request/auth-token token)))))) - -(defn- wrap-health - "Add /healthz entry point intercept." - [handler] - (fn [{:keys [:request/path] :as exchange}] - (if (= path "/readyz") - (assoc exchange - :response/status 200 - :response/body "OK") - (handler exchange)))) - -(defn- create-adapter - [handler] - (fn [req res] - (let [cookies (Cookies. req res) - headers (parse-headers req) - uri (u/uri (unchecked-get req "url")) - exchange {:request/method (str/lower (unchecked-get req "method")) - :request/path (:path uri) - :request/uri uri - :request/headers headers - :request/cookies cookies - :request req - :response res}] - (-> (p/do (handler exchange)) - (p/then handle-response))))) - -(defn- create-server - [handler] - (.createServer ^js http (create-adapter handler))) - -(def instance (atom nil)) - -(defn init - [] - (let [handler (-> (router/create handlers/handler) - (wrap-health) - (wrap-auth "auth-token") - (wrap-response-format) - (wrap-params) - (wrap-query-params) - (wrap-body-params) - (wrap-error handlers/on-error)) - server (create-server handler) - port (cf/get :http-server-port 6061)] - - (.listen server port) - (l/info :hint "welcome to penpot" - :module "exporter" - :flags cf/flags - :version (:full cf/version)) - (l/info :hint "starting http server" :port port) - (reset! instance server))) - -(defn stop - [] - (if-let [server @instance] - (p/create (fn [resolve] - (.close server (fn [] - (l/info :hint "shutdown http server") - (resolve))))) - (p/resolved nil))) diff --git a/exporter/src/app/jobs.cljs b/exporter/src/app/jobs.cljs index 5f38d86171..f304848b23 100644 --- a/exporter/src/app/jobs.cljs +++ b/exporter/src/app/jobs.cljs @@ -5,30 +5,17 @@ ;; Copyright (c) KALEIDOS SUBSIDIARY SL (ns app.jobs - "Export job model and lifecycle. + "The local arm of the job life this process runs. - 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." + The durable row of a job is the backend's business: the worker claims + it, beats against it and settles it through the management API. This + namespace keeps only the runtime bits that cannot live in a row -- the + cancel callbacks that terminate a render worker, the cancel signal + shared with that render worker, and the flag the parts of a run read + between renders -- keyed by the job id of the claimed job." (: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) + [app.common.time :as ct])) (def ^:private terminal-states #{"ended" "error" "cancelled"}) @@ -42,77 +29,16 @@ [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! (redis/->tenant-key (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 cancelled? + "True when the job has been marked cancelled in this process. The + check between objects reads it; inside a render the thread reads the + signal instead." + [job-id] + (boolean (:cancelled? (runtime job-id)))) (defn cancel-signal "Int32Array over a SharedArrayBuffer, readable from a worker thread: 0 while @@ -139,97 +65,11 @@ (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." + "Drops the local runtime entry once the job settled. The row stays in + the backend until it is read once more or expires." [job-id] (swap! registry dissoc (str job-id))) -(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 register! "The runtime record of a job this process runs. The local cancel arm reads it (`mark-cancelled`), so the in-flight render of a claimed job @@ -240,89 +80,35 @@ {:job {:id job-id :state "running" :total total} :cancelled? false})) +(defn- check-cancel-callback! + [job-id f] + (try + (f) + (catch :default cause + (l/warn :hint "error on job cancel callback" :job-id (str job-id) :cause cause)))) + (defn mark-cancelled "Signals one job of this process as cancelled: raises the cancel flag - (the check between objects sees it), arms the shared-array signal (the - wasm worker sees it inside a render) and runs the cancel callbacks - (the terminate of the render worker). The durable row is already - terminal in the backend; no store write, no publish happens here. + (the check between objects sees it), marks the local record terminal + (the row of the job is already terminal in the backend), arms the + shared-array signal (the wasm worker sees it inside a render) and + runs the cancel callbacks (the terminate of the render worker). The + mark happens before the callbacks run: one of them releases the job, + and a second mark for the same job must change nothing. A job this process does not own, or a settled one, is a no-op." [job-id] (let [k (str job-id) rt (get @registry k)] - (when-let [rt (and rt (:job rt))] - (when-not (terminal? rt) - (swap! registry update k (fn [rt] (some-> rt (assoc :cancelled? true)))) - (when-let [signal (:cancel-signal rt)] - (js/Atomics.store signal 0 1)) - (doseq [f (:cancel-fns rt)] - (try - (f) - (catch :default cause - (l/warn :hint "error on job cancel callback" :job-id k :cause cause)))) + (when-let [_job (and rt (:job rt))] + (when-not (terminal? (:job rt)) + (let [callbacks (:cancel-fns rt)] + (swap! registry update k + (fn [e] (some-> e + (assoc :cancelled? true) + (update :job assoc :state "cancelled" :ended-at (now-ms))))) + (when-let [signal (:cancel-signal (get @registry k))] + (js/Atomics.store signal 0 1)) + (doseq [f callbacks] + (check-cancel-callback! job-id f))) true)))) - -(defn mark-cancelled! - "Marks a job of this process as cancelled and runs its cancel - callbacks (what terminates the render worker a hard-cancel needs) — - without touching the shared store. - - It is the local arm of the manual `jobs/cancel-local!` has: the jobs - substrate marks the row `cancelled` in the backend, and the worker's - own records only need to know, so the in-flight render dies." - [job-id] - (cancel-local! job-id)) - -(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 deleted file mode 100644 index 320689d90d..0000000000 --- a/exporter/src/app/jobs/scheduler.cljs +++ /dev/null @@ -1,149 +0,0 @@ -;; 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.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, - as are headless jobs once every render worker is busy." - (: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] - [app.wasm.pool :as pool] - [promesa.core :as p])) - -(l/set-level! :debug) - -(defonce ^:private state - (atom {:running {} ;; job-id -> {:profile-id :headless?} - :queue []})) ;; vector of {:job :resolve :reject} - -(defn- max-concurrent [] (cf/get :exporter-max-concurrent-jobs 4)) -(defn- max-per-profile [] (cf/get :exporter-max-jobs-per-profile 2)) -(defn- max-queued [] (cf/get :exporter-queue-max 64)) - -(defn- headless? - [job] - (= "wasm" (:backend job))) - -(defn- running-for - [{:keys [running]} profile-id] - (count (filter #(= profile-id (:profile-id %)) (vals running)))) - -(defn- running-headless - [{:keys [running]}] - (count (filter :headless? (vals running)))) - -(defn- eligible? - [state job] - (and (< (count (:running state)) (max-concurrent)) - (< (running-for state (:profile-id job)) (max-per-profile)) - ;; A headless job holds one render worker for its whole run, so admitting - ;; more of them than there are workers would only move the wait inside - ;; the pool, with the job already reporting itself as running. - (or (not (headless? job)) - (< (running-headless state) (pool/capacity))))) - -(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 profile-id - :headless? (headless? job)}) - (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 (:job entry)) - 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 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 deleted file mode 100644 index 3cfb27b951..0000000000 --- a/exporter/src/app/jobs/store.cljs +++ /dev/null @@ -1,87 +0,0 @@ -;; 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.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 (redis/->key "job-cancel")) - -(defn- job-key - [job-id] - (redis/->key "job." job-id)) - -(defn- ttl - [] - (cf/get :exporter-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 "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/redis.cljs b/exporter/src/app/redis.cljs deleted file mode 100644 index 8fca0c335c..0000000000 --- a/exporter/src/app/redis.cljs +++ /dev/null @@ -1,177 +0,0 @@ -;; 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.redis - (: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] - [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 role] - (let [^js client (new redis/default uri)] - (.on client "connect" - (fn [] (l/info :hint "redis connection established" :uri uri :role role))) - (.on client "error" - (fn [cause] (l/error :hint "error on redis connection" :role role :cause cause))) - (.on client "close" - (fn [] (l/warn :hint "connection closed" :role role))) - (.on client "reconnect" - (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" :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 - [] - (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 ->tenant-key - "Namespaces `parts` under the tenant, the prefix the backend msgbus uses." - [& parts] - (dm/str tenant "." (apply str parts))) - -(defn ->key - "Namespaces `parts` under the exporter, inside the tenant." - [& parts] - (dm/str "penpot.exporter." tenant "." (apply str parts))) - -(defn pub! - "Publishes on `topic`, which must already be namespaced." - [topic payload] - (let [payload (if (map? payload) (t/encode-str payload) payload)] - (when-let [client @client] - (.publish ^js client topic payload)))) - -(defn sub! - "Subscribes `handler` (fn of the raw payload string) to `topic`, which must - already be namespaced. Returns a 0-arg fn that removes this handler." - [topic handler] - (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. - - Batches are accumulated in memory rather than consumed as a stream, which a - promise-returning fn cannot express. Fine for the job keyspace, but reading - redis wants a streaming or reactive interface before it is used for more." - [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/src/app/router.cljs b/exporter/src/app/router.cljs deleted file mode 100644 index 7b763edc8a..0000000000 --- a/exporter/src/app/router.cljs +++ /dev/null @@ -1,61 +0,0 @@ -;; 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.router - "Method + path dispatch. - - Requests arrive with whatever prefix the proxy in front uses (`/api/export` - in devenv, `/` when talking to the process directly), so routes are matched - on the remainder after that prefix." - (:require - [app.common.exceptions :as ex] - [app.handlers.jobs :as jobs.handlers] - [cuerdas.core :as str])) - -(def ^:private mount-point "/api/export") - -(defn- route-path - [path] - (let [path (or path "/") - path (if (str/starts-with? path mount-point) - (subs path (count mount-point)) - path) - path (str/rtrim path "/")] - (if (str/empty? path) "/" path))) - -(defn- job-id - [path prefix] - (let [id (subs path (count prefix))] - (when-not (or (str/empty? id) (str/includes? id "/")) - id))) - -(defn create - "Builds the request handler. `legacy-handler` serves the original - `POST /api/export` command multiplex." - [legacy-handler] - (fn [{:keys [:request/method :request/path] :as exchange}] - (let [path (route-path path)] - (cond - (and (= "post" method) (= "/" path)) - (legacy-handler exchange) - - (and (= "post" method) (= "/jobs" path)) - (jobs.handlers/create exchange) - - (and (= "get" method) (str/starts-with? path "/jobs/")) - (if-let [id (job-id path "/jobs/")] - (jobs.handlers/fetch exchange id) - (ex/raise :type :not-found :code :object-not-found :hint "unknown route")) - - (and (= "delete" method) (str/starts-with? path "/jobs/")) - (if-let [id (job-id path "/jobs/")] - (jobs.handlers/cancel exchange id) - (ex/raise :type :not-found :code :object-not-found :hint "unknown route")) - - :else - (ex/raise :type :not-found - :code :route-not-found - :hint (str "no route for " method " " path)))))) diff --git a/exporter/src/app/wasm/pool.cljs b/exporter/src/app/wasm/pool.cljs index 31a20527e7..482cb7a0c4 100644 --- a/exporter/src/app/wasm/pool.cljs +++ b/exporter/src/app/wasm/pool.cljs @@ -27,6 +27,7 @@ [app.common.logging :as l] [app.common.transit :as t] [app.config :as cf] + [app.consumer.config :as ccfg] [promesa.core :as p])) (l/set-level! :info) @@ -102,26 +103,23 @@ (p/resolved (true? (unchecked-get worker "__alive"))))}) (defn capacity - "How many renders can run at once, and so how many headless jobs the - scheduler may admit." + "How many renders can run at once: the pool holds as many render + workers as concurrent jobs the process runs, which is what the + pollers count. Clamped rather than rejected: a bad value should not + stop the exporter from booting, and a headless render has no other + backend to fall back to." [] - ;; Clamped rather than rejected: a bad value should not stop the exporter - ;; from booting, and a headless render has no other backend to fall back to. - (max 1 (cf/get :wasm-worker-pool-max 2))) + (max 1 (ccfg/concurrency))) (defn init [] - (let [configured (cf/get :wasm-worker-pool-max 2) - max-workers (capacity) + (let [max-workers (capacity) opts #js {:max max-workers :min (min max-workers (cf/get :wasm-worker-pool-min 1)) :testOnBorrow true :evictionRunIntervalMillis 30000 :numTestsPerEvictionRun 2 :idleTimeoutMillis 300000}] - (when (not= configured max-workers) - (l/warn :hint "wasm-worker-pool-max raised to the minimum of one" - :configured configured)) (l/info :hint "initializing render worker pool" :opts opts) (reset! pool (gp/createPool worker-pool-factory opts)) (p/resolved nil))) diff --git a/exporter/test/exporter_tests/consumer_config_test.cljs b/exporter/test/exporter_tests/consumer_config_test.cljs index a71ce17a01..0236769242 100644 --- a/exporter/test/exporter_tests/consumer_config_test.cljs +++ b/exporter/test/exporter_tests/consumer_config_test.cljs @@ -5,29 +5,11 @@ ;; 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." + "The consumer side of the config: how many jobs run at once and + which queue they come from." (: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 serves the worker role only: the exporter is - a consumer's process, no surface needs preserving" - (t/is (= #{:worker} (ccfg/roles nil))) - (t/is (= #{:worker} (ccfg/roles)))) - (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})))) + [cljs.test :as t :include-macros true])) (t/deftest concurrency-is-at-least-one (t/testing "the configured value is taken as given" diff --git a/exporter/test/exporter_tests/export_shapes_test.cljs b/exporter/test/exporter_tests/consumer_plan_test.cljs similarity index 83% rename from exporter/test/exporter_tests/export_shapes_test.cljs rename to exporter/test/exporter_tests/consumer_plan_test.cljs index 23a682a295..e5fedb7185 100644 --- a/exporter/test/exporter_tests/export_shapes_test.cljs +++ b/exporter/test/exporter_tests/consumer_plan_test.cljs @@ -4,11 +4,11 @@ ;; ;; Copyright (c) KALEIDOS SUBSIDIARY SL -(ns exporter-tests.export-shapes-test +(ns exporter-tests.consumer-plan-test "Chunking of the browser backend." (:require [app.common.uuid :as uuid] - [app.handlers.export-shapes :as export-shapes] + [app.consumer.plan :as plan] [cljs.test :as t :include-macros true])) (defn- exports @@ -26,6 +26,6 @@ (range n)))) (t/deftest browser-exports-are-chunked - (let [parts (export-shapes/prepare-exports (exports 120 :png 1) "token" false)] + (let [parts (plan/prepare-exports (exports 120 :png 1) "token" false)] (t/is (= 3 (count parts))) (t/is (= [50 50 20] (mapv (comp count :objects) parts))))) diff --git a/exporter/test/exporter_tests/jobs_test.cljs b/exporter/test/exporter_tests/jobs_test.cljs index c8baa9b9fd..7d9c35a31a 100644 --- a/exporter/test/exporter_tests/jobs_test.cljs +++ b/exporter/test/exporter_tests/jobs_test.cljs @@ -5,82 +5,85 @@ ;; Copyright (c) KALEIDOS SUBSIDIARY 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." + "The local arm of the claimed job: the flag and the pieces a hard + cancel needs. The row of the job is the backend's business; nothing + here touches it." (:require [app.common.uuid :as uuid] [app.jobs :as jobs] - [cljs.test :as t :include-macros true] - [promesa.core :as p])) + [cljs.test :as t :include-macros true])) -(defn- create! +(defn- register! [] - (jobs/create! {:profile-id (uuid/next) - :cmd :export-shapes - :backend "wasm" - :total 10 - :name "test" - :resource-id (uuid/next)} - (constantly (p/resolved nil)))) + (let [job-id (uuid/next)] + (jobs/register! job-id) + job-id)) -(t/deftest progress-does-not-resurrect-a-finished-job - (t/testing "a render reporting after the export failed cannot undo the failure" +(t/deftest mark-cancelled-runs-the-cancel-pieces-once + (t/testing "the signal arms, the callbacks run, and a second mark is a no-op" (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)) + (let [job-id (register!) + seen (atom [])] + (jobs/on-cancel job-id (fn [] (swap! seen conj :first))) + (jobs/on-cancel job-id (fn [] (swap! seen conj :second))) + + (t/is (true? (jobs/mark-cancelled job-id))) + (t/is (true? (jobs/cancelled? job-id))) + (t/is (some? (jobs/cancel-signal job-id))) + ;; the signal is armed: a render thread reading it stops + (let [^js signal (jobs/cancel-signal job-id)] + (t/is (= 1 (js/Atomics.load signal 0)))) + + (t/is (= [:first :second] @seen)) + + ;; a settled job is a no-op: nothing marked, nobody new called + (t/is (nil? (jobs/mark-cancelled job-id))) + (t/is (= [:first :second] @seen)) (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 the-callbacks-see-the-cancelled-flag + ;; a callback that stops the in-flight render reads the flag first: the + ;; terminate of the worker is what a hard-cancel is for, and the pieces + ;; below it must not race the writer + (t/async done + (let [job-id (register!) + seen (atom ::not-called)] + (jobs/on-cancel job-id (fn [] (reset! seen (jobs/cancelled? job-id)))) + (jobs/mark-cancelled job-id) + (t/is (true? @seen)) + (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/deftest a-signal-created-after-the-mark-is-armed + (t/testing "a job cancelled before its render leased a worker is + already stoppable for the render that comes next" (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)))))) + (let [job-id (register!)] + (jobs/mark-cancelled job-id) + (let [^js signal (jobs/cancel-signal job-id)] + (t/is (some? signal)) + (t/is (= 1 (js/Atomics.load signal 0)))) + (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/testing "a late cancel for a settled job changes nothing, and + release drops every piece" (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)))) + (let [job-id (register!) + seen (atom ::not-called)] + (jobs/on-cancel job-id (fn [] (reset! seen :callback-ran))) + (jobs/release! job-id) + + (t/is (nil? (jobs/mark-cancelled job-id))) + (t/is (= ::not-called @seen)) + (t/is (false? (jobs/cancelled? job-id))) + (t/is (nil? (jobs/cancel-signal job-id))) + (done))))) + +(t/deftest an-unowned-job-is-a-no-op + (t/testing "mark and pieces for a job this process never registered" + (t/async done + (let [job-id (uuid/next)] + (t/is (nil? (jobs/mark-cancelled job-id))) + (t/is (false? (jobs/cancelled? job-id))) + (t/is (nil? (jobs/cancel-signal job-id))) (done))))) diff --git a/exporter/test/exporter_tests/runner.cljs b/exporter/test/exporter_tests/runner.cljs index 187c02ed46..69a7e0c68d 100644 --- a/exporter/test/exporter_tests/runner.cljs +++ b/exporter/test/exporter_tests/runner.cljs @@ -12,10 +12,9 @@ [clojure.tools.cli :refer [parse-opts]] [exporter-tests.api-test] [exporter-tests.consumer-config-test] - [exporter-tests.export-shapes-test] + [exporter-tests.consumer-plan-test] [exporter-tests.jobs-test] [exporter-tests.renderer-svg-test] - [exporter-tests.scheduler-test] [exporter-tests.shell-test] [exporter-tests.wasm-pool-test] [exporter-tests.worker-test] @@ -31,13 +30,12 @@ (println (str "UNHANDLED-REJECTION: " reason)))) (def test-namespaces - ['exporter-tests.export-shapes-test + ['exporter-tests.consumer-plan-test 'exporter-tests.jobs-test 'exporter-tests.api-test 'exporter-tests.worker-test 'exporter-tests.consumer-config-test 'exporter-tests.renderer-svg-test - 'exporter-tests.scheduler-test 'exporter-tests.shell-test 'exporter-tests.wasm-pool-test]) diff --git a/exporter/test/exporter_tests/scheduler_test.cljs b/exporter/test/exporter_tests/scheduler_test.cljs deleted file mode 100644 index 04349480bf..0000000000 --- a/exporter/test/exporter_tests/scheduler_test.cljs +++ /dev/null @@ -1,81 +0,0 @@ -;; 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.scheduler-test - "Admission control. A headless job leases one render worker for its whole run, - so no more of them may start than there are workers." - (:require - [app.common.uuid :as uuid] - [app.jobs :as jobs] - [app.jobs.scheduler :as scheduler] - [app.wasm.pool :as pool] - [cljs.test :as t :include-macros true] - [promesa.core :as p])) - -(defn- pin-capacity! - "Fixes the worker count for the test. Returns the thunk that puts it back; - `with-redefs` cannot be used here, since the scheduler keeps admitting jobs - after the body of the test has returned." - [n] - (let [original pool/capacity] - (set! pool/capacity (constantly n)) - (fn [] (set! pool/capacity original)))) - -(defn- create! - [backend run-fn] - (jobs/create! {:profile-id (uuid/next) - :cmd :export-shapes - :backend backend - :total 1 - :name "test" - :resource-id (uuid/next)} - run-fn)) - -(defn- gate - "A promise and the fn that settles it, standing in for a render in flight." - [] - (let [resolve* (volatile! nil) - pending (p/create (fn [resolve _] (vreset! resolve* resolve)))] - [pending (fn [] (@resolve* nil))])) - -(t/deftest headless-jobs-wait-for-a-render-worker - (t/testing "with one worker, the second headless job stays queued until the first ends" - (t/async done - (let [restore! (pin-capacity! 1) - [render open] (gate) - started (atom [])] - (p/let [job1 (create! "wasm" (fn [_] (swap! started conj :one) render)) - job2 (create! "wasm" (fn [_] (swap! started conj :two) (p/resolved nil)))] - (let [p1 (scheduler/submit! job1) - p2 (scheduler/submit! job2)] - (p/do - (p/delay 10) - (t/is (= [:one] @started)) - (t/is (= "running" (:state (jobs/lookup (:id job1))))) - (t/is (= "queued" (:state (jobs/lookup (:id job2))))) - (open) - (p/all [p1 p2]) - (t/is (= [:one :two] @started)) - (restore!) - (done)))))))) - -(t/deftest a-browser-job-is-not-held-back-by-the-worker-pool - (t/testing "the headless cap applies to headless jobs only" - (t/async done - (let [restore! (pin-capacity! 1) - [render open] (gate) - started (atom [])] - (p/let [job1 (create! "wasm" (fn [_] (swap! started conj :wasm) render)) - job2 (create! "browser" (fn [_] (swap! started conj :browser) (p/resolved nil)))] - (let [p1 (scheduler/submit! job1) - p2 (scheduler/submit! job2)] - (p/do - (p/delay 10) - (t/is (= [:wasm :browser] @started)) - (open) - (p/all [p1 p2]) - (restore!) - (done))))))))