diff --git a/.serena/memories/backend/core.md b/.serena/memories/backend/core.md index ef916f93a4..e884cda4e4 100644 --- a/.serena/memories/backend/core.md +++ b/.serena/memories/backend/core.md @@ -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`. diff --git a/.serena/memories/backend/subtleties.md b/.serena/memories/backend/subtleties.md index 18af2954aa..c95dd5d174 100644 --- a/.serena/memories/backend/subtleties.md +++ b/.serena/memories/backend/subtleties.md @@ -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 diff --git a/.serena/memories/frontend/dashboard-viewer-subtleties.md b/.serena/memories/frontend/dashboard-viewer-subtleties.md index abd03e4f03..9c753d8875 100644 --- a/.serena/memories/frontend/dashboard-viewer-subtleties.md +++ b/.serena/memories/frontend/dashboard-viewer-subtleties.md @@ -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 diff --git a/backend/src/app/jobs.clj b/backend/src/app/jobs.clj index 19df02cc9b..caf19b6f4e 100644 --- a/backend/src/app/jobs.clj +++ b/backend/src/app/jobs.clj @@ -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 diff --git a/backend/src/app/rpc/commands/jobs.clj b/backend/src/app/rpc/commands/jobs.clj index 54c7c352ed..c3f590315b 100644 --- a/backend/src/app/rpc/commands/jobs.clj +++ b/backend/src/app/rpc/commands/jobs.clj @@ -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} diff --git a/backend/src/app/rpc/management/jobs.clj b/backend/src/app/rpc/management/jobs.clj index 1e71acb101..f5fc250c96 100644 --- a/backend/src/app/rpc/management/jobs.clj +++ b/backend/src/app/rpc/management/jobs.clj @@ -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 diff --git a/backend/src/app/tasks/import_binfile.clj b/backend/src/app/tasks/import_binfile.clj index 5dc286db08..754bb4bf25 100644 --- a/backend/src/app/tasks/import_binfile.clj +++ b/backend/src/app/tasks/import_binfile.clj @@ -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}) diff --git a/backend/test/backend_tests/binfile_jobs_test.clj b/backend/test/backend_tests/binfile_jobs_test.clj index 7b4619c96c..e03cb79421 100644 --- a/backend/test/backend_tests/binfile_jobs_test.clj +++ b/backend/test/backend_tests/binfile_jobs_test.clj @@ -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))))))) diff --git a/backend/test/backend_tests/helpers.clj b/backend/test/backend_tests/helpers.clj index 24759bae21..bbcc71173c 100644 --- a/backend/test/backend_tests/helpers.clj +++ b/backend/test/backend_tests/helpers.clj @@ -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)) diff --git a/backend/test/backend_tests/jobs_test.clj b/backend/test/backend_tests/jobs_test.clj index 5dee48bbb3..15c7e6d83b 100644 --- a/backend/test/backend_tests/jobs_test.clj +++ b/backend/test/backend_tests/jobs_test.clj @@ -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 diff --git a/backend/test/backend_tests/tasks/export_binfile_test.clj b/backend/test/backend_tests/tasks/export_binfile_test.clj index 80fcdedd36..40a67a549a 100644 --- a/backend/test/backend_tests/tasks/export_binfile_test.clj +++ b/backend/test/backend_tests/tasks/export_binfile_test.clj @@ -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}) diff --git a/backend/test/backend_tests/tasks/import_binfile_test.clj b/backend/test/backend_tests/tasks/import_binfile_test.clj index e2c619177f..eadc44104f 100644 --- a/backend/test/backend_tests/tasks/import_binfile_test.clj +++ b/backend/test/backend_tests/tasks/import_binfile_test.clj @@ -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))))))) diff --git a/frontend/src/app/main/ui/dashboard/import.cljs b/frontend/src/app/main/ui/dashboard/import.cljs index e03d67dbe9..347364ff67 100644 --- a/frontend/src/app/main/ui/dashboard/import.cljs +++ b/frontend/src/app/main/ui/dashboard/import.cljs @@ -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 diff --git a/frontend/test/frontend_tests/runner.cljs b/frontend/test/frontend_tests/runner.cljs index f742f81093..93bae2969f 100644 --- a/frontend/test/frontend_tests/runner.cljs +++ b/frontend/test/frontend_tests/runner.cljs @@ -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 diff --git a/frontend/test/frontend_tests/ui/dashboard_import_linking_test.cljs b/frontend/test/frontend_tests/ui/dashboard_import_linking_test.cljs new file mode 100644 index 0000000000..84d8585719 --- /dev/null +++ b/frontend/test/frontend_tests/ui/dashboard_import_linking_test.cljs @@ -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 {} {}))))))