github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-29 05:50:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

🐛 Validate and decode a job result with its declared schema (#12350)

Browse Source 
* 🐛 Decode job results with the schema of the job-def

A job result is stored as plain JSON, so a uuid comes back as the text
it is on disk and a uuid used as a map key comes back as a keyword. The
import resolution is a map keyed by file id, and the wizard lost that
identity: choosing a library in the link step created no link at all,
with no error and with the modal closing as if it had worked.

A job-def can now declare a result schema and its precompiled decoder,
the pair the params already have, and the substrate applies them when the
row is read. A def that declares neither answers with the JSON the row
holds, as before; a row whose job-def is gone is read untouched instead
of raising.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Type the import result so its resolution keeps uuid keys

The resolution of a version 3 import is a map keyed by the new file id,
and the result column is plain JSON: without a schema of its own every
key came back as a keyword while the :id of the entry came back as text,
so the dashboard wizard could not match the file it was linking and
created no library link at all.

Declare the shape of the resolution (files, done and pending entries and
their candidates) and give the job-def its result schema and decoder.
The check before the result is stored keeps validating the same schema.

Closes #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Take the imported file id from the entry, not the map key

The wizard looked a file up in the selection with the key of the
resolution map while it had indexed the selection by the :id of the
entry. The key is keywordized on its way through the job result, so the
lookup missed and the confirm button created no library link at all,
with no error and with the modal closing as if it had worked.

Take the id from the entry, the source the rest of the wizard already
uses, and move the decision to a pure function covered by tests of both
shapes of the resolution.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Test that an import result keeps its ids as uuids

The regression this branch fixes is only visible through the read path
of the substrate: the resolution of an import is a map keyed by the new
file id, and a test that works with the value in memory never sees the
keys coming back as keywords from the plain JSON column.

Import a package whose library needs a replacement chosen by the user,
read the result with `jobs/get-job` as the client does, and check the
keys, the file ids and the ids of the candidates. Without the result
decoder of the import job-def the test answers with text ids.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 📚 Document the result schema of a job-def

The memories and the docstring of the substrate said a job result has no
schema of its own and comes back as the JSON the row holds. A job-def can
now declare its result schema and decoder, and the substrate applies them
when the row is read.

Keep the rule next to it: a result map keyed by uuid needs its schema
declared, or every key comes back as a keyword, and the link-later step
of the import wizard takes the file id from the entry and never from the
key of the resolution map.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 📚 Fix the documented contract of a job result

The docstring of get-job said a result has no schema of its own and comes
back as the JSON the row holds, which is what this branch changes: the
result is decoded with the schema its job-def declares, and a result that
does not conform never reaches the column because the write validates it.

Say the same in the management command that stores the result of an
external worker, the other writer of the column.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Check the result of a job against its schema before storing it

The result of a job crosses a plain JSON column, so a uuid comes back as
text and a uuid used as a map key as a keyword. The job-def now declares
the schema of its result and the substrate compiles the decoder from it,
which is what the read path uses to get the types back.

The same schema is the contract of the write path: complete refuses a
result that does not conform, with the malli explanation, so nothing
invalid reaches the column, and the answer is the same for the runner and
for the external worker that completes a job through the management RPC.
A job that is no longer completable is skipped without checking, and the
read path decodes only: a row written by another version stays readable.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Check the result of an import inside its transaction

The check ran after the transaction of the import had committed, so a
result the schema does not accept failed the job with the files already
created in the destination project: a half imported project and an error.
The core decides whether the whole import is kept or rolled back, so the
check belongs inside its transaction, which is what turns a contract
violation into a rollback.

The substrate checks the result again when the job is completed, which is
the only net for a job that does not validate its own result.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 🐛 Cover the edges of an import result read by the client

The shape of a result with no libraries to resolve, and the one of a
version 1 import, is the most common in production and neither was covered
by the tests that read a result the way the client does.

Cover both through jobs/get-job, and the case of a registry that no longer
knows the kind of the row: it answers with the JSON of the column,
keywordized keys and uuid as text, instead of raising. Say in the two tests
that depend on the decoder what implementation without it makes them fail.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* 📚 Fix the prose and the fixtures of the result contract

The result schema of a job-def is what the read decodes with and what the
write checks: say that in the memories, with the rule that any uuid of a
result needs it, whether a value or a map key.

The motive of the wizard helper was a fact this branch makes false: what a
key of the resolution map holds depends on the result schema of the job-def
that wrote it, so the rule is to read the id from the entry. Same
correction in its memory and in the test that pins the shape.

The candidates of a pending entry match by slugified name, not by name, and
every test row now states the kind of its job, because the name is what
resolves its job-def when the result is read or checked.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* ♻️ Keep the wizard link helper private and cover its shape

The helper that decides which links the wizard creates belongs to its
namespace, as its neighbours do: reach it from the test with @#', the
pattern the frontend tests already use for a private helper.

Cover the shapes it had no test for: a file with two libraries pending
resolution, a file whose libraries were auto-linked, and a selection where
only one of two entries was resolved.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

* ♻️ Share one job-row helper across the job tests

Three test namespaces had their own copy of the same insert, differing only
in the kind of the job and in whether the row holds the resource it
consumes. Move it to the test helpers as create-job*, with the kind of the
job as its first argument: it is what resolves the job-def of the row when
its result is read or checked, so every test states it.

The bang of the name goes away with it: the design rules keep it for the
mutating API of app.db, and this is a fixture of the tests. The three
namespaces lose the requires that only their removed copy used.

Relates to #12339

AI-assisted-by: deepseek-v4.1-flash
Signed-off-by: Andrey Antukh <niwi@niwi.nz>

---------

Signed-off-by: Andrey Antukh <niwi@niwi.nz>
This commit is contained in:
Andrey Antukh 2026-10-15 19:18:04 +02:00 committed by GitHub
parent 7383e08683
commit 57fa280209
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 698 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.serena/memories/backend/core.md
View File

@ -55,7 +55,7 @@ For deeper details on transaction semantics, advisory locks, Transit vs JSON hel
## Background tasks (unified jobs)
Every background job is a job-def: a plain `(defn execute-X [cfg params] ...)` in its namespace + a malli params schema + an `ig/init-key` that returns the job-def map `{::jobs/name, ::jobs/schema, ::jobs/handler, ::jobs/decoder, ::jobs/validator}` (decoder/validator precompiled at init). The registry is the `::jobs/defs` wiring in `app.main`, which also populates a module-level registry used by `jobs/submit` as fallback — job-def components cannot ig/ref `::jobs/defs` (wiring cycle). Submit from RPC code passes its RPC cfg (it carries `::jobs/defs` via ig/ref).
Every background job is a job-def: a plain `(defn execute-X [cfg params] ...)` in its namespace + a malli params schema + an `ig/init-key` that returns the job-def map `{::jobs/name, ::jobs/schema, ::jobs/handler, ::jobs/decoder, ::jobs/validator}` (decoder/validator precompiled at init), optionally with `::jobs/result-schema` when the result carries types plain JSON does not keep (a uuid, whether a value or a map key): the substrate compiles the decoder from that schema on registry init and checks a result against the schema before storing it. The registry is the `::jobs/defs` wiring in `app.main`, which also populates a module-level registry used by `jobs/submit` as fallback — job-def components cannot ig/ref `::jobs/defs` (wiring cycle). Submit from RPC code passes its RPC cfg (it carries `::jobs/defs` via ig/ref).
- `::jobs/handler` is `(fn [context params] ...)`. It is NOT given the runner cfg: a job-def closes over its own dependencies, and handing it the runner cfg would let a handler reach the pool or any other component. A wrapper passes the context to its implementation only when the implementation takes it; most do not, and their implementation stays `[cfg params]`.
- The context is built by `jobs/make-context` and has exactly five keys: `id`, `name`, `label`, `resource-id`, `profile-id`, behind a closed schema. `profile-id` is the owner of a user-facing job and nil for an internal one, and it is there so a handler can revalidate permissions and check who owns its resource without a second query. `queue` is absent on purpose (a routing detail of the dispatcher, not a property of the job) and so are `retry-num`/`max-retries`/`attempt` (retry policy belongs to the runner, and an attempt number cannot be durable because the `noop` strategy re-runs without incrementing the counter).
@ -64,14 +64,14 @@ Every background job is a job-def: a plain `(defn execute-X [cfg params] ...)` i
- A job-def may carry optional metadata keys read from the registry by its consumers, never by the substrate: `::jobs/family` (e.g. `:export`) groups the jobs of a family for quotes, `::jobs/resource-role` (`:input` or `:output`) says whether the job's `resource_id` is what it consumes or what it produces, and `::jobs/queue-name` names the queue the creation command routes the job to (`submit-job` falls back to `:binfile` when the def does not set it).
- Job resources (`app.jobs.storage`) are the objects a job owns through `job.resource_id`: they live in the `job-resource` bucket, are written with the profile of the job as owner (only that profile can read them back from `/assets/by-id`) and are never deduplicated. They carry no expiry of their own, because `expired-at` is stored as `deleted_at` and an object with it is unreadable from birth: the retention of the artifact is the retention of its job row. The writer marks the object as touched right away, so an artifact that no row references (a crash between the upload and the completion) is reclaimed by `storage-gc-touched`, and a row that does reference it freezes it until the jobs GC deletes the row and touches the object.
- The binfile jobs (`:export-binfile` and `:import-binfile`) run the core of `bf.v1`/`bf.v3` through `app.binfile.jobs`, which turns the `:progress` taps of the core into `progress` events of the job with `jobs/heartbeat` (the `modified_at` touch at most once per second, progress events at up to ~200ms). Each event is a self-contained milestone: a `:stage` keyword of one vocabulary shared by the two formats (the same work has a different name in v1 and v3) and an optional `:counters` map, with the counter of the activity named after its own stage and the outer scopes in course as context, which the tap carries itself (the reader puts `::outer-counters` in every tap inside them — today only `:files`, extendable to `:teams`/`:projects` for multi-scope exports — so the adapter remembers nothing between taps and a tap that happens outside any scope has no context); the ids and names the taps carry never reach the event. Every tap is reported, so a tap with no units to count still keeps the job alive, and a milestone the throttle drops loses nothing because the next one carries the same state. The adapter checks that the job is still active before the run, and the beat raises an `:interrupt` right there on the run thread when the job is no longer active, so the work stops at its next event point inside the transaction of its handler, which rolls back. A run whose last tap predates the interruption is discarded by the check once it returns. The adapter passes the job id from the handler context. Their job-def components carry pool, storage, metrics and msgbus, because a job with a `profile_id` cannot store an event without a msgbus on its cfg. Both readers answer with the same shape (`:file-ids` and, in version 3, the resolution of the libraries), because the version 1 reader returns a bare set of file ids.
- Both binfile handlers revalidate their permission at run time, not only when the job is created, because it can be revoked while the job waits: the export checks the read permission of every file before reading anything and stores the package in `job-resource` as the resource of the job (answering with the completion envelope), and the import checks the edition permission of the destination and runs the core inside a transaction, which is what makes a cancelled import roll back instead of leaving half of a file. The import releases its package (marks it as deleted and drops the reference from the job row in the same transaction) and deletes its temporary copy once the job is terminal, whatever the outcome. Both handlers check their own result against a schema before returning it, because the job-def has no result schema of its own.
- Both binfile handlers revalidate their permission at run time, not only when the job is created, because it can be revoked while the job waits: the export checks the read permission of every file before reading anything and stores the package in `job-resource` as the resource of the job (answering with the completion envelope), and the import checks the edition permission of the destination and runs the core inside a transaction, which is what makes a cancelled import roll back instead of leaving half of a file. The import releases its package (marks it as deleted and drops the reference from the job row in the same transaction) and deletes its temporary copy once the job is terminal, whatever the outcome. Both handlers check their own result against a schema before returning it (the import job-def declares that same schema as its result schema, which is what types the resolution when the row is read).
- The domain cfg of a binfile request lives in `app.binfile.common` (`export-cfg`, `import-cfg`), so the legacy RPC and the jobs freeze the same ids, resolve the same team features and apply the same anti-zip-bomb limits. `resolve-export-type` is the translation the legacy RPC needs for its deprecated boolean flags; the creation command asks the caller for the type instead of guessing a default.
- Two retention knobs, easy to confuse: `:jobs-retention` sweeps terminal internal rows (the ones without a profile), while `:jobs-user-ttl` is the default expiry a user-facing job gets when it is created; the jobs GC deletes the row once `expires_at` passes and touches its resource. The heavy work of a user job runs on its own `:binfile` queue (`:worker-binfile-parallelism`, default 1) so it does not block the default runner.
- The quote of a family of jobs (`::export-jobs-per-profile`, `::import-jobs-per-profile`) counts the jobs of the profile that are still on their way (`new`/`scheduled`/`running`/`retry`) whose name the registry gives for that family, so a new job of the family is covered without touching the quote.
- A user-facing job is created through `app.rpc.commands.jobs`. One command per job type: `create-export-binfile-job` and `create-import-binfile-job` name the job they create and their body carries only `:params`, whose schema the command imports from the job namespace (`::sm/params`), so the RPC validates the business params up front, before the command runs; the job-def is still the place that owns them and validates them again at run time. The commands check the permission and the quote of the family before the expensive part, submit with `max-retries 0`, an expiry from `:jobs-user-ttl` and the queue the job-def routes to through `::jobs/queue-name` (`:binfile` when the def says nothing; read by the command, never by the substrate), and answer with a summary of the row. The import command has its own creation schema (`import-binfile/schema:create-params`, version optional) because the request is not the stored params: the command fills the version (from the caller or the package header) and the manifest metadata kept for audit. It assembles the upload, reads the manifest of a version 3 package for the audit (never its content) and stores the package in `job-resource` as the resource of the job, releasing it when the job never comes to own it.
- `get-job` is the read surface a profile has over its own jobs: it answers with the same summary the creation commands return plus the `result` of a finished job or its public error (through `decode-job-error`, so the keywords come back), and a job of another profile is not distinguishable from one that does not exist: both answer `not-found`. `decode-row` decodes `params`, `result` and `error`, and the result is the plain JSON the row holds, so a uuid comes back as the text it is on disk (a result has no schema of its own like the params do). The event log still has no read surface: a client that follows a job takes its progress from the websocket.
- `get-job` is the read surface a profile has over its own jobs: it answers with the same summary the creation commands return plus the `result` of a finished job or its public error (through `decode-job-error`, so the keywords come back), and a job of another profile is not distinguishable from one that does not exist: both answer `not-found`. `decode-row` decodes `params`, `result` and `error`, and the result is then decoded with the decoder the substrate compiled from the result schema the job-def the row names declares (`decode-result`; a def without one, or a row whose job-def is gone, keeps the plain JSON the row holds: a uuid as the text it is on disk, a uuid used as a map key as a keyword). The event log still has no read surface: a client that follows a job takes its progress from the websocket.
- `heartbeat` uses named options `[cfg & {:keys [job-id progress] :as options}]` and returns the number of durable writes, so 0 means there was nothing to write: no job context, or the throttle did not allow it. A write that finds the job no longer active (terminal or gone) raises an `:interrupt` (`code :job-interrupted`, carrying the job id and the status it found) through the shared `jobs/check-active` instead of counting zero, so a task in course stops at its next beat; the interrupt is internal and never reaches HTTP (the management progress report answers `:skip` for it). The job id falls back to `::jobs/job-id` on the cfg and then to the runner-bound `*job-id*`. The `modified_at` touch is throttled to ~1s and the progress event to ~200ms, so the check fires at most once per second with no extra query (the read of the row happens only when a write already found nothing). A progress report is a milestone (`schema:progress`): a required `:stage` keyword plus an optional `:counters` map of `scope -> {:current, :total?}`, and no other key. Counters are non-negative, a `total` is positive and never lower than its `current`, and stage and counter keys are names of at most 64 characters. A milestone without counters is valid (a point of the run with no units to count), the stage vocabulary belongs to the worker and not to the substrate, and a client must degrade for a stage it does not know. Read a stored payload back with `jobs/decode-progress`: the database holds the keywords as strings. The 1s beat (down from 60s, so a cancelled heavy job is seen as gone at its next beat) costs one `modified_at` touch per beating job per second: cheap while the only heavy beater runs alone on its queue, revisit if concurrent beating jobs ever grow.
- `job.error` conforms to `app.jobs/schema:job-error` (`type` and `code` keywords, `hint` text, extra details allowed). Read it back with `jobs/decode-job-error`; a plain `db/decode-json-pgobject` leaves the keywords as strings. `complete` takes named options and an optional `resource-id`, which is only ever set, never replaced: it reads the row locked inside its own transaction and applies the rules in a fixed order, skipping a job that is no longer completable before refusing one for its resource, so the answer is a function of the row and not of the order two callers arrive in.
- `job.error` conforms to `app.jobs/schema:job-error` (`type` and `code` keywords, `hint` text, extra details allowed). Read it back with `jobs/decode-job-error`; a plain `db/decode-json-pgobject` leaves the keywords as strings. `complete` takes named options and an optional `resource-id`, which is only ever set, never replaced: it reads the row locked inside its own transaction and applies the rules in a fixed order, skipping a job that is no longer completable before refusing one for its resource, so the answer is a function of the row and not of the order two callers arrive in. It checks the result against the schema the job-def of the row declares before it writes (a violation raises `:invalid-result` and nothing is stored), and only for a transition that is going to write: a terminal job answers `:skip` without validating. `complete` is the single door of a result, so the check covers the runner and the external worker that completes a job through the management RPC (`::complete-job`), which stores a result the substrate would otherwise take blind.
For worker dispatch, cron, retry semantics (`ex/raise :type ::wrk/retry` with `:delay`/`:strategy`), deduplication, job events, and queue internals: `mem:backend/subtleties`.

2
.serena/memories/backend/subtleties.md
View File

@ -29,7 +29,7 @@
- Whether a call is nested is decided by the CONNECTION, not by an ambient flag: `db/transact!` binds `*after-commit-conn*` around the body, and a transactable that is not identical to that connection is an independent unit of work that owns its own after-commit context. A `tx-run!` that passes a pool while a transaction is open therefore commits on its own and drains its own callbacks, instead of leaving them in a context that may roll back.
- `after-commit` callbacks run on commit only. A transaction marked `::db/rollback` (used by `srepl` and `app.tasks/offload-file-data`) commits nothing, so it runs none of them.
- `db/run!` opens/reuses one connection but does not create a transaction.
- `db/tjson` is Transit JSON for jsonb storage; `db/json` is plain JSON. Job params (`job.params`) are plain JSON, decoded with the job-def decoder (`app.jobs/decode-params`); only the legacy `task` props were Transit (`decode-transit-pgobject`).
- `db/tjson` is Transit JSON for jsonb storage; `db/json` is plain JSON. Job params (`job.params`) are plain JSON, decoded with the job-def decoder (`app.jobs/decode-params`); only the legacy `task` props were Transit (`decode-transit-pgobject`). Job results (`job.result`) are plain JSON too: they are read with the result decoder of the job-def (`app.jobs/decode-result`), and a result map keyed by uuid must declare its schema (`[:map-of ::sm/uuid ...]`, the import resolution does) or every key comes back as a keyword and every uuid as the text it is on disk.
- Advisory transaction locks accept UUIDs or ints. UUID locks are hashed using a zero-UUID seeded siphash.
## Workers and cron

2
.serena/memories/frontend/dashboard-viewer-subtleties.md
View File

@ -5,7 +5,7 @@
- Dashboard initialization fetches projects and fonts for the team, then listens to websocket messages only for global topic `uuid/zero` or the current profile id.
- Project fetch replaces each project map completely instead of merging, so fields such as `deleted-at` can disappear cleanly.
- Dashboard file/project mutations are often optimistic local updates with fire-and-forget RPC watchers. Bulk permanent delete/restore paths use SSE progress and progress notifications.
- The import wizard runs the package as a job of the backend: it reads the manifest of the file in the main thread (`app.main.data.imports`, no web worker) and follows the job, so its rows paint the milestone of the job (the stage and its counter, with the file in course) instead of the messages the old SSE stream sent. Before the job exists, the rows paint a client-side `:upload` milestone built from the chunked upload (`uploads/upload-blob-chunked` `:on-progress`, chunks done/total), which never comes from the backend; the chunks go one at a time and the row writes them as a percentage of the file, not as a counter. Once created and still waiting for a worker, each entry reports `:queued` ("Queued"); the `:start`/`:retry` events move it to bare progress, and only `:progress` carries a milestone. The export modal follows the same one-job-per-file model, but every file is queued up front (even the ones whose job is created only after the previous one is over) and its queue ends only with a milestone: `:start`/`:retry` do not end it there. Neither modal paints a generic "…ing file…" line any more: each row carries its own status. The `:error` of an import entry is user-facing text (the backend hint, or a translated label), never an i18n key: the wizard renders it as-is, it does not translate it again. The legacy SSE import stays in the backend for other clients.
- The import wizard runs the package as a job of the backend: it reads the manifest of the file in the main thread (`app.main.data.imports`, no web worker) and follows the job, so its rows paint the milestone of the job (the stage and its counter, with the file in course) instead of the messages the old SSE stream sent. Before the job exists, the rows paint a client-side `:upload` milestone built from the chunked upload (`uploads/upload-blob-chunked` `:on-progress`, chunks done/total), which never comes from the backend; the chunks go one at a time and the row writes them as a percentage of the file, not as a counter. Once created and still waiting for a worker, each entry reports `:queued` ("Queued"); the `:start`/`:retry` events move it to bare progress, and only `:progress` carries a milestone. The export modal follows the same one-job-per-file model, but every file is queued up front (even the ones whose job is created only after the previous one is over) and its queue ends only with a milestone: `:start`/`:retry` do not end it there. Neither modal paints a generic "…ing file…" line any more: each row carries its own status. The `:error` of an import entry is user-facing text (the backend hint, or a translated label), never an i18n key: the wizard renders it as-is, it does not translate it again. The link-later step takes the file id from the `:id` of its resolution entry, never from the key of the resolution map: the map comes from the job result, so what a key holds depends on the result schema of the job-def that wrote it, while the wizard indexes its selection by the id of the entry. The legacy SSE import stays in the backend for other clients.
- File creation/duplication strips file `:data` before putting file summaries into dashboard state.
## Viewer

107
backend/src/app/jobs.clj
View File

@ -9,11 +9,15 @@
Every job is defined as a job-def map returned by the `ig/init-key` of its
module: {::name, ::schema, ::handler, ::decoder, ::validator} — decoder and
validator are precompiled at init time. A job-def may also carry the
optional metadata {::family, ::resource-role}, which the consumers of the
job (quotes, storage helpers) read from the registry: the substrate never
reads them. The registry of job-defs is plain integrant wiring (::jobs/defs);
submit, dispatch and management consume it by reference.
validator are precompiled at init time. A def whose result needs its types
back declares the optional ::result-schema: the substrate compiles the
decoder from that schema and checks a result against it before storing it,
while a def without one answers with the JSON the row holds. A job-def may
also carry the optional metadata {::family, ::resource-role}, which the
consumers of the job (quotes, storage helpers) read from the registry: the
substrate never reads them. The registry of job-defs is plain integrant
wiring (::jobs/defs); submit, dispatch and management consume it by
reference.
Two execution modes are provided:
- `submit` (durable): validates + JSON-encodes params and inserts a row
@ -24,6 +28,11 @@
Params payloads are stored as plain JSON (not transit) in the `params` jsonb
column and decoded back to typed Clojure values using the job-def decoder.
Results are plain JSON in the `result` jsonb column: they are decoded with the
decoder the substrate compiled from the schema the def declares, and a result
that does not conform to that schema is refused before it is stored. Without a
schema a uuid comes back as the text it is on disk and a uuid used as a map
key as a keyword, so a def that keys a result map by uuid must declare it.
Any job that can run longer than `:jobs-lease` must call `heartbeat`
on its loop, otherwise the dispatcher marks it `aborted` (a system-side
@ -151,7 +160,14 @@
;; implementation stays [cfg params].
[::handler ::sm/fn]
[::decoder ::sm/fn]
[::validator ::sm/fn]])
[::validator ::sm/fn]
;; A result crosses a plain JSON column on its way out of the job, so a
;; uuid comes back as the text it is on disk and a uuid used as a map
;; key comes back as a keyword. A def that declares the schema of its
;; result gets its types back when the row is read (the substrate
;; compiles the decoder from the schema), and its result is checked
;; against that schema before it is stored.
[::result-schema {:optional true} any?]])
(def ^:private schema:job-defs
[:map-of :keyword schema:job-def])
@ -189,11 +205,24 @@
:job (d/name name)
:expected (d/name (::name job-def))))))
(defn- compile-result-schema
"Compile the decoder of a job-def from the result schema it declares, so the
pair cannot drift from the schema and a def without one keeps working
untouched. The write path checks the result against the schema itself."
[job-def]
(if-let [schema (::result-schema job-def)]
(assoc job-def ::result-decoder (sm/decoder schema sm/json-transformer))
job-def))
(defmethod ig/init-key ::defs
[_ defs]
(reset! defs-registry defs)
(l/inf :hint "job definitions initialized" :jobs (count defs))
defs)
(let [defs (reduce-kv (fn [acc name job-def]
(assoc acc name (compile-result-schema job-def)))
{}
defs)]
(reset! defs-registry defs)
(l/inf :hint "job definitions initialized" :jobs (count defs))
defs))
(defmethod ig/halt-key! ::defs
[_ defs]
@ -222,6 +251,32 @@
db/decode-json-pgobject)
((::decoder job-def))))
(defn decode-result
"Decode the raw JSON result (pgobject or decoded map) with the decoder the
substrate compiled from the result schema of the job-def. A def without one
answers with the JSON the row holds: every key keywordized and every uuid as
the text it is on disk."
[job-def result]
(when (some? result)
(let [value (cond-> result
(db/pgobject? result)
db/decode-json-pgobject)]
(if-let [decoder (::result-decoder job-def)]
(decoder value)
value))))
(defn- check-result
"Validate the result of a job against the schema its job-def declares, with
the malli explanation on failure. A job-def without a result schema has
nothing to check, and so has a job whose definition is gone."
[job-def result]
(when-let [schema (::result-schema job-def)]
(sm/check schema result
:hint "invalid result of a job"
:type :validation
:code :invalid-result))
result)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; SUBMIT API
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@ -432,11 +487,21 @@
(decode-json-col :result)
(decode-json-col :error)))
(defn- decode-result-of
"Decode the result of a row with the decoder of the job-def that owns it.
A kind the registry no longer knows (a retired job-def) answers with the
row untouched instead of raising."
[cfg row]
(if-let [job-def (get (get-defs cfg) (keyword (:name row)))]
(update row :result #(decode-result job-def %))
row))
(defn get-job
"Retrieve the job row (with the raw JSON params decoded to a plain map)."
"Retrieve the job row, with params and result decoded to plain maps (the
result only when its job-def declares a result decoder)."
[cfg job-id]
(some-> (db/get* cfg :job {:id job-id})
(decode-row)))
(when-let [row (db/get* cfg :job {:id job-id})]
(decode-result-of cfg (decode-row row))))
(defn get-user-status
"Map the internal job status to the user-facing status."
@ -1058,7 +1123,8 @@
- `:job-id` the job to complete.
- `:result` the JSON-encodable result, nil when the job returns
nothing.
nothing; a result that does not conform to the schema
its job-def declares is refused.
- `:resource-id` optional storage object to associate with the job. A
resource is only ever set, never replaced: passing one
for a job that already has a resource is a validation
@ -1073,11 +1139,13 @@
aborted by the dispatcher — is never overwritten) and writes no event. Returns
the number of affected rows.
The row is read locked inside that same transaction, so the two rules
are applied in a fixed order against the row and not against the
clock: a job that is no longer completable is skipped, and only a job
that is still completable can be refused for its resource. A
concurrent completion waits for the lock and then sees the outcome."
The row is read locked inside that same transaction, so the rules are
applied in a fixed order against the row and not against the clock: a
job that is no longer completable is skipped, only a job that is still
completable can be refused for its resource, and the result is checked
against the schema its job-def declares before it is stored (a def
without one has nothing to check). A concurrent completion waits for
the lock and then sees the outcome."
[cfg & {:keys [job-id result resource-id]}]
(assert (mtx/instance cfg) "missing ::mtx/metrics on the jobs cfg")
(check-resource-id resource-id)
@ -1097,6 +1165,9 @@
:hint "the job already has a resource, it cannot be replaced"
:job-id job-id
:resource-id (:resource-id job)))
(check-result (get (get-defs tx-cfg)
(keyword (:name job)))
result)
(if (db/exec-one! conn [sql:complete-job (ct/now) (ct/now)
encoded resource-id job-id])
(do

14
backend/src/app/rpc/commands/jobs.clj
View File

@ -330,11 +330,15 @@
"Read a job the caller created: its state and, once it is over, what it
produced or why it failed.
The result of a finished job is the plain JSON the row holds, because a
result has no schema of its own like the params do: a uuid comes back as
the text it is on disk. The progress and the history of the job are not
part of the answer: a client that follows a job takes its progress from
the websocket, and the event log still has no read surface."
The result of a finished job is the JSON the row holds, decoded with the
result schema its job-def declares: a job that keys its result by uuid
(the import resolution) answers with the types it stored, and a job whose
job-def declares no schema answers with the JSON as it is on disk, where a
uuid is text and a uuid used as a map key is a keyword. A result that does
not conform to the declared schema never reaches the column: the substrate
validates it when the job is completed. The progress and the history of
the job are not part of the answer: a client that follows a job takes its
progress from the websocket, and the event log still has no read surface."
{::doc/added "2.20"
::sm/params schema:get-job
::sm/result schema:job-detail}

12
backend/src/app/rpc/management/jobs.clj
View File

@ -106,7 +106,10 @@
[:job-id ::sm/uuid]
;; Per-job result: any JSON-encodable value, nil when the job returns
;; nothing. Nested maps and vectors are allowed; unserializable
;; values are dropped to nil by jobs/encode-result with a warning.
;; values are dropped to nil by jobs/encode-result with a warning. A
;; job whose job-def declares a result schema has its result validated
;; against it: one that does not conform is rejected and the job is
;; not completed.
[:result [:maybe :any]]
;; Storage object the worker produced for this job. A resource is only
;; set, never replaced: a job that already has one is rejected.
@ -117,6 +120,13 @@
[:action [:enum :run :skip]]])
(sv/defmethod ::complete-job
"Complete a job on behalf of the worker that ran it, with the result it
produced and, when it wrote one, the resource it stores.
The result of a job whose job-def declares a result schema is validated
against it before it is stored: one that does not conform is rejected and
the job is not completed. A job that is no longer completable is skipped,
and nothing is validated for it."
{::doc/added "2.20"
::sm/params schema:complete-job-params
::sm/result schema:complete-job-result

77
backend/src/app/tasks/import_binfile.clj
View File

@ -44,12 +44,53 @@
[:map-of ::sm/uuid [:string {:max 250}]]]]
[:version {:optional true} [:enum 1 3]]])
(def schema:candidate
"A library of the instance whose slugified name matches the one the package
used and is therefore offered to the user as a replacement."
[:map
[:id ::sm/uuid]
[:name :string]
[:project-id ::sm/uuid]
[:project-name :string]])
(def schema:done-entry
"A library of the package that was linked while it was read."
[:map
[:id ::sm/uuid]
[:name :string]
[:linked-to ::sm/uuid]])
(def schema:pending-entry
"A library of the package that needs the user to choose a replacement:
candidates are the libraries of the instance with the same slugified name
(`bfc/find-shared-files-by-slug`)."
[:map
[:id ::sm/uuid]
[:name :string]
[:candidates [:vector schema:candidate]]])
(def schema:resolution-entry
"The libraries one imported file uses, split in the ones already linked
and the ones pending resolution."
[:map
[:id ::sm/uuid]
[:name :string]
[:done [:vector schema:done-entry]]
[:pending [:vector schema:pending-entry]]])
(def schema:resolution
"The resolution of every imported file, keyed by its new file id. The key
is a uuid, so the map needs this schema to come back typed: the result
column is plain JSON, and without a decoder every key of the map would be
read back as a keyword."
[:map-of ::sm/uuid schema:resolution-entry])
(def schema:result
"What the job stores when the import succeeds: the files it created and
how the libraries of the package were resolved."
[:map {:title "import-binfile-result" :closed true}
[:file-ids [:vector ::sm/uuid]]
[:resolution {:optional true} :map]
[:resolution {:optional true} schema:resolution]
[:name [:or [:string {:max 250}]
[:map-of ::sm/uuid [:string {:max 250}]]]]
[:version [:enum 1 3]]])
@ -66,7 +107,11 @@
The edition permission is checked again here, and not only when the job
was created: it can be revoked while the job waits in its queue. The
core runs inside a transaction, so a cancellation detected between units
rolls the whole import back.
rolls the whole import back, and the result is checked against its schema
inside that same transaction: a result the contract does not accept rolls
the import back instead of leaving half of a file imported. The substrate
checks the result again when the job is completed, which is the only net
for a job that does not validate its own result.
The package is the input of the job: it is released and its temporary
copy deleted once the job is terminal, whatever the outcome. A crash is
@ -92,17 +137,18 @@
;; the core joins this transaction: it
;; is the one that decides whether the
;; whole import is kept or rolled back
(bfj/import-files tx-cfg context
{:profile-id profile-id
:project-id project-id
:team team
:name (:name params)
:input input
:version (:version params)})))]
(check-result
(assoc (bfj/import-files tx-cfg context
{:profile-id profile-id
:project-id project-id
:team team
:name (:name params)
:input input
:version (:version params)})
:name (:name params)
:version (:version params)))))]
(check-result (assoc result
:name (:name params)
:version (:version params))))
result)
(finally
(fs/delete input)
@ -123,4 +169,9 @@
::jobs/handler (fn [context params]
(execute-import cfg context params))
::jobs/decoder (sm/decoder schema:params sm/json-transformer)
::jobs/validator (sm/validator schema:params)})
::jobs/validator (sm/validator schema:params)
;; The resolution is keyed by file id: without the result schema the
;; keys come back as keywords and the dashboard wizard can no longer
;; match them with the file ids it holds. The substrate compiles the
;; decoder from it and checks the result against it before storing it.
::jobs/result-schema schema:result})

200
backend/test/backend_tests/binfile_jobs_test.clj
View File

@ -13,7 +13,6 @@
[app.binfile.v3 :as v3]
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.jobs :as jobs]
[app.jobs.storage :as js]
@ -49,27 +48,6 @@
(assoc ::bfc/input (fixture-path))
(v3/import-files!)))
(defn- make-job!
"A job row as the substrate writes it: running, owned by a profile and
holding a resource when the test needs one."
[profile-id resource-id]
(let [id (uuid/next)]
(th/db-insert! :job {:id id
:name "export-binfile"
:tenant (cf/get :tenant)
:queue "binfile"
:params (db/json {})
:priority 100
:max-retries 0
:retry-num 0
:status "running"
:profile-id profile-id
:resource-id resource-id
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
id))
(defn- job-cfg
"The system cfg as the runner hands it to a handler. The job id is NOT
bound here: the adapter takes it from the context, because the progress
@ -90,6 +68,46 @@
job-id])
(mapv #(jobs/decode-progress (:payload %)))))
(defn- export-link-later-package
"A package of one file that uses a shared library, exported so the
library is resolved while importing. The library of the package is gone
afterwards and two replacements share its name, so the import cannot
link it by itself and leaves the file pending resolution."
[profile]
(let [library (th/create-file* 1 {:profile-id (:id profile)
:project-id (:default-project-id profile)
:is-shared true
:name "Icons Library"})
file (th/create-file* 2 {:profile-id (:id profile)
:project-id (:default-project-id profile)
:is-shared false})
output (tmp/tempfile :suffix ".zip")]
(db/insert! th/*system* :file-library-rel
{:file-id (:id file)
:library-file-id (:id library)})
(v3/export-files! (-> th/*system*
(assoc ::bfc/ids #{(:id file)})
(assoc ::bfc/export-type :link-later))
(io/output-stream output))
(db/update! th/*system* :file
{:deleted-at (ct/now)}
{:id (:id library)})
(th/create-file* 3 {:profile-id (:id profile)
:project-id (:default-project-id profile)
:is-shared true
:name "Icons Library"})
(th/create-file* 4 {:profile-id (:id profile)
:project-id (:default-project-id profile)
:is-shared true
:name "Icons Library"})
{:library-id (:id library)
:output output}))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; ADAPTER
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@ -97,7 +115,7 @@
(t/deftest export-through-the-adapter-produces-the-artifact
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
job-id (make-job! (:id profile) nil)
job-id (th/create-job* :export-binfile {:profile-id (:id profile)})
output (tmp/tempfile* :suffix ".penpot")]
(bfj/export-files (job-cfg) (job-context job-id)
@ -129,7 +147,8 @@
{:content (sto/content (fixture-path))
:filename "package.penpot"
:mtype "application/zip"})
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
context (job-context job-id)
result (bfj/import-files (job-cfg) context
{:profile-id (:id profile)
@ -148,7 +167,7 @@
(t/deftest the-adapter-composes-a-milestone-per-tap
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
job-id (make-job! (:id profile) nil)
job-id (th/create-job* :export-binfile {:profile-id (:id profile)})
reports (atom [])]
;; the substrate throttles the durable events, so the whole stream of
@ -183,7 +202,7 @@
(t/deftest a-tap-carries-the-outer-counters-in-course
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
job-id (make-job! (:id profile) nil)
job-id (th/create-job* :export-binfile {:profile-id (:id profile)})
taps (atom [])]
;; the taps themselves, before the adapter turns them into events
@ -242,7 +261,8 @@
{:content (sto/content (fixture-path))
:filename "package.penpot"
:mtype "application/zip"})
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
context (job-context job-id)
reports (atom [])]
@ -278,7 +298,7 @@
(t/deftest a-job-that-is-no-longer-active-stops-the-work
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
job-id (make-job! (:id profile) nil)
job-id (th/create-job* :export-binfile {:profile-id (:id profile)})
context (job-context job-id)]
(t/testing "a running job is active"
@ -369,7 +389,7 @@
(t/deftest the-export-handler-runs-through-the-registry
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
job-id (make-job! (:id profile) nil)
job-id (th/create-job* :export-binfile {:profile-id (:id profile)})
result (jobs/invoke (-> th/*system*
(assoc ::jobs/name :export-binfile)
(assoc ::jobs/params {:file-ids #{file-id}
@ -389,7 +409,8 @@
{:content (sto/content (fixture-path))
:filename "package.penpot"
:mtype "application/zip"})
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
result (jobs/invoke (-> th/*system*
(assoc ::jobs/name :import-binfile)
(assoc ::jobs/params {:project-id (:default-project-id profile)
@ -401,3 +422,122 @@
(t/is (pos? (count (:file-ids result)))))
(t/testing "the handler reported progress without knowing the job id"
(t/is (pos? (count (progress-events job-id)))))))
(t/deftest the-import-result-keeps-the-type-of-its-ids
"The resolution of an import is a map keyed by the new file id, and it
crosses a plain JSON column: the result schema the job-def declares is
what keeps the keys and the values as uuids, which is what the dashboard
wizard matches against the ids it holds. Without the decoder the substrate
compiles from that schema, the key comes back as a keyword and the entry as
text, and this test fails."
(let [profile (th/create-profile* 1)
{:keys [library-id output]} (export-link-later-package profile)
staged (js/put-resource th/*system* (:id profile)
{:content (sto/content output)
:filename "package.penpot"
:mtype "application/zip"})
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
result (jobs/invoke (-> th/*system*
(assoc ::jobs/name :import-binfile)
(assoc ::jobs/params {:project-id (:default-project-id profile)
:name "imported"
:version 3})
(assoc ::jobs/context (job-context job-id))))]
(t/testing "the result of the job is stored as the substrate does"
(t/is (= 1 (jobs/complete th/*system* :job-id job-id :result result))))
(let [stored (:result (jobs/get-job th/*system* job-id))
[file-id file-res] (first (:resolution stored))]
(t/testing "the resolution comes back keyed by the new file id"
(t/is (= 1 (count (:resolution stored))))
(t/is (uuid? file-id))
(t/is (= file-id (:id file-res))))
(t/testing "the ids of the entries keep their type"
(t/is (= [file-id] (:file-ids stored)))
(t/is (= [] (:done file-res)))
(t/is (= 1 (count (:pending file-res))))
(let [pending (first (:pending file-res))]
(t/is (= library-id (:id pending)))
(t/is (= 2 (count (:candidates pending))))
(t/is (every? (comp uuid? :id) (:candidates pending)))
(t/is (every? (comp uuid? :project-id) (:candidates pending))))))))
(t/deftest a-package-without-external-libraries-reads-with-an-empty-resolution
(let [profile (th/create-profile* 1)
staged (js/put-resource th/*system* (:id profile)
{:content (sto/content (fixture-path))
:filename "package.penpot"
:mtype "application/zip"})
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
result (jobs/invoke (-> th/*system*
(assoc ::jobs/name :import-binfile)
(assoc ::jobs/params {:project-id (:default-project-id profile)
:name "imported"
:version 3})
(assoc ::jobs/context (job-context job-id))))]
(t/is (= 1 (jobs/complete th/*system* :job-id job-id :result result)))
(let [stored (:result (jobs/get-job th/*system* job-id))]
(t/testing "a package without external libraries resolves nothing"
(t/is (= {} (:resolution stored))))
(t/testing "and the file ids come back typed"
(t/is (= 1 (count (:file-ids stored))))
(t/is (every? uuid? (:file-ids stored)))))))
(t/deftest a-version-1-result-reads-without-a-resolution
"A version 1 import answers with the file ids and no resolution: the read
path decodes it with the schema of the job-def, which declares the
resolution as optional. Without that schema the file ids come back as text."
(let [profile (th/create-profile* 1)
file-id (uuid/next)
job-id (th/create-job* :import-binfile {:profile-id (:id profile)})]
(t/is (= 1 (jobs/complete th/*system* :job-id job-id
:result {:file-ids [file-id]
:name "imported-v1"
:version 1})))
(let [stored (:result (jobs/get-job th/*system* job-id))]
(t/testing "there is no resolution to decode"
(t/is (nil? (:resolution stored))))
(t/testing "and the file ids come back typed"
(t/is (= [file-id] (:file-ids stored)))
(t/is (= 1 (:version stored)))))))
(t/deftest a-result-of-a-retired-job-def-reads-as-plain-json
"A registry that no longer knows the kind of the row answers with the JSON
the column holds, keywordized keys and uuid as text, instead of raising."
(let [profile (th/create-profile* 1)
file-id (uuid/next)
job-id (th/create-job* :import-binfile {:profile-id (:id profile)})
key (keyword (str file-id))]
(t/is (= 1 (jobs/complete th/*system* :job-id job-id
:result {:file-ids [file-id]
:resolution {file-id {:id file-id
:name "card.penpot"
:done []
:pending []}}
:name "imported"
:version 3})))
(let [stored (:result (jobs/get-job (assoc th/*system* ::jobs/defs {}) job-id))]
(t/testing "the resolution comes back as it is on disk"
(t/is (= [key] (keys (:resolution stored))))
(t/is (= (str file-id) (get-in stored [:resolution key :id]))))
(t/testing "and so do the file ids"
(t/is (= [(str file-id)] (:file-ids stored)))))))

24
backend/test/backend_tests/helpers.clj
View File

@ -252,6 +252,30 @@
:features features}
params)))))))
(defn create-job*
"A job row in the state a worker finds it: running, in the `binfile` queue,
owned by `profile-id` and holding `resource-id` when it consumes one.
`job-name` is the kind of the job (a keyword like the `::jobs/name` a caller
would pass, or the text the row holds) and it is what resolves the job-def
of the row when its result is read or checked."
[job-name {:keys [profile-id resource-id]}]
(let [id (uuid/next)]
(db/insert! *pool* :job {:id id
:name (name job-name)
:tenant (cf/get :tenant)
:queue "binfile"
:params (db/json {})
:priority 100
:max-retries 0
:retry-num 0
:status "running"
:profile-id profile-id
:resource-id resource-id
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
id))
(defn mark-file-deleted*
([params]
(mark-file-deleted* *system* params))

129
backend/test/backend_tests/jobs_test.clj
View File

@ -65,10 +65,31 @@
::jobs/decoder (sm/decoder schema:echo-params sm/json-transformer)
::jobs/validator (sm/validator schema:echo-params)})
(def schema:typed-result
"A result with a uuid keyed map, the shape that needs a declared schema
to survive the plain JSON round trip of the `job.result` column."
[:map
[:file-ids [:vector ::sm/uuid]]
[:resolution [:map-of ::sm/uuid [:map [:id ::sm/uuid]]]]])
(defmethod ig/init-key ::typed-job
[_ cfg]
{::jobs/name :typed
::jobs/schema schema:echo-params
::jobs/handler (fn [_context params]
(echo-handler cfg params))
::jobs/decoder (sm/decoder schema:echo-params sm/json-transformer)
::jobs/validator (sm/validator schema:echo-params)
;; the substrate compiles the decoder from this schema and checks the
;; result against it before storing it
::jobs/result-schema schema:typed-result})
(defn- get-job-defs
[]
(-> {::jobs/defs {:echo (ig/ref ::echo-job)}
::echo-job {::db/pool th/*pool*}}
(-> {::jobs/defs {:echo (ig/ref ::echo-job)
:typed (ig/ref ::typed-job)}
::echo-job {::db/pool th/*pool*}
::typed-job {::db/pool th/*pool*}}
(ig/expand)
(ig/init)
(get ::jobs/defs)))
@ -1188,11 +1209,12 @@
(defn- mk-running
"A job already claimed by a worker, ready for a terminal transition."
[cfg]
(let [job-id (jobs/submit cfg {::jobs/name :echo
::jobs/params (make-params)})]
(jobs/claim cfg job-id (:scheduled-at (jobs/get-job cfg job-id)))
job-id))
([cfg] (mk-running cfg :echo))
([cfg name]
(let [job-id (jobs/submit cfg {::jobs/name name
::jobs/params (make-params)})]
(jobs/claim cfg job-id (:scheduled-at (jobs/get-job cfg job-id)))
job-id)))
(t/deftest lifecycle-writers-require-metrics
;; Every one of the six writers must check the metrics instance before
@ -1375,6 +1397,99 @@
(t/is (= 0 (jobs/cancel cfg job-id)))
(t/is (= ["cancelled"] (get-outcomes job-id))))))
(t/deftest get-job-decodes-the-result-with-the-schema-of-the-def
"A result with a declared schema comes back typed. Without it, the plain
JSON of the column keywordizes every key and leaves every uuid as text;
without the substrate compiling the decoder from that schema, the key of
the uuid map comes back as a keyword and this test fails."
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg :typed)
file-id (uuid/next)
lib-id (uuid/next)]
(t/is (= 1 (jobs/complete cfg :job-id job-id
:result {:file-ids [lib-id]
:resolution {file-id {:id file-id}}})))
(let [result (:result (jobs/get-job cfg job-id))]
(t/testing "the uuid keyed map keeps the type of its key and values"
(t/is (= [file-id] (keys (:resolution result))))
(t/is (uuid? (:id (get (:resolution result) file-id)))))
(t/testing "the vector of uuids keeps the type of its elements"
(t/is (= [lib-id] (:file-ids result)))))))
(t/deftest get-job-leaves-a-result-without-a-schema-untouched
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg)
file-id (uuid/next)]
(t/is (= 1 (jobs/complete cfg :job-id job-id
:result {:resolution {file-id {:id file-id}}})))
(t/testing "the raw JSON shape survives: keywordized keys, uuid as text"
(let [result (:result (jobs/get-job cfg job-id))
key (keyword (str file-id))]
(t/is (= [key] (keys (:resolution result))))
(t/is (= (str file-id) (get-in result [:resolution key :id])))))))
(t/deftest get-job-reads-a-row-whose-job-def-is-gone
(let [job-id (mk-running (make-cfg (get-job-defs)))
;; a registry that no longer knows the kind of the row (a retired
;; job-def): reading its result must not raise
row (jobs/get-job (make-cfg {}) job-id)]
(t/is (some? row))
(t/is (= "echo" (:name row)))))
(t/deftest complete-refuses-a-result-that-does-not-conform
"The schema a job-def declares is checked before the result is stored: the
row keeps no result and the job is not completed."
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg :typed)]
(t/is (thrown-with-msg? Exception #"invalid result of a job"
(jobs/complete cfg :job-id job-id
:result {:resolution "not-a-map"})))
(t/testing "the row was not completed and holds no result"
(let [row (jobs/get-job cfg job-id)]
(t/is (= "running" (:status row)))
(t/is (nil? (:result row)))))
(t/testing "a job that is no longer completable is skipped without checking"
(th/db-update! :job {:status "aborted"} {:id job-id})
(t/is (zero? (jobs/complete cfg :job-id job-id
:result {:resolution "not-a-map"}))))))
(t/deftest a-stored-result-that-does-not-conform-is-read-decode-only
"A row written by another version may hold a result the schema of today no
longer accepts: reading it decodes and never raises."
(let [cfg (make-cfg (get-job-defs))
job-id (mk-running cfg :typed)]
(th/db-update! :job
{:status "completed"
:result (db/json {:resolution "not-a-map"})}
{:id job-id})
(let [row (jobs/get-job cfg job-id)]
(t/is (= "completed" (:status row)))
(t/is (= {:resolution "not-a-map"} (:result row))))))
(t/deftest decode-result-accepts-a-pgobject
(let [job-def (jobs/get-job-def (get (make-cfg (get-job-defs)) ::jobs/defs) :typed)
file-id (uuid/next)]
(t/testing "a raw column value is decoded with the schema of the def"
(let [decoded (jobs/decode-result job-def
(db/json {:file-ids [file-id]
:resolution {file-id {:id file-id}}}))]
(t/is (= [file-id] (:file-ids decoded)))
(t/is (= [file-id] (keys (:resolution decoded))))))
(t/testing "no result answers nil"
(t/is (nil? (jobs/decode-result job-def nil))))))
(t/deftest aborted-job-is-terminal-for-all-writers
"An aborted job (system-side terminal set by the orphan sweep) cannot be
moved by any lifecycle writer: first-terminal-wins, with no event and no

31
backend/test/backend_tests/tasks/export_binfile_test.clj
View File

@ -11,10 +11,6 @@
[app.binfile.common :as bfc]
[app.binfile.v3 :as v3]
[app.common.exceptions :as ex]
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.jobs :as jobs]
[app.storage :as sto]
[app.storage.tmp :as tmp]
@ -46,25 +42,6 @@
(assoc ::bfc/input (-> fixture io/resource jio/file))
(v3/import-files!)))
(defn- make-job!
"A job row as the substrate writes it: running and owned by a profile."
[profile-id]
(let [id (uuid/next)]
(th/db-insert! :job {:id id
:name "export-binfile"
:tenant (cf/get :tenant)
:queue "binfile"
:params (db/json {})
:priority 100
:max-retries 0
:retry-num 0
:status "running"
:profile-id profile-id
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
id))
(defn- run-export
"Run the export handler of a job, as the runner does."
[job-id params]
@ -99,7 +76,7 @@
(t/deftest the-export-handler-stores-the-artifact-of-its-job
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
job-id (make-job! (:id profile))
job-id (th/create-job* :export-binfile {:profile-id (:id profile)})
result (run-export job-id {:file-ids #{file-id}
:export-type :detach-libraries})
object (sto/get-object (get-storage) (:resource-id result))]
@ -136,7 +113,7 @@
(let [owner (th/create-profile* 1)
other (th/create-profile* 2)
file-id (first (:file-ids (import-fixture! owner)))
job-id (make-job! (:id other))
job-id (th/create-job* :export-binfile {:profile-id (:id other)})
before (count-storage-objects)]
(t/testing "a job whose owner cannot read a file does not export it"
@ -154,7 +131,7 @@
(t/deftest the-export-handler-leaves-nothing-when-the-storage-fails
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
job-id (make-job! (:id profile))
job-id (th/create-job* :export-binfile {:profile-id (:id profile)})
before (count-storage-objects)]
(with-redefs [sto/put-object! (fn [& _]
@ -177,7 +154,7 @@
(t/deftest the-export-handler-leaves-nothing-behind-when-the-job-is-cancelled
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
job-id (make-job! (:id profile))
job-id (th/create-job* :export-binfile {:profile-id (:id profile)})
before (count-storage-objects)]
(th/db-update! :job {:status "cancelled"} {:id job-id})

74
backend/test/backend_tests/tasks/import_binfile_test.clj
View File

@ -9,11 +9,10 @@
it leaves behind when it fails."
(:require
[app.binfile.common :as bfc]
[app.binfile.jobs :as bfj]
[app.binfile.v1 :as v1]
[app.binfile.v3 :as v3]
[app.common.time :as ct]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.jobs :as jobs]
[app.jobs.storage :as js]
@ -65,27 +64,6 @@
:filename "package.penpot"
:mtype "application/zip"}))
(defn- make-job!
"A job row as the substrate writes it: running, owned by a profile and
holding the package it consumes."
[profile-id resource-id]
(let [id (uuid/next)]
(th/db-insert! :job {:id id
:name "import-binfile"
:tenant (cf/get :tenant)
:queue "binfile"
:params (db/json {})
:priority 100
:max-retries 0
:retry-num 0
:status "running"
:profile-id profile-id
:resource-id resource-id
:scheduled-at (ct/now)
:created-at (ct/now)
:modified-at (ct/now)})
id))
(defn- run-import
"Run the import handler of a job, as the runner does."
[job-id params]
@ -119,7 +97,8 @@
(t/deftest the-import-handler-imports-the-package-of-its-job
(let [profile (th/create-profile* 1)
staged (stage-package! profile (fixture-path))
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
result (run-import job-id {:project-id (:default-project-id profile)
:name "imported"
:version 3})]
@ -142,7 +121,8 @@
(let [profile (th/create-profile* 1)
project-id (:default-project-id profile)
staged (stage-package! profile (fixture-path))
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
before (project-modified-at project-id)]
(run-import job-id {:project-id project-id
@ -157,7 +137,8 @@
(let [profile (th/create-profile* 1)
file-id (first (:file-ids (import-fixture! profile)))
staged (stage-package! profile (export-to-v1! file-id))
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
result (run-import job-id {:project-id (:default-project-id profile)
:name "imported-v1"
:version 1})]
@ -171,7 +152,8 @@
staged (stage-package! other (fixture-path))
;; the package belongs to `other`, but the destination project
;; belongs to `owner`: an import there is not allowed
job-id (make-job! (:id other) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id other)
:resource-id (:resource-id staged)})
before (project-files (:default-project-id owner))]
(t/testing "a job that cannot edit the destination does not import it"
@ -189,7 +171,8 @@
(t/deftest a-job-that-was-already-cancelled-never-imports
(let [profile (th/create-profile* 1)
staged (stage-package! profile (fixture-path))
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
before (project-files (:default-project-id profile))]
(th/db-update! :job {:status "cancelled"} {:id job-id})
@ -207,7 +190,8 @@
(t/deftest an-import-cancelled-while-it-runs-leaves-nothing-behind
(let [profile (th/create-profile* 1)
staged (stage-package! profile (fixture-path))
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
before (project-files (:default-project-id profile))]
;; the user cancels while the import is running: the first progress
@ -230,7 +214,8 @@
(t/deftest a-package-that-is-not-a-penpot-file-fails-without-writing
(let [profile (th/create-profile* 1)
staged (stage-package! profile (fixture-path))
job-id (make-job! (:id profile) (:resource-id staged))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
before (project-files (:default-project-id profile))]
;; the bytes of the resource are replaced by something that is not a
@ -252,3 +237,32 @@
(t/testing "and the broken package is released too"
(t/is (some? (:deleted-at (input-row (:resource-id broken)))))))))
(t/deftest a-result-that-does-not-conform-leaves-nothing-behind
(let [profile (th/create-profile* 1)
project-id (:default-project-id profile)
staged (stage-package! profile (fixture-path))
job-id (th/create-job* :import-binfile {:profile-id (:id profile)
:resource-id (:resource-id staged)})
before (project-files project-id)]
;; the core does its work inside the transaction and answers with a result
;; the schema of the task does not accept: the check runs before the
;; transaction commits, so the work is rolled back with it
(with-redefs [bfj/import-files
(fn [tx-cfg _context _params]
(th/create-file* tx-cfg 7 {:profile-id (:id profile)
:project-id project-id
:is-shared false
:name "half imported"})
{:file-ids [(uuid/next)]
:resolution "not-a-map"})]
(t/testing "the handler refuses the result"
(t/is (thrown-with-msg? Exception #"invalid result of an import job"
(run-import job-id {:project-id project-id
:name "imported"
:version 3}))))
(t/testing "and the import is rolled back: the destination keeps no files"
(t/is (= before (project-files project-id)))))))

41
frontend/src/app/main/ui/dashboard/import.cljs
View File

@ -156,6 +156,20 @@
0
resolution))
(defn- pending-library-links
"The library links the wizard must create for the files of a resolution:
one per pending entry the user chose a candidate for. The file id comes
from the entry and never from the key of the map: what a key holds depends
on the result schema of the job-def that wrote it, while the wizard indexes
its selection by the id of the entry."
[resolution selection]
(->> (vals resolution)
(mapcat (fn [{:keys [id pending]}]
(keep (fn [entry]
(when-let [library-id (get-in selection [id (:id entry)])]
{:file-id id :library-id library-id}))
pending)))))
(defn- analyze-entries
[state entries]
(->> (imp/analyze entries)
@ -866,22 +880,17 @@
(mf/deps resolution selection on-finish-import)
(fn [event]
(dom/prevent-default event)
(let [slc selection]
;; For each file with pending candidates, link it to the selected libraries
(->> (rx/from (seq resolution))
(rx/merge-map
(fn [[file-id resolution-file]]
(->> (rx/from (:pending resolution-file))
(rx/merge-map
(fn [{:keys [id]}]
(when-let [selected-lib (get-in slc [file-id id])]
(link-files-to-library! [file-id] selected-lib)))))))
(rx/subs! (constantly nil)
(constantly nil)
(fn []
(st/emit! (modal/hide))
(when (fn? on-finish-import)
(on-finish-import))))))))
;; For each file with pending candidates, link it to the selected libraries
(->> (rx/from (pending-library-links resolution selection))
(rx/merge-map
(fn [{:keys [file-id library-id]}]
(link-files-to-library! [file-id] library-id)))
(rx/subs! (constantly nil)
(constantly nil)
(fn []
(st/emit! (modal/hide))
(when (fn? on-finish-import)
(on-finish-import)))))))
on-wizard-next
(mf/use-fn

2
frontend/test/frontend_tests/runner.cljs
View File

@ -115,6 +115,7 @@
[frontend-tests.ui.comment-input-ime-test]
[frontend-tests.ui.comments-clustering-test]
[frontend-tests.ui.comments-position-modifier-test]
[frontend-tests.ui.dashboard-import-linking-test]
[frontend-tests.ui.drawarea-test]
[frontend-tests.ui.ds-controls-numeric-input-test]
[frontend-tests.ui.error-boundary-test]
@ -271,6 +272,7 @@
'frontend-tests.ui.comment-input-ime-test
'frontend-tests.ui.comments-clustering-test
'frontend-tests.ui.comments-position-modifier-test
'frontend-tests.ui.dashboard-import-linking-test
'frontend-tests.ui.drawarea-test
'frontend-tests.ui.ds-controls-numeric-input-test
'frontend-tests.ui.error-boundary-test

128
frontend/test/frontend_tests/ui/dashboard_import_linking_test.cljs Normal file
View File

@ -0,0 +1,128 @@
;; 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 frontend-tests.ui.dashboard-import-linking-test
"The links the import wizard creates for a file of the package whose
library the auto-link could not resolve. The resolution map comes from the
job result, so the type of its keys is not something the wizard can rely
on: the file id of a link must come from the entry and never from the key
of the map."
(:require
[app.common.uuid :as uuid]
[app.main.ui.dashboard.import :as import]
[cljs.test :as t :include-macros true]))
(def ^:private pending-library-links
@#'import/pending-library-links)
(defn- pending-entry
"One library of the package the user has to resolve, with a candidate."
[library-id]
{:id library-id
:name "Tokens"
:candidates [{:id (uuid/next)
:name "Tokens"
:project-id (uuid/next)
:project-name "Design system"}]})
(defn- file-resolution
"One file of a resolution, with a single library pending resolution."
[file-id library-id]
{:id file-id
:name "card.penpot"
:done []
:pending [(pending-entry library-id)]})
(t/deftest the-file-id-of-a-link-comes-from-the-entry-and-not-from-the-key
(let [file-id (uuid/next)
old-lib (uuid/next)
new-lib (uuid/next)
;; the shape a result without a declared schema has: the key of the
;; map is a keyword and the entry carries the same id as text
resolution {(keyword (str file-id)) (file-resolution (str file-id) (str old-lib))}
selection {(str file-id) {(str old-lib) (str new-lib)}}]
(t/is (= [{:file-id (str file-id) :library-id (str new-lib)}]
(pending-library-links resolution selection)))))
(t/deftest the-links-are-the-same-with-a-typed-resolution
(let [file-id (uuid/next)
old-lib (uuid/next)
new-lib (uuid/next)
resolution {file-id (file-resolution file-id old-lib)}
selection {file-id {old-lib new-lib}}]
(t/is (= [{:file-id file-id :library-id new-lib}]
(pending-library-links resolution selection)))))
(t/deftest a-file-with-several-pending-libraries-creates-one-link-each
(let [file-id (uuid/next)
lib-a (uuid/next)
lib-b (uuid/next)
new-a (uuid/next)
new-b (uuid/next)
resolution {file-id {:id file-id
:name "card.penpot"
:done []
:pending [(pending-entry lib-a)
(pending-entry lib-b)]}}
selection {file-id {lib-a new-a
lib-b new-b}}]
(t/testing "both entries the user resolved become a link"
(t/is (= #{{:file-id file-id :library-id new-a}
{:file-id file-id :library-id new-b}}
(set (pending-library-links resolution selection)))))))
(t/deftest a-partial-selection-creates-only-the-chosen-link
(let [file-id (uuid/next)
lib-a (uuid/next)
lib-b (uuid/next)
new-a (uuid/next)
resolution {file-id {:id file-id
:name "card.penpot"
:done []
:pending [(pending-entry lib-a)
(pending-entry lib-b)]}}
selection {file-id {lib-a new-a}}]
(t/testing "the entry with no candidate chosen emits nothing"
(t/is (= [{:file-id file-id :library-id new-a}]
(pending-library-links resolution selection))))))
(t/deftest a-done-entry-creates-no-link
(let [file-id (uuid/next)
library (uuid/next)]
(t/testing "a file whose libraries were auto-linked has nothing to emit"
(t/is (= [] (pending-library-links
{file-id {:id file-id
:name "card.penpot"
:done [{:id library
:name "Tokens"
:linked-to library}]
:pending []}}
{file-id {library (uuid/next)}}))))))
(t/deftest pending-entries-without-a-selection-create-no-link
(let [file-id (uuid/next)
old-lib (uuid/next)]
(t/testing "an entry the user skipped creates nothing"
(t/is (= [] (pending-library-links
{file-id (file-resolution file-id old-lib)}
{file-id {}}))))
(t/testing "a file with nothing pending creates nothing"
(t/is (= [] (pending-library-links
{file-id {:id file-id
:name "card.penpot"
:done []
:pending []}}
{file-id {old-lib (uuid/next)}}))))
(t/testing "an empty resolution creates nothing"
(t/is (= [] (pending-library-links {} {}))))))