diff --git a/CHANGELOG.md b/CHANGELOG.md index 3608993a..44da5d3b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Authenticated operators can now perform owner-scoped durable-job cancellation for `PENDING` and `RUNNING` work through an idempotent action that commits terminal `CANCELLED`, clears payload and lease state, stores only `cancellation_key_hash` plus a fixed code and timestamp, and returns stable RFC 9457 conflicts when success or failure already won. +- Cancellation-first races now make the former exact lease stale and roll back transactional target and response-ledger effects; success-first races remain `SUCCEEDED`, while same-key cancellation replays and different-key reuse fails closed. - Owner-scoped durable-job status responses now emit deterministic weak SHA-256 `ETag` validators; ordinary and wildcard `If-None-Match` requests return an empty RFC 9110 `304 Not Modified` response only after authenticated owner-safe lookup, while `Cache-Control: no-store` remains unchanged. - Active durable-job status responses now emit an RFC 9110 `Retry-After` delay for `PENDING` and `RUNNING` states only when local worker execution is enabled, derived from the bounded worker fixed-delay configuration with upward whole-second rounding; terminal states and intake-only maintenance mode omit the advisory. - The durable job pagination index now uses PostgreSQL `CREATE INDEX CONCURRENTLY` with a migration-local Flyway `executeInTransaction=false` companion configuration, preserving production writers and documenting invalid-index recovery and concurrent rollback. @@ -35,6 +37,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Transactional migration `V6__add_etl_job_cancellation.sql`, owner-safe cancellation API and replay model, exact lease-invalidation and cancellation-versus-success integration tests, plus rollout, incident, connector-limitation, and rollback evidence in `docs/operations/durable-job-cancellation.md`. - Deterministic ordinary, wildcard, changed-state, changed-failure-code, null-versus-empty, and unrelated-response conditional polling tests, complete controller Javadoc, privacy and rollback guidance, and APA 7th standards evidence in `docs/etl/durable-job-polling.md`. - Controller-scoped polling advice, deterministic active/terminal lifecycle tests, disabled-worker fail-closed behavior, sub-second rounding coverage, rollback guidance, and APA 7th standards evidence in `docs/etl/durable-job-polling.md`. - Owner-scoped durable job list models and HTTP contract, strict cursor and page-limit validation, one-extra-row next-page detection, the descriptive `etl_job_owner_pagination_index`, deterministic tenant-isolation and equal-timestamp tests, migration rollback guidance, and APA 7th standards evidence in `docs/etl/durable-job-intake.md`. @@ -74,6 +77,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- Cancellation stores only a principal-scoped SHA-256 replay identity and a fixed machine code, exposes no raw principal, key, hash, payload, lease, SQL, exception, or target detail, and requires an owner-matched conditional database update before reporting success. - Conditional status validators are SHA-256 digests of only the complete owner-authorized operator-safe representation; payloads, raw principals, idempotency keys, internal hashes, leases, SQL, and exception text remain excluded, and wildcard evaluation occurs only after owner-safe lookup. - Polling advice exposes only a bounded delay integer and is omitted when local execution is disabled or terminal; it never contains job, lease, principal, key, hash, payload, SQL, exception, target, or queue-depth data. - Job-list cursors contain only ordering keys, never authority or sensitive values; every page query independently binds the hashed authenticated principal, malformed and non-canonical cursors fail before database access, and list responses exclude payloads, principals, keys, hashes, lease identifiers, SQL, and exception text. diff --git a/docs/doctoring/durable-job-cancellation-key-domain-separation.md b/docs/doctoring/durable-job-cancellation-key-domain-separation.md new file mode 100644 index 00000000..5254eca1 --- /dev/null +++ b/docs/doctoring/durable-job-cancellation-key-domain-separation.md @@ -0,0 +1,88 @@ +# Durable-job cancellation replay identity domain separation + +## Decision + +mightyETL never stores a raw cancellation `Idempotency-Key`. It stores one lowercase SHA-256 replay identity computed from an explicit versioned domain, the authenticated-principal hash, the durable job identifier, and the normalized key: + +```text +SHA-256( + "mightyetl:durable-job-cancellation:v1:" + || principal_scope_hash + || ":" + || job_record_id + || ":" + || normalized_cancellation_key +) +``` + +This value is used only to prove that a later request addresses the same principal, the same job, and the same semantic cancellation key. It is not an authentication credential and grants no job authority; every transition and replay read independently binds the owner hash and job identifier in SQL. + +## Threat addressed + +Hashing the raw client key alone would hide its plaintext but preserve equality across every row. A database observer could correlate two jobs or tenants that reused the same cancellation key even though ordinary API representations never reveal the key or hash. + +The versioned contextual prefix and explicit principal/job components partition the replay identity. The same normalized raw key therefore produces: + +- the same stored hash for the same principal and job, preserving deterministic replay; +- a different stored hash for another job in the same principal namespace; +- a different stored hash for the same job-shaped identifier under another principal namespace; +- a different stored hash after a deliberate future domain-version change. + +The implementation does not claim to use cSHAKE, TupleHash, or another NIST SP 800-185 primitive. It uses the existing SHA-256 utility with an unambiguous fixed-layout contextual input. NIST SP 800-185 is cited as primary methodological evidence for customization and tuple/domain separation concepts, not as an implementation-conformance claim. + +## Compatibility boundary + +The exact domain string is persisted protocol behavior: + +```text +mightyetl:durable-job-cancellation:v1: +``` + +Changing it would make every existing cancelled row fail same-key replay comparison. A future `v2` requires an explicit migration and dual-read compatibility window or a documented replay-breaking release. Silent replacement of the prefix is prohibited. + +The current concatenation is unambiguous because: + +- `principal_scope_hash` is exactly 64 lowercase hexadecimal characters; +- the separator is a literal colon; +- `job_record_id` is the canonical UUID text form; +- the second separator is a literal colon; +- the normalized cancellation key follows the bounded safe-ASCII profile and is the final component. + +If a later version introduces variable-width or independently nested components, use explicit length prefixes or a tuple-hash construction rather than extending this layout informally. + +## Test-first evidence + +`EtlJobCancellationKeyDomainIntegrationTest` uses the same raw cancellation key for: + +1. two different jobs owned by one principal; +2. one job owned by another principal. + +The test requires three distinct 64-character stored hashes. Existing service-integration tests separately prove that a quoted and legacy-raw representation of the same key on the same job replay one cancellation, while a genuinely different key fails with `etl_job_cancellation_key_reused`. + +## Privacy and logging + +The raw cancellation key and resulting hash remain absent from: + +- HTTP response bodies and headers; +- RFC 9457 problem details; +- ordinary logs; +- metric labels; +- status, list, polling, and ETag representations; +- worker lease models. + +The hash is a replay identity stored in `cancellation_key_hash`; it is not safe to publish merely because it is one-way. Database access, backups, exports, and support tooling must treat it as internal pseudonymous security data. + +## Rollback + +Rolling application code back across this change can make same-key replay behavior inconsistent if an older binary derives a raw-key-only hash. Keep the domain-separated implementation deployed while rows created by it are active. A rollback requires either: + +- retaining the new comparison algorithm in the older release line; or +- a reviewed data migration with explicit compatibility evidence. + +Never rewrite hashes from user-supplied guesses and never log candidate keys while diagnosing replay mismatches. + +## References — APA 7th + +Kelsey, J., Chang, S., & Perlner, R. (2016). *SHA-3 derived functions: cSHAKE, KMAC, TupleHash, and ParallelHash* (NIST Special Publication 800-185). National Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-185 + +National Institute of Standards and Technology. (2025, March 12). *Decision to update FIPS 202 and revise SP 800-185*. https://csrc.nist.gov/news/2025/decision-to-update-fips-202-and-revise-sp-800-185 diff --git a/docs/etl/durable-job-intake.md b/docs/etl/durable-job-intake.md index c00121d7..419ed8ce 100644 --- a/docs/etl/durable-job-intake.md +++ b/docs/etl/durable-job-intake.md @@ -5,7 +5,8 @@ `POST /api/etl/jobs` creates a durable, authenticated-principal-scoped ETL job resource. A separate lease-fenced worker claims accepted jobs across replicas, replays or writes the durable response ledger, writes target rows, and commits terminal state atomically. Authenticated operators can also -list recent jobs in their own principal namespace through deterministic keyset pagination. +list recent jobs in their own principal namespace through deterministic keyset pagination, inspect +status, and cancel pending or running work through a database-owned terminal transition. Both intake and execution capabilities are fail-closed: @@ -122,6 +123,31 @@ The representation exposes only the opaque job identifier, stable lifecycle stat count, stable failure code where applicable, and timestamps. It excludes request payload, raw principal, raw submission key, internal hashes, lease identifiers, SQL, and response-ledger data. +## Cancel owned work + +```http +POST /api/etl/jobs/{job_record_id}/cancellation HTTP/1.1 +Authorization: Basic +Idempotency-Key: "70dc8b50-e8b2-4e1a-8c5f-d84814708a77" +``` + +The endpoint can move only an owner-matched `PENDING` or `RUNNING` row to terminal `CANCELLED`. The +same update clears the retained payload and every lease field, stores only a SHA-256 cancellation-key +identity, a fixed `etl_job_cancelled_by_owner` code, and a database timestamp, and then returns the +existing operator-safe status representation. It never persists the raw principal or raw key. + +A first transition returns `Idempotency-Replayed: false`; the same normalized key returns the existing +cancelled resource with `Idempotency-Replayed: true`. Another key returns +`422 etl_job_cancellation_key_reused`. A cancellation after committed success or failure returns +`409 etl_job_already_succeeded` or `409 etl_job_already_failed`. Malformed, missing, and foreign-owned +identifiers retain the same `404 etl_job_not_found` surface. + +The conditional database update is the authority. If cancellation commits before exact-lease success, +the former worker's success predicate updates zero rows and Spring rolls back target and +`etl_idempotency_records` effects. If success commits first, cancellation cannot rewrite it. The +complete operational race, rollout, incident, connector-limitation, and rollback contract is in +`docs/operations/durable-job-cancellation.md`. + ## Lifecycle and distribution Flyway migrations create descriptive multi-word `snake_case` objects: @@ -133,9 +159,11 @@ Flyway migrations create descriptive multi-word `snake_case` objects: `etl_job_claim_eligibility_index` for oldest eligible queue-like claims without blocking writers; - `V5__add_etl_job_owner_pagination_index.sql` concurrently adds `etl_job_owner_pagination_index` on `principal_scope_hash`, `created_at DESC`, and - `job_record_id DESC` for the exact owner-scoped ordering contract. + `job_record_id DESC` for the exact owner-scoped ordering contract; +- `V6__add_etl_job_cancellation.sql` transactionally adds `cancellation_key_hash`, + `cancellation_code`, `job_cancelled_at`, and the five-state terminal lifecycle constraints. -The stable lifecycle is `PENDING`, `RUNNING`, `SUCCEEDED`, and `FAILED`. +The stable lifecycle is `PENDING`, `RUNNING`, `SUCCEEDED`, `FAILED`, and `CANCELLED`. Each fixed-delay poll handles at most one job. PostgreSQL, not scheduler uniqueness, distributes work: @@ -149,11 +177,12 @@ Each fixed-delay poll handles at most one job. PostgreSQL, not scheduler uniquen 5. an existing matching response is replayed, or target rows and `etl_idempotency_records` are written; 6. `SUCCEEDED` is committed only for the exact unexpired claim in the same transaction. -If the lease is expired or superseded, the final transition fails and rolls back target and ledger -writes. An expired row can be reclaimed with a new claim identifier. A stale worker therefore cannot -commit duplicate target effects or terminalize a newer owner's work. +If the lease is expired, superseded, or cleared by cancellation, the final transition fails and rolls +back target and ledger writes. An expired row can be reclaimed with a new claim identifier. A stale +worker therefore cannot commit duplicate target effects, overwrite a cancelled job, or terminalize a +newer owner's work. -## Retry and failure behavior +## Retry, failure, and cancellation behavior Transient database failures return the job to `PENDING` while attempts remain. At the configured limit they become `FAILED` with `etl_target_unavailable`. Non-transient database failures use @@ -162,8 +191,10 @@ limit they become `FAILED` with `etl_target_unavailable`. Non-transient database validation retains its existing stable `etl_*` request code. Eligible rows already at the attempt limit use `etl_worker_attempts_exhausted`. -Every retry or terminal transition repeats the exact live lease predicate. A zero-row transition is -stale evidence and does not overwrite the authoritative owner. +Owner cancellation is not a worker failure. It commits `CANCELLED` with a fixed cancellation code and +no failure code. Every retry, success, failure, or cancellation transition repeats an authoritative +state and owner predicate. A zero-row transition is stale or concurrent evidence and does not +overwrite the committed outcome. ## Validation, privacy, and retention @@ -172,25 +203,27 @@ bounds. The complete body must be a JSON array, duplicate JSON fields are reject must be an object with a safe textual `id`, and normalized field names must remain unique. The database stores an opaque UUID, SHA-256 hashes of principal scope, semantic submission key, and -exact JSON text, the retained request payload while nonterminal, lifecycle and attempt fields, and -lease metadata while running. Raw principal names and raw idempotency keys are never persisted. +exact JSON text, the retained request payload while nonterminal, lifecycle and attempt fields, lease +metadata while running, and a hashed cancellation identity plus fixed code and timestamp only while +cancelled. Raw principal names and raw idempotency or cancellation keys are never persisted. The request payload inherits the source records' data classification. Database constraints require a payload for nonterminal rows and require it to be null for terminal rows. Success, deterministic -failure, attempts exhaustion, and non-retryable failure clear the payload in their terminal -transition. Apply least privilege, encryption, backup, restore, and retention controls while data is -retained. - -List and status representations exclude payloads, raw principals, raw keys, hashes, lease identifiers, -SQL, and exception messages. Metrics and ordinary logs must not include those values or unbounded -error classes. Operational procedures and metric contracts are authoritative in -`docs/operations/durable-job-worker.md`. Claim-index deployment and invalid-index recovery are +failure, attempts exhaustion, non-retryable failure, and cancellation clear the payload in their +terminal transition. Apply least privilege, encryption, backup, restore, and retention controls while +data is retained. + +List, status, and cancellation representations exclude payloads, raw principals, raw keys, hashes, +lease identifiers, SQL, and exception messages. Metrics and ordinary logs must not include those +values or unbounded error classes. Worker procedures and metric contracts are authoritative in +`docs/operations/durable-job-worker.md`; cancellation procedures are in +`docs/operations/durable-job-cancellation.md`. Claim-index deployment and invalid-index recovery are specified in `docs/operations/durable-job-claim-index-rollout.md`. ## Migration and rollback -Apply Flyway migrations in version order. V3 remains transactional so lease columns, legacy-data -repair, and lifecycle constraints commit together. V4 and V5 are isolated additive index migrations. +Apply Flyway migrations in version order. V3 and V6 remain transactional so lease or cancellation +columns and lifecycle constraints commit together. V4 and V5 are isolated additive index migrations. Both use PostgreSQL `CREATE INDEX CONCURRENTLY` so inserts, updates, and deletes remain available while the indexes are built. Their matching `.sql.conf` files contain `executeInTransaction=false` because PostgreSQL rejects concurrent index creation inside a transaction block. The application also sets @@ -219,10 +252,17 @@ cause an unacceptable owner-list scan cost. Dropping the claim index while worke cause expensive claim scans and contention. Do not remove either index until its traffic is withdrawn and an execution-plan review confirms the rollback boundary. +V6 is not backward-compatible while cancelled rows exist because an older binary recognizes only the +four earlier states. Stop serving cancellation before application rollback. Preserve the V6 schema +until cancelled rows and audit evidence have been archived under an approved retention policy. Never +silently map `CANCELLED` to `FAILED` or `SUCCEEDED`, and never edit an applied Flyway migration in +place. + ## Standards basis - RFC 9110 Section 15.3.3 defines `202 Accepted` as noncommittal and recommends a current-status - representation and status monitor. + representation and status monitor; Section 15.5.10 defines `409 Conflict` for a request that + conflicts with the target resource's current state. - RFC 8288 defines the Web Linking model and the HTTP `Link` header used for the optional next-page relationship. - RFC 9457 supplies deterministic problem-details representations. @@ -235,6 +275,8 @@ and an execution-plan review confirms the rollback boundary. construction preserves writes with additional scans, waits, and invalid-index recovery caveats. - PostgreSQL 18 documents `SKIP LOCKED` as unsuitable for a general consistent view but useful for avoiding contention among multiple consumers of a queue-like table. +- PostgreSQL 18 documents that `UPDATE` acquires row-level writer exclusion and reports only rows that + satisfy the authoritative predicate. - Flyway script configuration supports a migration-matched `.sql.conf` file and the `executeInTransaction=false` override required for non-transactional PostgreSQL DDL. - Flyway's PostgreSQL integration documents session-level migration locking for statements such as @@ -251,7 +293,7 @@ https://www.rfc-editor.org/rfc/rfc9110 Nottingham, M. (2017). *Web linking* (RFC 8288). RFC Editor. https://doi.org/10.17487/RFC8288 -Nottingham, M., & Wilde, E. (2023). *Problem details for HTTP APIs* (RFC 9457). RFC Editor. +Nottingham, M., Wilde, E., & Dalal, S. (2023). *Problem details for HTTP APIs* (RFC 9457). RFC Editor. https://www.rfc-editor.org/rfc/rfc9457 Nottingham, M., & Kamp, P. (2024). *Structured field values for HTTP* (RFC 9651). RFC Editor. @@ -267,8 +309,11 @@ https://www.postgresql.org/docs/18/sql-createindex.html PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Introduction to indexes*. https://www.postgresql.org/docs/18/indexes-intro.html +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: UPDATE*. +https://www.postgresql.org/docs/18/sql-update.html + Redgate Software. (2026). *Flyway PostgreSQL transactional lock setting*. https://documentation.red-gate.com/fd/flyway-postgresql-transactional-lock-setting-277579114.html Redgate Software. (2026). *Flyway script configuration*. -https://documentation.red-gate.com/flyway/reference/script-configuration +https://documentation.red-gate.com/flyway/reference/script-configuration \ No newline at end of file diff --git a/docs/operations/durable-job-cancellation.md b/docs/operations/durable-job-cancellation.md new file mode 100644 index 00000000..f533976c --- /dev/null +++ b/docs/operations/durable-job-cancellation.md @@ -0,0 +1,244 @@ +# Durable ETL job cancellation + +## Purpose + +Authenticated operators can stop a durable ETL job that is still `PENDING` or `RUNNING` through: + +```http +POST /api/etl/jobs/{job_record_id}/cancellation HTTP/1.1 +Authorization: Basic +Idempotency-Key: "70dc8b50-e8b2-4e1a-8c5f-d84814708a77" +``` + +A successful response proves that the database transition committed or that the same principal, +job identifier, and normalized cancellation key had already committed. HTTP request acceptance by +itself is never treated as cancellation success. + +The first slice establishes a database-owned terminal state. It does not forcibly terminate a Java +thread, interrupt arbitrary connector computation, or compensate a non-transactional external +warehouse. + +## HTTP contract + +A first cancellation returns: + +```http +HTTP/1.1 200 OK +Cache-Control: no-store +Idempotency-Replayed: false +ETag: W/"" +Content-Type: application/json + +{ + "jobRecordId": "cf4f083f-8c90-4f34-a8b6-b53761de44ef", + "jobStatus": "CANCELLED", + "attemptCount": 1, + "createdAt": "2026-08-05T01:00:00Z", + "updatedAt": "2026-08-06T03:00:00Z" +} +``` + +Repeating the same semantic key returns `Idempotency-Replayed: true` and the same terminal resource. +A different key for the already-cancelled job returns +`422 etl_job_cancellation_key_reused`. Malformed, missing, and foreign-owned identifiers share the +same `404 etl_job_not_found` response. Cancellation after committed success or failure returns +`409 etl_job_already_succeeded` or `409 etl_job_already_failed` under RFC 9110 current-state conflict +semantics. + +Every success and covered failure uses `Cache-Control: no-store`. The response exposes no payload, +principal, cancellation key, key hash, lease identifier, SQL, exception message, or target identity. + +## State machine + +```mermaid +stateDiagram-v2 + [*] --> PENDING + PENDING --> RUNNING: lease-fenced claim + PENDING --> CANCELLED: owner cancellation + RUNNING --> PENDING: exact-lease retry + RUNNING --> SUCCEEDED: target + ledger + exact-lease commit + RUNNING --> FAILED: exact-lease terminal failure + RUNNING --> CANCELLED: owner cancellation wins + CANCELLED --> CANCELLED: same-key replay +``` + +`SUCCEEDED`, `FAILED`, and `CANCELLED` are terminal. Cancelled status never receives `Retry-After`. +Its state and `updatedAt` value also invalidate every earlier weak status `ETag`. + +## Database authority + +`EtlJobService.cancelOwned` first validates the job identifier, cancellation key, and authenticated +principal, then performs one conditional update inside a Spring transaction: + +```sql +UPDATE etl_job_records +SET job_status = 'CANCELLED', + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + cancellation_key_hash = ?, + cancellation_code = 'etl_job_cancelled_by_owner', + job_cancelled_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP +WHERE job_record_id = ? + AND principal_scope_hash = ? + AND job_status IN ('PENDING', 'RUNNING'); +``` + +The update count is the authority. A follow-up owner-scoped read classifies a zero-row result as: + +- identical `CANCELLED` replay; +- conflicting cancellation key; +- already `SUCCEEDED`; +- already `FAILED`; +- an active row whose concurrent transition is still unresolved; or +- owner-safe not found. + +The raw principal and key are never stored. Their lowercase SHA-256 values are used only for owner +selection and replay identity. + +## Cancellation-versus-success race + +### Cancellation commits first + +```mermaid +sequenceDiagram + participant C as Cancellation request + participant D as PostgreSQL + participant W as Worker transaction + C->>D: conditional PENDING/RUNNING → CANCELLED + D-->>C: one row committed + W->>D: target + response ledger writes + W->>D: markSucceeded(exact former lease) + D-->>W: zero rows updated + W-->>D: rollback target + ledger +``` + +The cancelled row has no lease fields. The former worker's exact-live-lease success predicate updates +zero rows and raises `StaleEtlJobLeaseException`; Spring rolls back its target and +`etl_idempotency_records` writes. + +### Success commits first + +```mermaid +sequenceDiagram + participant W as Worker transaction + participant D as PostgreSQL + participant C as Cancellation request + W->>D: target + ledger + SUCCEEDED commit + C->>D: conditional PENDING/RUNNING → CANCELLED + D-->>C: zero rows updated + C->>D: owner-scoped terminal read + D-->>C: SUCCEEDED + C-->>C: 409 etl_job_already_succeeded +``` + +Exactly one terminal state wins. The endpoint never rewrites `SUCCEEDED` or `FAILED` as cancelled. + +## Migration + +`V6__add_etl_job_cancellation.sql` adds these descriptive multi-word `snake_case` columns: + +- `cancellation_key_hash`; +- `cancellation_code`; +- `job_cancelled_at`. + +It replaces lifecycle checks so that: + +- `PENDING` and `RUNNING` retain a payload; +- `SUCCEEDED`, `FAILED`, and `CANCELLED` have no payload; +- only `RUNNING` has lease fields; +- only `FAILED` has `failure_code`; +- only `CANCELLED` has the three cancellation fields; +- hash and code values satisfy bounded fixed formats. + +The migration is transactional. Before production rollout, rehearse it against a representative +PostgreSQL 18 copy and confirm that no out-of-contract legacy row violates the replacement checks. +Monitor migration duration, lock wait, transaction age, replication lag, and application error rates. + +## Rollout + +1. Verify PR exact-head CI on Ubuntu, macOS, and Windows with no skipped project test. +2. Verify dependency review, CycloneDX SBOM, SAST, security scan, unresolved-thread, and independent + current-head approval gates. +3. Apply Flyway V6 before serving the new route. +4. Keep `mightyetl.etl.jobs.intake-enabled=false` during a conservative schema-only rollout if the + deployment process cannot guarantee application/schema ordering. +5. Enable the new application build and perform an owner-isolation smoke test with a disposable job. +6. Verify a first cancellation, same-key replay, different-key rejection, and a status read. +7. Confirm cancelled rows have null payload and lease fields and a fixed cancellation code. +8. Observe worker `stale` outcomes during deliberate running-job cancellation; this is expected + fencing evidence, not a duplicate-execution success. + +## Monitoring + +The cancellation endpoint uses fixed observation name `etl.jobs.cancel`. Do not attach job IDs, +principals, raw keys, key hashes, lease IDs, payloads, SQL, exception classes, messages, target +identities, or queue depth as metric labels. + +Monitor at least: + +- request rate and HTTP outcome count; +- cancellation latency; +- database update latency and lock waits; +- worker `stale` outcome changes; +- cancellation replay and key-conflict rate; +- cancelled rows retaining payload or lease fields, which must remain zero; +- target or ledger effects associated with cancellation-first tests, which must remain zero. + +## Incident response + +### Cancellation returns `etl_job_cancellation_in_progress` + +Re-read the owner-scoped status. A concurrent claim, retry, success, failure, or cancellation may have +won after the request's conditional update. Do not retry with a new idempotency key until the current +terminal or active state is understood. + +### Worker reports stale after cancellation + +This is the expected safety outcome when cancellation invalidates a running lease. Confirm the target +and response-ledger transaction rolled back. Repeated stale outcomes without operator cancellations +may indicate lease expiry, another worker, or database clock/latency problems. + +### Cancelled row retains payload or lease data + +Treat this as a high-severity lifecycle integrity incident. Stop intake and workers, preserve the row +and transaction evidence, verify the deployed schema constraints and application SHA, and do not +manually rewrite the state until the root cause and rollback effects are understood. + +## Rollback + +Stop serving the cancellation endpoint before application rollback. Older binaries do not understand +`CANCELLED`, so they must not read or process cancelled rows as if only four states existed. + +Do not drop V6 columns or restore the old status constraint while any cancelled row remains. A +controlled database rollback must first archive cancelled resources and their audit evidence under an +approved retention policy. Never silently map `CANCELLED` to `FAILED` or `SUCCEEDED`. + +After cancelled rows are safely removed and every older binary is deployed, an explicit reviewed +migration may drop the V6 constraints and columns and restore the four-state lifecycle. Do not edit or +repair the applied V6 migration file in place. + +## Connector limitation + +The cancellation-first rollback guarantee is valid for target and response-ledger writes that join the +same transaction and database as the job state. A remote warehouse, file upload, external API, or +message broker that cannot participate in that transaction requires connector-native cancellation, +idempotency, or compensation before the same guarantee can be advertised. This release deliberately +does not claim arbitrary external side-effect reversal. + +## References — APA 7th + +Fielding, R., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC 9110). RFC Editor. +https://www.rfc-editor.org/rfc/rfc9110 + +Nottingham, M., Wilde, E., & Dalal, S. (2023). *Problem details for HTTP APIs* (RFC 9457). RFC Editor. +https://www.rfc-editor.org/rfc/rfc9457 + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Data consistency checks at +the application level*. https://www.postgresql.org/docs/18/applevel-consistency.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: UPDATE*. +https://www.postgresql.org/docs/18/sql-update.html diff --git a/docs/superpowers/plans/2026-08-06-durable-job-cancellation.md b/docs/superpowers/plans/2026-08-06-durable-job-cancellation.md new file mode 100644 index 00000000..ad29cdb5 --- /dev/null +++ b/docs/superpowers/plans/2026-08-06-durable-job-cancellation.md @@ -0,0 +1,223 @@ +# Durable ETL Job Cancellation Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add owner-safe, idempotent, lease-fenced cancellation for pending and running durable ETL jobs. + +**Architecture:** Extend the existing durable job service with one transactional conditional-update authority. Persist only hashed cancellation identity and fixed machine codes, keep the current operator-safe response model, and let existing exact-lease predicates roll back a worker transaction when cancellation wins. Add a PostgreSQL migration, HTTP endpoint, deterministic integration tests, and standards-backed operations documentation. + +**Tech Stack:** Java 25, Spring Framework transaction management, Spring MVC, JdbcTemplate, PostgreSQL 18, H2 integration tests, JUnit 5, Mockito, JaCoCo, Maven. + +## Global Constraints + +- Preserve standalone operation and modular MSA integration. +- Do not modify the existing review agent, provider configuration, or credential names. +- Do not use `COPILOT_GITHUB_TOKEN`. +- Every introduced database object name contains at least two descriptive words and uses `snake_case`. +- Raw principals, cancellation keys, payloads, hashes, lease identifiers, SQL, and exception messages never enter client responses, logs, or metric tags. +- Added production statement and branch coverage remains 100%. +- Every public production API has beginner-readable Javadoc. +- No project test may be skipped. +- Update `CHANGELOG.md` and APA 7th doctoring before merge. + +--- + +### Task 1: Lock the cancellation state and migration contract + +**Files:** +- Modify: `etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java` +- Create: `etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql` + +**Interfaces:** +- Consumes: existing `etl_job_records` status, payload, lease, and failure constraints. +- Produces: `CANCELLED`, `cancellation_key_hash`, `cancellation_code`, and `job_cancelled_at` schema contract. + +- [ ] **Step 1: Write the failing migration assertions** + +Require the V6 file, `CANCELLED` lifecycle, cancellation hash/code/timestamp fields, terminal payload clearing, non-running lease clearing, format constraints, and explicit rollback guidance. + +- [ ] **Step 2: Run the focused test and observe failure** + +Run: + +```bash +./mvnw -B -pl etl-service -Dtest=EtlJobMigrationDocumentationTest test +``` + +Expected: failure because V6 does not exist. + +- [ ] **Step 3: Add the migration** + +Use `ALTER TABLE ... DROP CONSTRAINT ... ADD CONSTRAINT ...` so a clean install and an upgrade both converge on the same lifecycle invariants. + +- [ ] **Step 4: Run the focused test** + +Expected: pass. + +- [ ] **Step 5: Commit** + +```bash +git add etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql \ + etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java +git commit -m "feat(etl): add durable job cancellation schema" +``` + +### Task 2: Define the service-level cancellation contract test-first + +**Files:** +- Modify: `etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java` +- Create: `etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobCancellation.java` +- Modify: `etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java` +- Modify: `etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java` +- Modify: `etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java` + +**Interfaces:** +- Produces: `EtlJobCancellation cancelOwned(UUID, String, String)`. +- Produces: `EtlJobCancellation(EtlJobSnapshot snapshot, boolean replayed)`. + +- [ ] **Step 1: Add failing tests** + +Cover pending cancellation, payload clearing, same-key replay, quoted/raw key normalization, different-key rejection, owner isolation, missing identifier, succeeded conflict, failed conflict, running lease clearing, and invalid-key validation before JDBC access. + +- [ ] **Step 2: Run the focused integration test** + +Expected: compile or assertion failure because cancellation APIs and schema fields are absent. + +- [ ] **Step 3: Add `CANCELLED` and stable request errors** + +Add fixed RFC 9457 metadata for cancellation-key required/reused/in-progress and already-succeeded/already-failed conflicts. + +- [ ] **Step 4: Add the immutable cancellation result** + +Validate both fields and expose no persistence identity. + +- [ ] **Step 5: Implement one conditional-update authority** + +Validate inputs before database access, hash principal and normalized key, perform the PENDING/RUNNING update, then classify a zero-row result through one owner-scoped read. + +- [ ] **Step 6: Run the focused test** + +Expected: pass. + +- [ ] **Step 7: Commit** + +```bash +git add etl-service/src/main/java/com/xtrmetl/etl/job \ + etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java \ + etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java +git commit -m "feat(etl): cancel owner-scoped durable jobs" +``` + +### Task 3: Add the authenticated HTTP cancellation action + +**Files:** +- Modify: `etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java` +- Modify: `etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java` +- Modify: `etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java` + +**Interfaces:** +- Consumes: `EtlJobService.cancelOwned`. +- Produces: `POST /api/etl/jobs/{jobRecordId}/cancellation`. + +- [ ] **Step 1: Add failing controller tests** + +Cover first cancellation, replay header, authentication, missing key, malformed identifier, typed conflict, data-access failure, and unexpected failure. + +- [ ] **Step 2: Run focused controller tests** + +Expected: failure because the route is absent. + +- [ ] **Step 3: Implement the endpoint** + +Parse authentication and identifier before service access, preserve typed/data-access failures, wrap unexpected runtime failures, return `200`, `no-store`, weak ETag, and `Idempotency-Replayed`. + +- [ ] **Step 4: Run focused tests** + +Expected: pass. + +- [ ] **Step 5: Commit** + +```bash +git add etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java \ + etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java \ + etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java +git commit -m "feat(etl): expose durable job cancellation action" +``` + +### Task 4: Prove worker and representation compatibility + +**Files:** +- Modify: `etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseRepositoryIntegrationTest.java` +- Modify: `etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java` +- Modify: `etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobConditionalStatusTest.java` + +**Interfaces:** +- Consumes: existing exact-live-lease success predicate, polling advice, and ETag generation. +- Produces: regression evidence that cancellation is terminal and invalidates stale workers and validators. + +- [ ] **Step 1: Add a running-cancellation lease test** + +Claim a job, cancel its row through the service contract, and prove `markSucceeded` raises `StaleEtlJobLeaseException` with no terminal overwrite. + +- [ ] **Step 2: Add polling and conditional tests** + +Prove `CANCELLED` never emits `Retry-After` and the committed cancellation produces a different validator from the active representation. + +- [ ] **Step 3: Run focused tests** + +Expected: pass. + +- [ ] **Step 4: Commit** + +```bash +git add etl-service/src/test/java/com/xtrmetl/etl +git commit -m "test(etl): prove cancellation race and HTTP invariants" +``` + +### Task 5: Complete operations, changelog, and exact-head verification + +**Files:** +- Modify: `docs/etl/durable-job-intake.md` +- Create: `docs/operations/durable-job-cancellation.md` +- Modify: `CHANGELOG.md` +- Modify: `etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java` + +**Interfaces:** +- Produces: operator rollout/rollback, race, privacy, and connector-limitation evidence. + +- [ ] **Step 1: Update documentation tests first** + +Require endpoint, replay, conflicts, cancellation-first/success-first outcomes, transactional-target limitation, migration name, and rollback instructions. + +- [ ] **Step 2: Update authoritative documentation** + +Include Mermaid state/race diagrams, PostgreSQL locking behavior, RFC 9110/9457 mapping, telemetry, rollout, and rollback. + +- [ ] **Step 3: Update `CHANGELOG.md` under Unreleased** + +Record the buyer-visible cancellation action, terminal state, idempotency, lease invalidation, migration, and limitations. + +- [ ] **Step 4: Run all verification** + +```bash +./mvnw -B test +git diff --check +git status --short +``` + +Expected: every reactor module succeeds, JaCoCo configured production statement and branch coverage remains 100%, and no file is skipped. + +- [ ] **Step 5: Commit** + +```bash +git add docs CHANGELOG.md etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java +git commit -m "docs(etl): document durable job cancellation" +``` + +## Plan self-review + +- Every design requirement maps to a task. +- No raw principal or cancellation key crosses the persistence or response boundary. +- `EtlJobCancellation`, controller, migration, service status branches, worker stale outcome, polling behavior, and ETag invalidation all have explicit tests. +- Public signatures and names are consistent across tasks. +- No placeholder or deferred implementation instruction remains. diff --git a/docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md b/docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md new file mode 100644 index 00000000..d3c1a2e7 --- /dev/null +++ b/docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md @@ -0,0 +1,167 @@ +# Durable ETL Job Cancellation Design + +## Purpose + +mightyETL already accepts, executes, discovers, polls, and conditionally validates durable ETL jobs. Enterprise operators also need to stop work that is no longer wanted without leaking cross-tenant existence, weakening lease fencing, or claiming that a cancellation succeeded before the database transition committed. + +This design adds an authenticated owner-scoped cancellation action and one terminal `CANCELLED` lifecycle state. The first slice is deliberately limited to transactional target effects that participate in the same transaction as the durable response ledger and terminal job transition. + +## API contract + +```http +POST /api/etl/jobs/{job_record_id}/cancellation +Authorization: +Idempotency-Key: "client-generated-safe-key" +``` + +A successful first cancellation returns `200 OK`, the existing operator-safe status representation, `Cache-Control: no-store`, a weak `ETag`, and: + +```http +Idempotency-Replayed: false +``` + +Repeating the same principal, job identifier, and normalized cancellation key returns the same cancelled resource with `Idempotency-Replayed: true`. + +No request body or free-text reason is accepted in the first slice. This prevents unbounded sensitive text from entering persistence, logs, metrics, or problem details. + +## State machine + +```mermaid +stateDiagram-v2 + [*] --> PENDING + PENDING --> RUNNING: exact database claim + PENDING --> CANCELLED: owner cancellation + RUNNING --> SUCCEEDED: target + ledger + exact lease commit + RUNNING --> FAILED: exact lease terminal failure + RUNNING --> PENDING: exact lease retry release + RUNNING --> CANCELLED: owner cancellation wins row update + CANCELLED --> CANCELLED: same-key replay +``` + +`SUCCEEDED`, `FAILED`, and `CANCELLED` are terminal. A cancellation request against `SUCCEEDED` or `FAILED` returns an RFC 9457 problem with `409 Conflict`; it never rewrites a completed outcome. + +## Database authority + +One conditional `UPDATE` is the cancellation authority: + +```sql +UPDATE etl_job_records +SET job_status = 'CANCELLED', + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + cancellation_key_hash = ?, + cancellation_code = 'etl_job_cancelled_by_owner', + job_cancelled_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP +WHERE job_record_id = ? + AND principal_scope_hash = ? + AND job_status IN ('PENDING', 'RUNNING'); +``` + +The update count determines whether cancellation won. PostgreSQL row-update locking serializes competing writers. A follow-up owner-scoped read classifies a zero-row result as same-key replay, conflicting cancellation key, already-succeeded, already-failed, or owner-safe not-found. Read-then-write logic is never the authority. + +## Race outcomes + +### Cancellation commits first + +```text +owner cancellation updates PENDING or RUNNING to CANCELLED +→ payload and lease fields are cleared atomically +→ worker markSucceeded exact-live-lease predicate updates zero rows +→ StaleEtlJobLeaseException aborts the worker transaction +→ target and response-ledger writes roll back +→ final state CANCELLED +``` + +### Success commits first + +```text +worker target + response ledger + SUCCEEDED commit +→ cancellation conditional update affects zero rows +→ owner-scoped read observes SUCCEEDED +→ API returns 409 etl_job_already_succeeded +→ final state SUCCEEDED +``` + +Exactly one terminal outcome wins. + +## Persistence contract + +Migration `V6__add_etl_job_cancellation.sql` adds: + +- `cancellation_key_hash CHAR(64)`; +- `cancellation_code VARCHAR(128)`; +- `job_cancelled_at TIMESTAMPTZ`. + +All names contain multiple descriptive words and use `snake_case`. + +The migration replaces lifecycle constraints so that: + +- `CANCELLED` is a valid status; +- `PENDING` and `RUNNING` retain a non-null payload; +- `SUCCEEDED`, `FAILED`, and `CANCELLED` have a null payload; +- only `RUNNING` has lease fields; +- only `FAILED` has `failure_code`; +- only `CANCELLED` has all three cancellation fields; +- cancellation and identity hashes remain lowercase 64-character SHA-256 text. + +Raw principals and raw cancellation keys are never persisted. + +## Error taxonomy + +| HTTP | Stable code | Meaning | +| ---: | --- | --- | +| 400 | `etl_job_cancellation_key_required` | The key is absent or outside the supported bounded profile. | +| 404 | `etl_job_not_found` | The identifier is malformed, missing, or foreign-owned. | +| 409 | `etl_job_cancellation_in_progress` | An eligible row remained non-terminal after a failed authoritative update. | +| 409 | `etl_job_already_succeeded` | Success committed before cancellation. | +| 409 | `etl_job_already_failed` | Failure committed before cancellation. | +| 422 | `etl_job_cancellation_key_reused` | A cancelled job is replayed with a different cancellation key. | + +Problem responses use the existing RFC 9457 `application/problem+json` handler and contain no SQL, stack trace, exception message, principal, key, hash, payload, lease identifier, or target identity. + +## HTTP representation and caching + +The existing status representation remains the wire model. `jobStatus=CANCELLED` and the updated timestamp communicate the terminal outcome without exposing the cancellation key hash or internal code. The status `ETag` already covers lifecycle state and update time, so a committed cancellation invalidates every earlier validator. `Retry-After` remains absent because the polling advice emits it only for active states. + +## Observability + +The endpoint is annotated with the fixed observation name `etl.jobs.cancel`. No user-controlled tag is added. A successful response means the authoritative cancellation transition committed or an identical cancellation replay was proven; request acceptance alone is never reported as success. + +## Verification + +The exact-head suite must prove: + +1. pending cancellation prevents a later claim and clears the payload; +2. running cancellation clears lease fields and makes exact-lease success stale; +3. same-key cancellation replays one cancelled resource; +4. a different key fails with `etl_job_cancellation_key_reused`; +5. foreign-owned and missing identifiers share `etl_job_not_found`; +6. success-before-cancellation and failure-before-cancellation return stable conflicts; +7. malformed and absent keys fail before database access; +8. controller success, replay, authentication, malformed identifier, typed failure, database failure, and unexpected failure paths are covered; +9. `CANCELLED` receives no `Retry-After`; +10. conditional status validation changes after cancellation; +11. migration lifecycle, naming, privacy, and rollback documentation are tested; +12. every added production statement and branch remains covered with no skipped project test. + +## Operational limitation + +Cancellation invalidates the durable database lease and prevents transactional effects from committing. It does not forcibly terminate arbitrary computation. A connector whose target effects cannot join the mightyETL database transaction requires a separately designed compensation or connector-native cancellation contract before mightyETL may claim equivalent safety. + +## Rollback + +Application rollback must stop serving the cancellation endpoint before schema rollback. Retain the `CANCELLED` vocabulary while any cancelled row exists. A controlled data migration may archive cancelled rows before dropping cancellation columns and restoring older constraints. Never map cancelled rows to `FAILED` or `SUCCEEDED` silently. + +## References — APA 7th + +Fielding, R., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC 9110). RFC Editor. https://www.rfc-editor.org/rfc/rfc9110 + +Nottingham, M., Wilde, E., & Dalal, S. (2023). *Problem details for HTTP APIs* (RFC 9457). RFC Editor. https://www.rfc-editor.org/rfc/rfc9457 + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Data consistency checks at the application level*. https://www.postgresql.org/docs/18/applevel-consistency.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: UPDATE*. https://www.postgresql.org/docs/18/sql-update.html diff --git a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java index bb2c1e0e..8f5b6b08 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java @@ -1,6 +1,7 @@ package com.xtrmetl.etl.controller; import com.xtrmetl.etl.job.EtlJobAcceptedResponse; +import com.xtrmetl.etl.job.EtlJobCancellation; import com.xtrmetl.etl.job.EtlJobPage; import com.xtrmetl.etl.job.EtlJobPageResponse; import com.xtrmetl.etl.job.EtlJobService; @@ -35,7 +36,7 @@ import java.util.UUID; /** - * Exposes durable asynchronous ETL job submission, discovery, and status resources. + * Exposes durable asynchronous ETL job submission, discovery, status, and cancellation resources. * *

Submission requires authentication and an {@code Idempotency-Key}. The accepted response is * intentionally noncommittal under RFC 9110: it reports the durable pending state and supplies a @@ -45,11 +46,16 @@ * under RFC 8288 only when another page exists. The service independently binds every list query to * the authenticated principal hash, so cursor contents never grant authority.

* + *

Owner cancellation uses a separate principal-scoped {@code Idempotency-Key}. A successful + * response proves that the database cancellation transition committed or that the same semantic + * cancellation had already committed. It does not expose cancellation identity, lease data, or a + * retained request payload.

+ * *

Success and covered failure responses use {@code Cache-Control: no-store}. Malformed, absent, * and foreign-owned job identifiers use the same owner-safe not-found classification so the status - * endpoint does not become a cross-principal existence oracle. Successful status responses also - * carry a weak entity tag so an authenticated client can explicitly validate an unchanged - * representation without authorizing shared-cache persistence.

+ * and cancellation endpoints do not become cross-principal existence oracles. Successful status + * representations carry weak entity tags so an authenticated client can explicitly validate an + * unchanged representation without authorizing shared-cache persistence.

*/ @ConditionalOnBooleanProperty( prefix = "xtrmetl.etl.jobs", @@ -61,7 +67,7 @@ @RequestMapping("/api/etl/jobs") public class EtlJobController { - /** Response header indicating whether a prior durable submission was replayed. */ + /** Response header indicating whether a prior durable operation was replayed. */ public static final String IDEMPOTENCY_REPLAYED_HEADER = "Idempotency-Replayed"; private static final String DEFAULT_JOB_PAGE_LIMIT_TEXT = "50"; @@ -229,6 +235,60 @@ public ResponseEntity status( .body(responseBody); } + /** + * Cancels an owner-scoped pending or running durable job. + * + *

The response is successful only after the authoritative database transition committed or + * an identical principal-scoped cancellation replay was proven. A committed cancellation + * clears the payload and any active lease, so a worker cannot later commit through the former + * exact-lease predicate. Completed success or failure instead returns a stable conflict.

+ * + * @param jobRecordIdText opaque durable job identifier text + * @param idempotencyKey required client-generated cancellation key + * @param principal authenticated principal namespace + * @return cancelled operator-safe status and replay evidence + */ + @PostMapping("/{jobRecordId}/cancellation") + @Observed(name = "etl.jobs.cancel", contextualName = "etl-job-cancellation") + public ResponseEntity cancel( + @PathVariable("jobRecordId") String jobRecordIdText, + @RequestHeader(value = "Idempotency-Key", required = false) + @Nullable String idempotencyKey, + @Nullable Principal principal + ) { + if (principal == null) { + throw new EtlRequestException(EtlRequestError.IDEMPOTENCY_PRINCIPAL_REQUIRED); + } + if (idempotencyKey == null) { + throw new EtlRequestException(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED); + } + + UUID jobRecordId = parseJobRecordId(jobRecordIdText); + final EtlJobCancellation cancellation; + try { + cancellation = etlJobService.cancelOwned( + jobRecordId, + idempotencyKey, + principal.getName() + ); + } catch (EtlRequestException | DataAccessException exception) { + throw exception; + } catch (RuntimeException exception) { + throw new EtlUnexpectedException(exception); + } + + EtlJobStatusResponse responseBody = EtlJobStatusResponse.from(cancellation.snapshot()); + return ResponseEntity.ok() + .cacheControl(CacheControl.noStore()) + .eTag(statusEntityTag(responseBody)) + .header( + IDEMPOTENCY_REPLAYED_HEADER, + Boolean.toString(cancellation.replayed()) + ) + .contentType(MediaType.APPLICATION_JSON) + .body(responseBody); + } + /** * Builds an opaque weak validator from the complete status representation. * diff --git a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java index 3973f581..998139e5 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java @@ -20,9 +20,9 @@ * *

The advice is scoped to {@link EtlJobController}. When the durable worker is enabled, it * derives a positive whole-second {@code Retry-After} value from the validated fixed delay and - * rounds any fractional second upward. Pending and running jobs advertise that cadence; succeeded - * and failed jobs remove the header because their state is terminal. A disabled worker also removes - * the header so maintenance-mode intake does not imply that execution is progressing.

+ * rounds any fractional second upward. Pending and running jobs advertise that cadence; succeeded, + * failed, and cancelled jobs remove the header because their state is terminal. A disabled worker + * also removes the header so maintenance-mode intake does not imply that execution is progressing.

* *

The header is only client guidance. PostgreSQL lease fencing, principal-scoped authorization, * lifecycle validation, rate limiting, and client-side backoff remain independent correctness and @@ -124,7 +124,9 @@ public Object beforeBodyWrite( response.getHeaders().remove(HttpHeaders.RETRY_AFTER); } } - case SUCCEEDED, FAILED -> response.getHeaders().remove(HttpHeaders.RETRY_AFTER); + case SUCCEEDED, FAILED, CANCELLED -> response.getHeaders().remove( + HttpHeaders.RETRY_AFTER + ); } } return body; diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobCancellation.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobCancellation.java new file mode 100644 index 00000000..9f8bf074 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobCancellation.java @@ -0,0 +1,32 @@ +package com.xtrmetl.etl.job; + +import java.util.Objects; + +/** + * Reports one committed or replayed owner-scoped durable job cancellation. + * + *

The result contains only the existing operator-safe status snapshot and a replay flag. It + * never exposes the authenticated principal, raw cancellation key, cancellation-key hash, lease + * identity, request payload, SQL, or internal exception text.

+ * + * @param snapshot cancelled owner-authorized job representation + * @param replayed {@code true} when the same normalized cancellation key had already committed + */ +public record EtlJobCancellation(EtlJobSnapshot snapshot, boolean replayed) { + + /** + * Validates the immutable cancellation result. + * + * @param snapshot cancelled owner-authorized job representation + * @param replayed whether this response proves an earlier identical cancellation + */ + public EtlJobCancellation { + EtlJobSnapshot requiredSnapshot = Objects.requireNonNull( + snapshot, + "snapshot must not be null" + ); + if (requiredSnapshot.jobStatus() != EtlJobStatus.CANCELLED) { + throw new IllegalArgumentException("snapshot must be in CANCELLED state"); + } + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java index e1fde8be..91986f5d 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java @@ -31,7 +31,7 @@ import java.util.regex.Pattern; /** - * Creates, reads, and lists durable principal-scoped asynchronous ETL job resources. + * Creates, reads, lists, and cancels durable principal-scoped asynchronous ETL job resources. * *

The intake path validates the complete bounded JSON batch before persistence. Raw * authentication principals and idempotency keys are never stored. Instead, independent SHA-256 @@ -48,10 +48,22 @@ * opaque cursor is non-authoritative: it contains only the last returned ordering key, while the * principal hash remains an independent mandatory query predicate. Malformed or non-canonical * cursors fail closed before database access.

+ * + *

Cancellation is owned by one conditional database update. It can move only an owner-matched + * pending or running row to {@link EtlJobStatus#CANCELLED}, clearing the retained payload and every + * lease field atomically. A worker that tries to publish success afterward fails its exact-live- + * lease predicate and rolls back the target and response-ledger transaction. Raw cancellation keys + * are never persisted; replay compares only a domain-separated principal-and-job-scoped SHA-256 + * value so identical client keys cannot be correlated across jobs or tenants in storage.

*/ @Service public class EtlJobService { + /** Stable terminal code persisted when the authenticated owner cancels a durable job. */ + public static final String CANCELLED_BY_OWNER_CODE = "etl_job_cancelled_by_owner"; + + private static final String CANCELLATION_KEY_DOMAIN = + "mightyetl:durable-job-cancellation:v1:"; private static final String INSERT_JOB_SQL = """ INSERT INTO etl_job_records ( job_record_id, @@ -77,6 +89,29 @@ INSERT INTO etl_job_records ( WHERE job_record_id = ? AND principal_scope_hash = ? """; + private static final String SELECT_OWNED_CANCELLATION_SQL = """ + SELECT job_record_id, job_status, attempt_count, failure_code, + cancellation_key_hash, created_at, updated_at + FROM etl_job_records + WHERE job_record_id = ? + AND principal_scope_hash = ? + """; + private static final String CANCEL_OWNED_JOB_SQL = """ + UPDATE etl_job_records + SET job_status = 'CANCELLED', + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + cancellation_key_hash = ?, + cancellation_code = ?, + job_cancelled_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE job_record_id = ? + AND principal_scope_hash = ? + AND job_status IN ('PENDING', 'RUNNING') + """; private static final String SELECT_OWNED_JOB_PAGE_SQL = """ SELECT job_record_id, job_status, attempt_count, failure_code, created_at, updated_at @@ -195,7 +230,7 @@ public EtlJobSubmission submit( String validatedKey = validateIdempotencyKey(idempotencyKey); String validatedScope = validatePrincipalScope(principalScope); String validatedPayload = validatePayload(requestPayload); - requireActiveTransaction(); + requireActiveSubmissionTransaction(); String principalScopeHash = Sha256Digest.digest(validatedScope); String submissionKeyHash = Sha256Digest.digest(validatedKey); @@ -233,6 +268,81 @@ public EtlJobSubmission submit( return new EtlJobSubmission(jobRecordId, EtlJobStatus.PENDING, false); } + /** + * Cancels an owner-scoped pending or running job, or replays an identical cancellation. + * + *

The conditional update is the only cancellation authority. It changes no row after a + * terminal success or failure. Cancelling a running row clears its claim token, owner, expiry, + * and payload in the same transaction, so a concurrent worker cannot publish terminal success + * through the former lease. Replays compare a domain-separated principal-and-job-scoped SHA-256 + * key and return the already committed operator-safe status without rewriting the row.

+ * + * @param jobRecordId opaque durable job identifier + * @param cancellationKey required quoted Structured Field String or legacy raw safe value + * @param principalScope authenticated principal namespace + * @return newly committed or replayed cancelled status + * @throws NullPointerException when {@code jobRecordId} is {@code null} + * @throws EtlRequestException when authentication, key, ownership, or state contracts fail + * @throws IllegalStateException when invoked without an actual transaction + */ + @Transactional + public EtlJobCancellation cancelOwned( + UUID jobRecordId, + @Nullable String cancellationKey, + @Nullable String principalScope + ) { + UUID validatedJobId = Objects.requireNonNull(jobRecordId, "jobRecordId must not be null"); + String validatedKey = validateCancellationKey(cancellationKey); + String validatedScope = validatePrincipalScope(principalScope); + requireActiveCancellationTransaction(); + + String principalScopeHash = Sha256Digest.digest(validatedScope); + String cancellationKeyHash = Sha256Digest.digest( + CANCELLATION_KEY_DOMAIN + + principalScopeHash + + ':' + + validatedJobId + + ':' + + validatedKey + ); + int updatedRows = jdbcTemplate.update( + CANCEL_OWNED_JOB_SQL, + cancellationKeyHash, + CANCELLED_BY_OWNER_CODE, + validatedJobId, + principalScopeHash + ); + + StoredCancellation storedCancellation = findOwnedCancellation( + validatedJobId, + principalScopeHash + ); + if (storedCancellation == null) { + throw new EtlRequestException(EtlRequestError.JOB_NOT_FOUND); + } + if (updatedRows == 1) { + return new EtlJobCancellation(storedCancellation.snapshot(), false); + } + + return switch (storedCancellation.snapshot().jobStatus()) { + case CANCELLED -> { + if (!cancellationKeyHash.equals(storedCancellation.cancellationKeyHash())) { + throw new EtlRequestException( + EtlRequestError.JOB_CANCELLATION_KEY_REUSED + ); + } + yield new EtlJobCancellation(storedCancellation.snapshot(), true); + } + case SUCCEEDED -> throw new EtlRequestException( + EtlRequestError.JOB_ALREADY_SUCCEEDED + ); + case FAILED -> throw new EtlRequestException(EtlRequestError.JOB_ALREADY_FAILED); + case PENDING, RUNNING -> throw new EtlRequestException( + EtlRequestError.JOB_CANCELLATION_IN_PROGRESS + ); + }; + } + /** * Returns one job only when it belongs to the authenticated principal. * @@ -253,12 +363,14 @@ public EtlJobSnapshot findOwned( String principalScopeHash = Sha256Digest.digest(validatePrincipalScope(principalScope)); List jobs = jdbcTemplate.query( SELECT_OWNED_JOB_SQL, - (resultSet, rowNumber) -> mapSnapshot(resultSet.getObject("job_record_id", UUID.class), + (resultSet, rowNumber) -> mapSnapshot( + resultSet.getObject("job_record_id", UUID.class), resultSet.getString("job_status"), resultSet.getInt("attempt_count"), resultSet.getString("failure_code"), resultSet.getTimestamp("created_at"), - resultSet.getTimestamp("updated_at")), + resultSet.getTimestamp("updated_at") + ), validatedJobId, principalScopeHash ); @@ -344,6 +456,30 @@ private StoredJobRecord findSubmission(String principalScopeHash, String submiss return jobs.isEmpty() ? null : jobs.getFirst(); } + @Nullable + private StoredCancellation findOwnedCancellation( + UUID jobRecordId, + String principalScopeHash + ) { + List jobs = jdbcTemplate.query( + SELECT_OWNED_CANCELLATION_SQL, + (resultSet, rowNumber) -> new StoredCancellation( + mapSnapshot( + resultSet.getObject("job_record_id", UUID.class), + resultSet.getString("job_status"), + resultSet.getInt("attempt_count"), + resultSet.getString("failure_code"), + resultSet.getTimestamp("created_at"), + resultSet.getTimestamp("updated_at") + ), + resultSet.getString("cancellation_key_hash") + ), + jobRecordId, + principalScopeHash + ); + return jobs.isEmpty() ? null : jobs.getFirst(); + } + private static EtlJobSnapshot mapSnapshotRow( java.sql.ResultSet resultSet, int rowNumber @@ -532,6 +668,22 @@ private static String validateIdempotencyKey(@Nullable String idempotencyKey) { throw new EtlRequestException(EtlRequestError.INVALID_IDEMPOTENCY_KEY); } + private static String validateCancellationKey(@Nullable String cancellationKey) { + if (cancellationKey == null) { + throw new EtlRequestException(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED); + } + var structuredFieldMatcher = IDEMPOTENCY_KEY_STRUCTURED_FIELD_PROFILE.matcher( + cancellationKey + ); + if (structuredFieldMatcher.matches()) { + return structuredFieldMatcher.group(1); + } + if (IDEMPOTENCY_KEY_VALUE_PROFILE.matcher(cancellationKey).matches()) { + return cancellationKey; + } + throw new EtlRequestException(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED); + } + private static String validatePrincipalScope(@Nullable String principalScope) { if (principalScope == null || principalScope.isBlank() @@ -542,7 +694,7 @@ private static String validatePrincipalScope(@Nullable String principalScope) { return principalScope; } - private static void requireActiveTransaction() { + private static void requireActiveSubmissionTransaction() { if (!TransactionSynchronizationManager.isActualTransactionActive()) { throw new IllegalStateException( "Durable ETL job submission requires an active transaction" @@ -550,6 +702,14 @@ private static void requireActiveTransaction() { } } + private static void requireActiveCancellationTransaction() { + if (!TransactionSynchronizationManager.isActualTransactionActive()) { + throw new IllegalStateException( + "Durable ETL job cancellation requires an active transaction" + ); + } + } + private record StoredJobRecord(String requestDigest, EtlJobSnapshot snapshot) { private StoredJobRecord { Objects.requireNonNull(requestDigest, "requestDigest must not be null"); @@ -557,6 +717,15 @@ private record StoredJobRecord(String requestDigest, EtlJobSnapshot snapshot) { } } + private record StoredCancellation( + EtlJobSnapshot snapshot, + @Nullable String cancellationKeyHash + ) { + private StoredCancellation { + Objects.requireNonNull(snapshot, "snapshot must not be null"); + } + } + private record PageCursor(Instant createdAt, UUID jobRecordId) { private PageCursor { Objects.requireNonNull(createdAt, "createdAt must not be null"); diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java index 8e6fdd2c..8ae84f44 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java @@ -3,10 +3,9 @@ /** * Stable lifecycle states exposed by the asynchronous ETL job resource. * - *

This intake slice creates jobs only in {@link #PENDING}. The remaining values reserve the - * compatibility-safe state names required by the following worker and lease-fencing slice, so a - * deployed status reader can deserialize later transitions without a schema or API vocabulary - * change.

+ *

Pending and running jobs retain a bounded request payload. Succeeded, failed, and cancelled + * jobs are terminal and clear the payload. Exact database predicates, rather than scheduler or HTTP + * request timing, determine which terminal outcome wins.

*/ public enum EtlJobStatus { @@ -20,5 +19,8 @@ public enum EtlJobStatus { SUCCEEDED, /** The job reached a terminal failure and the retained payload was cleared. */ - FAILED + FAILED, + + /** The authenticated owner cancelled the job and invalidated any active lease. */ + CANCELLED } diff --git a/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java b/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java index a7786ee6..332f13f6 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java @@ -122,6 +122,51 @@ public enum EtlRequestError { "The job page cursor is invalid or no longer follows the supported opaque format." ), + /** The cancellation key is absent or outside the bounded safe idempotency profile. */ + JOB_CANCELLATION_KEY_REQUIRED( + HttpStatus.BAD_REQUEST, + "etl_job_cancellation_key_required", + "urn:mightyetl:problem:etl-job-cancellation-key-required", + "ETL job cancellation key required", + "Cancellation requires a supported principal-scoped Idempotency-Key." + ), + + /** A cancelled job was addressed with a different cancellation identity. */ + JOB_CANCELLATION_KEY_REUSED( + HttpStatus.UNPROCESSABLE_ENTITY, + "etl_job_cancellation_key_reused", + "urn:mightyetl:problem:etl-job-cancellation-key-reused", + "ETL job cancellation key reused", + "The durable job was already cancelled with a different Idempotency-Key." + ), + + /** An eligible job remained active after the authoritative cancellation update. */ + JOB_CANCELLATION_IN_PROGRESS( + HttpStatus.CONFLICT, + "etl_job_cancellation_in_progress", + "urn:mightyetl:problem:etl-job-cancellation-in-progress", + "ETL job cancellation in progress", + "The durable job cancellation could not yet establish a terminal outcome." + ), + + /** Durable success committed before the cancellation transition. */ + JOB_ALREADY_SUCCEEDED( + HttpStatus.CONFLICT, + "etl_job_already_succeeded", + "urn:mightyetl:problem:etl-job-already-succeeded", + "ETL job already succeeded", + "The durable job succeeded before cancellation could commit." + ), + + /** Durable failure committed before the cancellation transition. */ + JOB_ALREADY_FAILED( + HttpStatus.CONFLICT, + "etl_job_already_failed", + "urn:mightyetl:problem:etl-job-already-failed", + "ETL job already failed", + "The durable job failed before cancellation could commit." + ), + /** The requested job does not exist in the authenticated principal's namespace. */ JOB_NOT_FOUND( HttpStatus.NOT_FOUND, diff --git a/etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql b/etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql new file mode 100644 index 00000000..b7a1b7af --- /dev/null +++ b/etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql @@ -0,0 +1,81 @@ +-- Add owner-scoped durable job cancellation without retaining raw principals or keys. +ALTER TABLE etl_job_records + ADD COLUMN cancellation_key_hash CHAR(64), + ADD COLUMN cancellation_code VARCHAR(128), + ADD COLUMN job_cancelled_at TIMESTAMPTZ; + +-- Replace lifecycle constraints so clean installations and upgrades converge on one state machine. +ALTER TABLE etl_job_records + DROP CONSTRAINT etl_job_status_value_check, + DROP CONSTRAINT etl_job_payload_lifecycle_check, + DROP CONSTRAINT etl_job_lease_lifecycle_check, + DROP CONSTRAINT etl_job_failure_lifecycle_check; + +ALTER TABLE etl_job_records + ADD CONSTRAINT etl_job_status_value_check CHECK ( + job_status IN ('PENDING', 'RUNNING', 'SUCCEEDED', 'FAILED', 'CANCELLED') + ), + ADD CONSTRAINT etl_job_payload_lifecycle_check CHECK ( + ( + job_status IN ('PENDING', 'RUNNING') + AND request_payload IS NOT NULL + ) + OR + ( + job_status IN ('SUCCEEDED', 'FAILED', 'CANCELLED') + AND request_payload IS NULL + ) + ), + ADD CONSTRAINT etl_job_lease_lifecycle_check CHECK ( + ( + job_status = 'RUNNING' + AND lease_claim_id IS NOT NULL + AND lease_owner_id IS NOT NULL + AND lease_expires_at IS NOT NULL + ) + OR + ( + job_status <> 'RUNNING' + AND lease_claim_id IS NULL + AND lease_owner_id IS NULL + AND lease_expires_at IS NULL + ) + ), + ADD CONSTRAINT etl_job_failure_lifecycle_check CHECK ( + ( + job_status = 'FAILED' + AND failure_code IS NOT NULL + ) + OR + ( + job_status <> 'FAILED' + AND failure_code IS NULL + ) + ), + ADD CONSTRAINT etl_job_cancellation_key_hash_format CHECK ( + cancellation_key_hash IS NULL + OR cancellation_key_hash ~ '^[0-9a-f]{64}$' + ), + ADD CONSTRAINT etl_job_cancellation_code_format CHECK ( + cancellation_code IS NULL + OR cancellation_code ~ '^[a-z][a-z0-9_]{2,127}$' + ), + ADD CONSTRAINT etl_job_cancellation_code_value CHECK ( + cancellation_code IS NULL + OR cancellation_code = 'etl_job_cancelled_by_owner' + ), + ADD CONSTRAINT etl_job_cancellation_lifecycle_check CHECK ( + ( + job_status = 'CANCELLED' + AND cancellation_key_hash IS NOT NULL + AND cancellation_code IS NOT NULL + AND job_cancelled_at IS NOT NULL + ) + OR + ( + job_status <> 'CANCELLED' + AND cancellation_key_hash IS NULL + AND cancellation_code IS NULL + AND job_cancelled_at IS NULL + ) + ); diff --git a/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java index e21dd479..e51a4bc1 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java @@ -71,6 +71,7 @@ void terminalJobsRemoveAnyRetryAfterSuggestion() { assertTerminalHeaderRemoved(advice, EtlJobStatus.SUCCEEDED); assertTerminalHeaderRemoved(advice, EtlJobStatus.FAILED); + assertTerminalHeaderRemoved(advice, EtlJobStatus.CANCELLED); } @Test diff --git a/etl-service/src/test/java/com/xtrmetl/etl/documentation/DurableJobCancellationDocumentationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/documentation/DurableJobCancellationDocumentationTest.java new file mode 100644 index 00000000..8bfdffa4 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/documentation/DurableJobCancellationDocumentationTest.java @@ -0,0 +1,121 @@ +package com.xtrmetl.etl.documentation; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Keeps the durable cancellation API, race contract, replay identity, rollout, rollback, and + * changelog aligned. + */ +class DurableJobCancellationDocumentationTest { + + @Test + void operationsRunbookDocumentsTheAuthoritativeCancellationContract() throws IOException { + String runbook = read("docs/operations/durable-job-cancellation.md") + .replaceAll("\\s+", " "); + + assertTrue(runbook.contains("POST /api/etl/jobs/{job_record_id}/cancellation")); + assertTrue(runbook.contains("Idempotency-Replayed: false")); + assertTrue(runbook.contains("Idempotency-Replayed: true")); + assertTrue(runbook.contains("etl_job_cancellation_key_reused")); + assertTrue(runbook.contains("etl_job_already_succeeded")); + assertTrue(runbook.contains("etl_job_already_failed")); + assertTrue(runbook.contains("Cancellation commits first")); + assertTrue(runbook.contains("Success commits first")); + assertTrue(runbook.contains("StaleEtlJobLeaseException")); + assertTrue(runbook.contains( + "rolls back its target and `etl_idempotency_records` writes" + )); + assertTrue(runbook.contains("V6__add_etl_job_cancellation.sql")); + assertTrue(runbook.contains("mightyetl.etl.jobs.intake-enabled=false")); + assertTrue(runbook.contains("does not claim arbitrary external side-effect reversal")); + assertTrue(runbook.contains( + "Never silently map `CANCELLED` to `FAILED` or `SUCCEEDED`" + )); + assertTrue(runbook.contains("RFC 9110")); + assertTrue(runbook.contains("RFC 9457")); + assertTrue(runbook.contains("PostgreSQL Global Development Group. (2026)")); + } + + @Test + void designAndPlanKeepDatabaseAuthorityAndVerificationExplicit() throws IOException { + String design = read( + "docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md" + ).replaceAll("\\s+", " "); + String plan = read( + "docs/superpowers/plans/2026-08-06-durable-job-cancellation.md" + ).replaceAll("\\s+", " "); + + assertTrue(design.contains("One conditional `UPDATE` is the cancellation authority")); + assertTrue(design.contains("Exactly one terminal outcome wins")); + assertTrue(design.contains("same-key replay")); + assertTrue(design.contains("transactional target effects")); + assertTrue(plan.contains("Added production statement and branch coverage remains 100%")); + assertTrue(plan.contains("No project test may be skipped")); + assertTrue(plan.contains("Run all verification")); + } + + @Test + void doctoringDocumentsDomainSeparatedReplayIdentityAndCompatibility() throws IOException { + String evidence = read( + "docs/doctoring/durable-job-cancellation-key-domain-separation.md" + ).replaceAll("\\s+", " "); + + assertTrue(evidence.contains("mightyetl:durable-job-cancellation:v1:")); + assertTrue(evidence.contains("principal_scope_hash")); + assertTrue(evidence.contains("job_record_id")); + assertTrue(evidence.contains("normalized_cancellation_key")); + assertTrue(evidence.contains("same normalized raw key")); + assertTrue(evidence.contains("another job in the same principal namespace")); + assertTrue(evidence.contains("another principal namespace")); + assertTrue(evidence.contains("EtlJobCancellationKeyDomainIntegrationTest")); + assertTrue(evidence.contains("Changing it would make every existing cancelled row fail")); + assertTrue(evidence.contains("does not claim to use cSHAKE")); + assertTrue(evidence.contains("NIST Special Publication 800-185")); + } + + @Test + void changelogRecordsTheBuyerVisibleCancellationSlice() throws IOException { + String changelog = read("CHANGELOG.md").replaceAll("\\s+", " "); + + assertTrue(changelog.contains("owner-scoped durable-job cancellation")); + assertTrue(changelog.contains("CANCELLED")); + assertTrue(changelog.contains("cancellation_key_hash")); + assertTrue(changelog.contains("Cancellation-first")); + assertTrue(changelog.contains("transactional target and response-ledger effects")); + } + + private static String read(String relativePath) throws IOException { + return Files.readString(projectRoot().resolve(relativePath), StandardCharsets.UTF_8); + } + + /** + * Finds the reactor root from repository-root or module-scoped Maven execution. + * + * @return repository root containing the Maven reactor + */ + private static Path projectRoot() { + Path current = Paths.get(System.getProperty("user.dir")).toAbsolutePath(); + Path lastPomParent = null; + while (current != null) { + if (Files.exists(current.resolve(".git"))) { + return current; + } + if (Files.exists(current.resolve("pom.xml"))) { + lastPomParent = current; + } + current = current.getParent(); + } + if (lastPomParent != null) { + return lastPomParent; + } + throw new IllegalStateException("Could not find project root"); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationAtomicityIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationAtomicityIntegrationTest.java new file mode 100644 index 00000000..ceef2625 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationAtomicityIntegrationTest.java @@ -0,0 +1,251 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.support.TransactionTemplate; + +import javax.sql.DataSource; +import java.time.Duration; +import java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Proves that cancellation committed before exact-lease success rolls back target and ledger work. + */ +@SpringJUnitConfig(EtlJobCancellationAtomicityIntegrationTest.TestConfiguration.class) +class EtlJobCancellationAtomicityIntegrationTest { + + private static final String OWNER_ID = "worker-alpha"; + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + private static final String PRINCIPAL_SCOPE_HASH = "a".repeat(64); + private static final String SUBMISSION_KEY_HASH = "b".repeat(64); + private static final String REQUEST_DIGEST = "c".repeat(64); + + private final EtlJobLeaseRepository leaseRepository; + private final JdbcTemplate jdbcTemplate; + private final PlatformTransactionManager transactionManager; + + @Autowired + EtlJobCancellationAtomicityIntegrationTest( + EtlJobLeaseRepository leaseRepository, + JdbcTemplate jdbcTemplate, + PlatformTransactionManager transactionManager + ) { + this.leaseRepository = leaseRepository; + this.jdbcTemplate = jdbcTemplate; + this.transactionManager = transactionManager; + } + + @BeforeEach + void createTables() { + jdbcTemplate.execute("DROP TABLE IF EXISTS processed_data"); + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_idempotency_records"); + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_job_records"); + jdbcTemplate.execute(""" + CREATE TABLE etl_job_records ( + job_record_id UUID PRIMARY KEY, + principal_scope_hash CHAR(64) NOT NULL, + submission_key_hash CHAR(64) NOT NULL, + request_digest CHAR(64) NOT NULL, + request_payload CLOB, + job_status VARCHAR(32) NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE NOT NULL, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL + ) + """); + jdbcTemplate.execute(""" + CREATE TABLE processed_data ( + processed_record_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + data VARCHAR(8192) NOT NULL + ) + """); + jdbcTemplate.execute(""" + CREATE TABLE etl_idempotency_records ( + idempotency_key_hash CHAR(64) PRIMARY KEY, + request_digest CHAR(64) NOT NULL, + response_body CLOB NOT NULL, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + """); + } + + @Test + void cancellationCommitMakesSuccessStaleAndRollsBackEarlierEffects() { + UUID jobRecordId = insertPendingJob(); + EtlJobLease lease = leaseRepository.claimNext( + OWNER_ID, + Duration.ofMinutes(5), + 3 + ).orElseThrow(); + + EtlJobIdempotencyService idempotencyService = mock(EtlJobIdempotencyService.class); + when(idempotencyService.process(lease)).thenAnswer(invocation -> { + jdbcTemplate.update( + "INSERT INTO processed_data (data) VALUES (?)", + "ID:record_alpha," + ); + jdbcTemplate.update( + """ + INSERT INTO etl_idempotency_records ( + idempotency_key_hash, request_digest, response_body + ) VALUES (?, ?, ?) + """, + "d".repeat(64), + REQUEST_DIGEST, + "ID:record_alpha," + ); + cancelInIndependentTransaction(jobRecordId); + return "ID:record_alpha,"; + }); + EtlJobExecutionService executionService = new EtlJobExecutionService( + idempotencyService, + leaseRepository + ); + TransactionTemplate executionTransaction = new TransactionTemplate(transactionManager); + + assertThrows( + StaleEtlJobLeaseException.class, + () -> executionTransaction.executeWithoutResult( + status -> executionService.execute(lease) + ) + ); + + assertEquals(0, tableCount("processed_data")); + assertEquals(0, tableCount("etl_idempotency_records")); + assertEquals("CANCELLED", textColumn(jobRecordId, "job_status")); + assertNull(textColumn(jobRecordId, "request_payload")); + assertEquals( + EtlJobService.CANCELLED_BY_OWNER_CODE, + textColumn(jobRecordId, "cancellation_code") + ); + } + + private void cancelInIndependentTransaction(UUID jobRecordId) { + TransactionTemplate cancellationTransaction = new TransactionTemplate(transactionManager); + cancellationTransaction.setPropagationBehavior( + TransactionDefinition.PROPAGATION_REQUIRES_NEW + ); + cancellationTransaction.executeWithoutResult(status -> { + int cancelledRows = jdbcTemplate.update( + """ + UPDATE etl_job_records + SET job_status = 'CANCELLED', + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + cancellation_key_hash = ?, + cancellation_code = ?, + job_cancelled_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE job_record_id = ? + AND principal_scope_hash = ? + AND job_status IN ('PENDING', 'RUNNING') + """, + "e".repeat(64), + EtlJobService.CANCELLED_BY_OWNER_CODE, + jobRecordId, + PRINCIPAL_SCOPE_HASH + ); + if (cancelledRows != 1) { + throw new AssertionError("test cancellation did not update exactly one row"); + } + }); + } + + private UUID insertPendingJob() { + UUID jobRecordId = UUID.fromString("da381d2a-a4a5-41ad-a12a-9ac9b857193f"); + Instant now = Instant.parse("2026-08-06T00:00:00Z"); + jdbcTemplate.update( + """ + INSERT INTO etl_job_records ( + job_record_id, principal_scope_hash, submission_key_hash, + request_digest, request_payload, job_status, attempt_count, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, 'PENDING', 0, ?, ?) + """, + jobRecordId, + PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, + REQUEST_DIGEST, + PAYLOAD, + now, + now + ); + return jobRecordId; + } + + private int tableCount(String tableName) { + Integer count = jdbcTemplate.queryForObject( + "SELECT COUNT(*) FROM " + tableName, + Integer.class + ); + return count == null ? 0 : count; + } + + private String textColumn(UUID jobRecordId, String columnName) { + return jdbcTemplate.queryForObject( + "SELECT " + columnName + " FROM etl_job_records WHERE job_record_id = ?", + String.class, + jobRecordId + ); + } + + /** Minimal transaction-enabled context for cancellation-versus-success atomicity. */ + @Configuration + @EnableTransactionManagement + static class TestConfiguration { + + @Bean + DataSource dataSource() { + return new EmbeddedDatabaseBuilder() + .generateUniqueName(true) + .setType(EmbeddedDatabaseType.H2) + .build(); + } + + @Bean + JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + PlatformTransactionManager transactionManager(DataSource dataSource) { + return new DataSourceTransactionManager(dataSource); + } + + @Bean + EtlJobLeaseRepository etlJobLeaseRepository( + JdbcTemplate jdbcTemplate, + PlatformTransactionManager transactionManager + ) { + return new EtlJobLeaseRepository(jdbcTemplate, transactionManager); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationClaimIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationClaimIntegrationTest.java new file mode 100644 index 00000000..f949ec23 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationClaimIntegrationTest.java @@ -0,0 +1,165 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestLock; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; + +import javax.sql.DataSource; +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Proves that a pending cancellation removes the job from the database-owned worker queue. + */ +@SpringJUnitConfig(EtlJobCancellationClaimIntegrationTest.TestConfiguration.class) +class EtlJobCancellationClaimIntegrationTest { + + private static final String SUBMISSION_KEY = "550e8400-e29b-41d4-a716-446655440000"; + private static final String CANCELLATION_KEY = "70dc8b50-e8b2-4e1a-8c5f-d84814708a77"; + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + + private final EtlJobService jobService; + private final EtlJobLeaseRepository leaseRepository; + private final JdbcTemplate jdbcTemplate; + + @Autowired + EtlJobCancellationClaimIntegrationTest( + EtlJobService jobService, + EtlJobLeaseRepository leaseRepository, + JdbcTemplate jdbcTemplate + ) { + this.jobService = jobService; + this.leaseRepository = leaseRepository; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createJobTable() { + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_job_records"); + jdbcTemplate.execute(""" + CREATE TABLE etl_job_records ( + job_record_id UUID PRIMARY KEY, + principal_scope_hash CHAR(64) NOT NULL, + submission_key_hash CHAR(64) NOT NULL, + request_digest CHAR(64) NOT NULL, + request_payload CLOB, + job_status VARCHAR(32) NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT etl_job_submission_scope_unique + UNIQUE (principal_scope_hash, submission_key_hash) + ) + """); + } + + @Test + void cancelledPendingJobCannotBeClaimedByAnyWorker() { + EtlJobSubmission submission = jobService.submit( + PAYLOAD, + SUBMISSION_KEY, + "tenant_alpha" + ); + + EtlJobCancellation cancellation = jobService.cancelOwned( + submission.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ); + + assertEquals(EtlJobStatus.CANCELLED, cancellation.snapshot().jobStatus()); + assertTrue(leaseRepository.claimNext( + "worker-alpha", + Duration.ofMinutes(5), + 3 + ).isEmpty()); + assertEquals(0, jdbcTemplate.queryForObject( + "SELECT attempt_count FROM etl_job_records WHERE job_record_id = ?", + Integer.class, + submission.jobRecordId() + )); + } + + /** Minimal transaction-enabled context for cancellation and worker claim integration. */ + @Configuration + @EnableTransactionManagement + static class TestConfiguration { + + @Bean + DataSource dataSource() { + return new EmbeddedDatabaseBuilder() + .generateUniqueName(true) + .setType(EmbeddedDatabaseType.H2) + .build(); + } + + @Bean + JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + PlatformTransactionManager transactionManager(DataSource dataSource) { + return new DataSourceTransactionManager(dataSource); + } + + @Bean + ObjectMapper objectMapper() { + return new ObjectMapper(); + } + + @Bean + EtlBatchProperties etlBatchProperties() { + return new EtlBatchProperties(); + } + + @Bean + EtlRequestLock etlRequestLock() { + return lockHash -> true; + } + + @Bean + EtlJobService etlJobService( + JdbcTemplate jdbcTemplate, + ObjectMapper objectMapper, + EtlBatchProperties batchProperties, + EtlRequestLock requestLock + ) { + return new EtlJobService( + jdbcTemplate, + objectMapper, + batchProperties, + requestLock + ); + } + + @Bean + EtlJobLeaseRepository etlJobLeaseRepository( + JdbcTemplate jdbcTemplate, + PlatformTransactionManager transactionManager + ) { + return new EtlJobLeaseRepository(jdbcTemplate, transactionManager); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConcurrencyIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConcurrencyIntegrationTest.java new file mode 100644 index 00000000..86090241 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConcurrencyIntegrationTest.java @@ -0,0 +1,257 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import com.xtrmetl.etl.service.EtlRequestLock; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; + +import javax.sql.DataSource; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Proves concurrent owner cancellation requests converge on one authoritative terminal transition. + */ +@SpringJUnitConfig(EtlJobCancellationConcurrencyIntegrationTest.TestConfiguration.class) +class EtlJobCancellationConcurrencyIntegrationTest { + + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + private static final String SUBMISSION_KEY = "550e8400-e29b-41d4-a716-446655440000"; + private static final String CANCELLATION_KEY = "70dc8b50-e8b2-4e1a-8c5f-d84814708a77"; + private static final String OTHER_CANCELLATION_KEY = + "a52b165f-9d45-4399-ae84-1e93e8fe1e68"; + + private final EtlJobService jobService; + private final JdbcTemplate jdbcTemplate; + private final ExecutorService executor = Executors.newFixedThreadPool(2); + + @Autowired + EtlJobCancellationConcurrencyIntegrationTest( + EtlJobService jobService, + JdbcTemplate jdbcTemplate + ) { + this.jobService = jobService; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createJobTable() { + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_job_records"); + jdbcTemplate.execute(""" + CREATE TABLE etl_job_records ( + job_record_id UUID PRIMARY KEY, + principal_scope_hash CHAR(64) NOT NULL, + submission_key_hash CHAR(64) NOT NULL, + request_digest CHAR(64) NOT NULL, + request_payload CLOB, + job_status VARCHAR(32) NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT etl_job_submission_scope_unique + UNIQUE (principal_scope_hash, submission_key_hash) + ) + """); + } + + @AfterEach + void stopExecutor() throws InterruptedException { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + + @Test + void concurrentIdenticalRequestsProduceOneTransitionAndOneReplay() throws Exception { + UUID jobRecordId = submitPendingJob(); + List> futures = startTogether( + jobRecordId, + List.of(CANCELLATION_KEY, CANCELLATION_KEY) + ); + + List results = List.of( + futures.get(0).get(10, TimeUnit.SECONDS), + futures.get(1).get(10, TimeUnit.SECONDS) + ); + + assertEquals(1, results.stream().filter(result -> !result.replayed()).count()); + assertEquals(1, results.stream().filter(EtlJobCancellation::replayed).count()); + assertTrue(results.stream().allMatch( + result -> result.snapshot().jobStatus() == EtlJobStatus.CANCELLED + )); + assertEquals(1, cancelledRowCount(jobRecordId)); + } + + @Test + void concurrentDifferentKeysProduceOneTransitionAndOneStableConflict() throws Exception { + UUID jobRecordId = submitPendingJob(); + List> futures = startTogether( + jobRecordId, + List.of(CANCELLATION_KEY, OTHER_CANCELLATION_KEY) + ); + + int successes = 0; + int keyConflicts = 0; + for (Future future : futures) { + try { + EtlJobCancellation result = future.get(10, TimeUnit.SECONDS); + assertFalse(result.replayed()); + successes++; + } catch (ExecutionException exception) { + Throwable cause = exception.getCause(); + EtlRequestException requestException = assertThrows( + EtlRequestException.class, + () -> { + throw cause; + } + ); + assertEquals( + EtlRequestError.JOB_CANCELLATION_KEY_REUSED, + requestException.error() + ); + keyConflicts++; + } + } + + assertEquals(1, successes); + assertEquals(1, keyConflicts); + assertEquals(1, cancelledRowCount(jobRecordId)); + } + + private UUID submitPendingJob() { + return jobService.submit(PAYLOAD, SUBMISSION_KEY, "tenant_alpha").jobRecordId(); + } + + private List> startTogether( + UUID jobRecordId, + List cancellationKeys + ) throws InterruptedException { + CountDownLatch ready = new CountDownLatch(cancellationKeys.size()); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (String cancellationKey : cancellationKeys) { + futures.add(executor.submit(() -> { + ready.countDown(); + if (!start.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("concurrent cancellation start timed out"); + } + return jobService.cancelOwned( + jobRecordId, + cancellationKey, + "tenant_alpha" + ); + })); + } + assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + return List.copyOf(futures); + } + + private int cancelledRowCount(UUID jobRecordId) { + Integer count = jdbcTemplate.queryForObject( + """ + SELECT COUNT(*) + FROM etl_job_records + WHERE job_record_id = ? + AND job_status = 'CANCELLED' + AND request_payload IS NULL + AND lease_claim_id IS NULL + AND lease_owner_id IS NULL + AND lease_expires_at IS NULL + AND cancellation_key_hash IS NOT NULL + AND cancellation_code = ? + AND job_cancelled_at IS NOT NULL + """, + Integer.class, + jobRecordId, + EtlJobService.CANCELLED_BY_OWNER_CODE + ); + return count == null ? 0 : count; + } + + /** Minimal transaction-enabled context for concurrent cancellation integration. */ + @Configuration + @EnableTransactionManagement + static class TestConfiguration { + + @Bean + DataSource dataSource() { + return new EmbeddedDatabaseBuilder() + .generateUniqueName(true) + .setType(EmbeddedDatabaseType.H2) + .build(); + } + + @Bean + JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + PlatformTransactionManager transactionManager(DataSource dataSource) { + return new DataSourceTransactionManager(dataSource); + } + + @Bean + ObjectMapper objectMapper() { + return new ObjectMapper(); + } + + @Bean + EtlBatchProperties etlBatchProperties() { + return new EtlBatchProperties(); + } + + @Bean + EtlRequestLock etlRequestLock() { + return lockHash -> true; + } + + @Bean + EtlJobService etlJobService( + JdbcTemplate jdbcTemplate, + ObjectMapper objectMapper, + EtlBatchProperties batchProperties, + EtlRequestLock requestLock + ) { + return new EtlJobService( + jdbcTemplate, + objectMapper, + batchProperties, + requestLock + ); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConditionalStatusTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConditionalStatusTest.java new file mode 100644 index 00000000..135e6a22 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConditionalStatusTest.java @@ -0,0 +1,86 @@ +package com.xtrmetl.etl.job; + +import com.xtrmetl.etl.controller.EtlApiProblemHandler; +import com.xtrmetl.etl.controller.EtlJobController; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpHeaders; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +import java.security.Principal; +import java.time.Instant; +import java.util.UUID; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.not; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +/** + * Proves that a committed cancellation invalidates every active status representation validator. + */ +class EtlJobCancellationConditionalStatusTest { + + private static final UUID JOB_RECORD_ID = UUID.fromString( + "cf4f083f-8c90-4f34-a8b6-b53761de44ef" + ); + private static final Principal PRINCIPAL = () -> "tenant_alpha"; + private static final Instant CREATED_AT = Instant.parse("2026-08-05T01:00:00Z"); + + private EtlJobService etlJobService; + private MockMvc mockMvc; + + @BeforeEach + void setUp() { + etlJobService = mock(EtlJobService.class); + mockMvc = MockMvcBuilders + .standaloneSetup(new EtlJobController(etlJobService)) + .setControllerAdvice(new EtlApiProblemHandler()) + .build(); + } + + @Test + void cancelledStatusInvalidatesThePriorActiveEntityTag() throws Exception { + when(etlJobService.findOwned(JOB_RECORD_ID, "tenant_alpha")) + .thenReturn( + snapshot(EtlJobStatus.RUNNING, Instant.parse("2026-08-05T01:00:05Z")), + snapshot(EtlJobStatus.CANCELLED, Instant.parse("2026-08-05T01:00:06Z")) + ); + + String activeEntityTag = mockMvc.perform(statusRequest()) + .andExpect(status().isOk()) + .andReturn() + .getResponse() + .getHeader(HttpHeaders.ETAG); + assertNotNull(activeEntityTag); + + mockMvc.perform(statusRequest().header(HttpHeaders.IF_NONE_MATCH, activeEntityTag)) + .andExpect(status().isOk()) + .andExpect(header().string(HttpHeaders.ETAG, not(equalTo(activeEntityTag)))) + .andExpect(header().string(HttpHeaders.CACHE_CONTROL, "no-store")) + .andExpect(jsonPath("$.jobStatus").value("CANCELLED")) + .andExpect(jsonPath("$.failureCode").doesNotExist()); + } + + private static org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder + statusRequest() { + return get("/api/etl/jobs/" + JOB_RECORD_ID).principal(PRINCIPAL); + } + + private static EtlJobSnapshot snapshot(EtlJobStatus status, Instant updatedAt) { + return new EtlJobSnapshot( + JOB_RECORD_ID, + status, + 1, + null, + CREATED_AT, + updatedAt + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationKeyDomainIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationKeyDomainIntegrationTest.java new file mode 100644 index 00000000..f7ca9ef7 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationKeyDomainIntegrationTest.java @@ -0,0 +1,169 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestLock; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; + +import javax.sql.DataSource; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; + +/** + * Proves cancellation replay identities cannot correlate one raw key across jobs or tenants. + */ +@SpringJUnitConfig(EtlJobCancellationKeyDomainIntegrationTest.TestConfiguration.class) +class EtlJobCancellationKeyDomainIntegrationTest { + + private static final String CANCELLATION_KEY = "70dc8b50-e8b2-4e1a-8c5f-d84814708a77"; + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + + private final EtlJobService jobService; + private final JdbcTemplate jdbcTemplate; + + @Autowired + EtlJobCancellationKeyDomainIntegrationTest( + EtlJobService jobService, + JdbcTemplate jdbcTemplate + ) { + this.jobService = jobService; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createJobTable() { + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_job_records"); + jdbcTemplate.execute(""" + CREATE TABLE etl_job_records ( + job_record_id UUID PRIMARY KEY, + principal_scope_hash CHAR(64) NOT NULL, + submission_key_hash CHAR(64) NOT NULL, + request_digest CHAR(64) NOT NULL, + request_payload CLOB, + job_status VARCHAR(32) NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT etl_job_submission_scope_unique + UNIQUE (principal_scope_hash, submission_key_hash) + ) + """); + } + + @Test + void identicalRawKeysProduceDistinctStoredHashesAcrossJobsAndTenants() { + EtlJobSubmission alphaFirst = submit( + "550e8400-e29b-41d4-a716-446655440000", + "tenant_alpha" + ); + EtlJobSubmission alphaSecond = submit( + "b176c1b8-3294-4b78-9939-17e3a09aa0e7", + "tenant_alpha" + ); + EtlJobSubmission betaFirst = submit( + "550e8400-e29b-41d4-a716-446655440000", + "tenant_beta" + ); + + jobService.cancelOwned(alphaFirst.jobRecordId(), CANCELLATION_KEY, "tenant_alpha"); + jobService.cancelOwned(alphaSecond.jobRecordId(), CANCELLATION_KEY, "tenant_alpha"); + jobService.cancelOwned(betaFirst.jobRecordId(), CANCELLATION_KEY, "tenant_beta"); + + String alphaFirstHash = cancellationHash(alphaFirst.jobRecordId()); + String alphaSecondHash = cancellationHash(alphaSecond.jobRecordId()); + String betaFirstHash = cancellationHash(betaFirst.jobRecordId()); + + assertEquals(64, alphaFirstHash.length()); + assertEquals(64, alphaSecondHash.length()); + assertEquals(64, betaFirstHash.length()); + assertNotEquals(alphaFirstHash, alphaSecondHash); + assertNotEquals(alphaFirstHash, betaFirstHash); + assertNotEquals(alphaSecondHash, betaFirstHash); + } + + private EtlJobSubmission submit(String submissionKey, String principalScope) { + return jobService.submit(PAYLOAD, submissionKey, principalScope); + } + + private String cancellationHash(UUID jobRecordId) { + return jdbcTemplate.queryForObject( + "SELECT cancellation_key_hash FROM etl_job_records WHERE job_record_id = ?", + String.class, + jobRecordId + ); + } + + /** Minimal transaction-enabled context for cancellation replay-identity privacy tests. */ + @Configuration + @EnableTransactionManagement + static class TestConfiguration { + + @Bean + DataSource dataSource() { + return new EmbeddedDatabaseBuilder() + .generateUniqueName(true) + .setType(EmbeddedDatabaseType.H2) + .build(); + } + + @Bean + JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + PlatformTransactionManager transactionManager(DataSource dataSource) { + return new DataSourceTransactionManager(dataSource); + } + + @Bean + ObjectMapper objectMapper() { + return new ObjectMapper(); + } + + @Bean + EtlBatchProperties etlBatchProperties() { + return new EtlBatchProperties(); + } + + @Bean + EtlRequestLock etlRequestLock() { + return lockHash -> true; + } + + @Bean + EtlJobService etlJobService( + JdbcTemplate jdbcTemplate, + ObjectMapper objectMapper, + EtlBatchProperties batchProperties, + EtlRequestLock requestLock + ) { + return new EtlJobService( + jdbcTemplate, + objectMapper, + batchProperties, + requestLock + ); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationLeaseFenceIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationLeaseFenceIntegrationTest.java new file mode 100644 index 00000000..75610b0b --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationLeaseFenceIntegrationTest.java @@ -0,0 +1,174 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.annotation.Transactional; + +import javax.sql.DataSource; +import java.time.Duration; +import java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Proves that a committed owner cancellation makes every previously issued lease stale. + */ +@SpringJUnitConfig(EtlJobCancellationLeaseFenceIntegrationTest.TestConfiguration.class) +class EtlJobCancellationLeaseFenceIntegrationTest { + + private static final String PRINCIPAL_SCOPE_HASH = "a".repeat(64); + private static final String SUBMISSION_KEY_HASH = "b".repeat(64); + private static final String REQUEST_DIGEST = "c".repeat(64); + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + + private final EtlJobLeaseRepository leaseRepository; + private final JdbcTemplate jdbcTemplate; + + @Autowired + EtlJobCancellationLeaseFenceIntegrationTest( + EtlJobLeaseRepository leaseRepository, + JdbcTemplate jdbcTemplate + ) { + this.leaseRepository = leaseRepository; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createJobTable() { + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_job_records"); + jdbcTemplate.execute(""" + CREATE TABLE etl_job_records ( + job_record_id UUID PRIMARY KEY, + principal_scope_hash CHAR(64) NOT NULL, + submission_key_hash CHAR(64) NOT NULL, + request_digest CHAR(64) NOT NULL, + request_payload CLOB, + job_status VARCHAR(32) NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE NOT NULL, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL + ) + """); + } + + @Test + @Transactional + void cancellationWinsBeforeSuccessAndThePriorLeaseCannotOverwriteIt() { + UUID jobRecordId = UUID.fromString("66b818a1-fd36-4ea9-aac0-a4ad8ca05fc1"); + Instant createdAt = Instant.parse("2026-08-06T00:00:00Z"); + jdbcTemplate.update( + """ + INSERT INTO etl_job_records ( + job_record_id, principal_scope_hash, submission_key_hash, + request_digest, request_payload, job_status, attempt_count, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, 'PENDING', 0, ?, ?) + """, + jobRecordId, + PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, + REQUEST_DIGEST, + PAYLOAD, + createdAt, + createdAt + ); + EtlJobLease lease = leaseRepository.claimNext( + "worker-alpha", + Duration.ofMinutes(5), + 3 + ).orElseThrow(); + + int cancelledRows = jdbcTemplate.update( + """ + UPDATE etl_job_records + SET job_status = 'CANCELLED', + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + cancellation_key_hash = ?, + cancellation_code = ?, + job_cancelled_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE job_record_id = ? + AND principal_scope_hash = ? + AND job_status IN ('PENDING', 'RUNNING') + """, + "d".repeat(64), + EtlJobService.CANCELLED_BY_OWNER_CODE, + jobRecordId, + PRINCIPAL_SCOPE_HASH + ); + + assertEquals(1, cancelledRows); + assertThrows(StaleEtlJobLeaseException.class, () -> leaseRepository.markSucceeded(lease)); + assertEquals("CANCELLED", textColumn(jobRecordId, "job_status")); + assertNull(textColumn(jobRecordId, "request_payload")); + assertNull(textColumn(jobRecordId, "lease_owner_id")); + assertEquals( + EtlJobService.CANCELLED_BY_OWNER_CODE, + textColumn(jobRecordId, "cancellation_code") + ); + } + + private String textColumn(UUID jobRecordId, String columnName) { + return jdbcTemplate.queryForObject( + "SELECT " + columnName + " FROM etl_job_records WHERE job_record_id = ?", + String.class, + jobRecordId + ); + } + + /** Minimal transaction-enabled SQL context for cancellation lease-fencing tests. */ + @Configuration + @EnableTransactionManagement + static class TestConfiguration { + + @Bean + DataSource dataSource() { + return new EmbeddedDatabaseBuilder() + .generateUniqueName(true) + .setType(EmbeddedDatabaseType.H2) + .build(); + } + + @Bean + JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + PlatformTransactionManager transactionManager(DataSource dataSource) { + return new DataSourceTransactionManager(dataSource); + } + + @Bean + EtlJobLeaseRepository etlJobLeaseRepository( + JdbcTemplate jdbcTemplate, + PlatformTransactionManager transactionManager + ) { + return new EtlJobLeaseRepository(jdbcTemplate, transactionManager); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationServiceBoundaryTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationServiceBoundaryTest.java new file mode 100644 index 00000000..e144e0c8 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationServiceBoundaryTest.java @@ -0,0 +1,151 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** + * Covers fail-closed cancellation validation, transaction, and race classification boundaries. + */ +class EtlJobCancellationServiceBoundaryTest { + + private static final UUID JOB_RECORD_ID = UUID.fromString( + "2f4e2926-03dd-461f-873a-a70ad2256680" + ); + private static final String CANCELLATION_KEY = + "5d09cd43-50d1-4b82-a94f-a43f5bc6e56b"; + + @AfterEach + void clearSyntheticTransactionState() { + TransactionSynchronizationManager.clear(); + } + + @Test + void refusesCancellationWithoutAnActualTransactionBeforeDatabaseWork() { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + EtlJobService service = service(jdbcTemplate); + + IllegalStateException exception = assertThrows( + IllegalStateException.class, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, "tenant_alpha") + ); + + assertEquals( + "Durable ETL job cancellation requires an active transaction", + exception.getMessage() + ); + verifyNoInteractions(jdbcTemplate); + } + + @Test + void rejectsInvalidCancellationIdentityBeforeDatabaseWork() { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + EtlJobService service = service(jdbcTemplate); + + assertThrows( + NullPointerException.class, + () -> service.cancelOwned(null, CANCELLATION_KEY, "tenant_alpha") + ); + assertError( + EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED, + () -> service.cancelOwned(JOB_RECORD_ID, null, "tenant_alpha") + ); + assertError( + EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED, + () -> service.cancelOwned(JOB_RECORD_ID, "unsafe key", "tenant_alpha") + ); + assertError( + EtlRequestError.IDEMPOTENCY_PRINCIPAL_REQUIRED, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, null) + ); + assertError( + EtlRequestError.IDEMPOTENCY_PRINCIPAL_REQUIRED, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, " ") + ); + verifyNoInteractions(jdbcTemplate); + } + + @Test + void reportsAnActiveRowThatDidNotTransitionAsCancellationInProgress() { + TransactionSynchronizationManager.setActualTransactionActive(true); + EtlJobService service = service(new UnchangedPendingJobJdbcTemplate(JOB_RECORD_ID)); + + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, "tenant_alpha") + ); + + assertEquals(EtlRequestError.JOB_CANCELLATION_IN_PROGRESS, exception.error()); + } + + private static EtlJobService service(JdbcTemplate jdbcTemplate) { + return new EtlJobService( + jdbcTemplate, + new ObjectMapper(), + new EtlBatchProperties(), + idempotencyKeyHash -> true + ); + } + + private static void assertError(EtlRequestError expected, Runnable invocation) { + EtlRequestException exception = assertThrows(EtlRequestException.class, invocation::run); + assertEquals(expected, exception.error()); + } + + /** + * Deterministic JDBC double for the concurrency branch where another writer retains PENDING. + */ + private static final class UnchangedPendingJobJdbcTemplate extends JdbcTemplate { + + private final UUID jobRecordId; + + private UnchangedPendingJobJdbcTemplate(UUID jobRecordId) { + this.jobRecordId = jobRecordId; + } + + @Override + public int update(String sql, Object... args) { + return 0; + } + + @Override + public List query(String sql, RowMapper rowMapper, Object... args) { + ResultSet resultSet = mock(ResultSet.class); + try { + when(resultSet.getObject("job_record_id", UUID.class)).thenReturn(jobRecordId); + when(resultSet.getString("job_status")).thenReturn("PENDING"); + when(resultSet.getInt("attempt_count")).thenReturn(0); + when(resultSet.getString("failure_code")).thenReturn(null); + when(resultSet.getString("cancellation_key_hash")).thenReturn(null); + when(resultSet.getTimestamp("created_at")).thenReturn( + Timestamp.from(Instant.parse("2026-08-06T00:00:00Z")) + ); + when(resultSet.getTimestamp("updated_at")).thenReturn( + Timestamp.from(Instant.parse("2026-08-06T00:00:01Z")) + ); + return List.of(rowMapper.mapRow(resultSet, 0)); + } catch (SQLException exception) { + throw new AssertionError("row mapper unexpectedly rejected the test row", exception); + } + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationTest.java new file mode 100644 index 00000000..7d23140d --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationTest.java @@ -0,0 +1,45 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Covers immutable cancellation-result validation and replay semantics. + */ +class EtlJobCancellationTest { + + @Test + void acceptsOnlyCancelledOperatorSafeSnapshots() { + EtlJobSnapshot cancelledSnapshot = snapshot(EtlJobStatus.CANCELLED); + + EtlJobCancellation first = new EtlJobCancellation(cancelledSnapshot, false); + EtlJobCancellation replay = new EtlJobCancellation(cancelledSnapshot, true); + + assertEquals(cancelledSnapshot, first.snapshot()); + assertFalse(first.replayed()); + assertTrue(replay.replayed()); + assertThrows(NullPointerException.class, () -> new EtlJobCancellation(null, false)); + assertThrows( + IllegalArgumentException.class, + () -> new EtlJobCancellation(snapshot(EtlJobStatus.PENDING), false) + ); + } + + private static EtlJobSnapshot snapshot(EtlJobStatus status) { + return new EtlJobSnapshot( + UUID.fromString("75f61ec2-4f96-49e4-bf8e-66e2b75fb175"), + status, + 1, + null, + Instant.parse("2026-08-06T00:00:00Z"), + Instant.parse("2026-08-06T00:00:01Z") + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java index 0418ebaa..30a2f420 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java @@ -26,13 +26,15 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; /** - * Covers typed data-access, malformed resource identifiers, and unexpected failures at both - * durable job controller boundaries. + * Covers typed data-access, malformed resource identifiers, and unexpected failures at durable job + * submission, status, and cancellation boundaries. */ class EtlJobControllerFailureTest { private static final String JOBS_PATH = "/api/etl/jobs"; private static final String IDEMPOTENCY_KEY = "\"550e8400-e29b-41d4-a716-446655440000\""; + private static final String CANCELLATION_KEY = + "\"70dc8b50-e8b2-4e1a-8c5f-d84814708a77\""; private static final Principal PRINCIPAL = () -> "tenant_alpha"; private EtlJobService etlJobService; @@ -127,6 +129,61 @@ void mapsStatusUnexpectedFailuresWithoutLeakingMessages() throws Exception { .andExpect(jsonPath("$.errorCode").value("etl_internal_error")); } + @Test + void preservesTypedCancellationConflicts() throws Exception { + when(etlJobService.cancelOwned(any(UUID.class), anyString(), anyString())) + .thenThrow(new EtlRequestException(EtlRequestError.JOB_ALREADY_SUCCEEDED)); + + performCancellation() + .andExpect(status().isConflict()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_job_already_succeeded")) + .andExpect(jsonPath("$.detail").value( + "The durable job succeeded before cancellation could commit." + )); + } + + @Test + void treatsMalformedCancellationIdentifiersAsOwnerSafeNotFound() throws Exception { + mockMvc.perform(post(JOBS_PATH + "/not-a-uuid/cancellation") + .principal(PRINCIPAL) + .header("Idempotency-Key", CANCELLATION_KEY)) + .andExpect(status().isNotFound()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_job_not_found")); + + verifyNoInteractions(etlJobService); + } + + @Test + void mapsCancellationDatabaseFailuresWithoutLeakingMessages() throws Exception { + DataAccessException databaseFailure = new DataAccessException("secret database detail") { }; + when(etlJobService.cancelOwned(any(UUID.class), anyString(), anyString())) + .thenThrow(databaseFailure); + + performCancellation() + .andExpect(status().isInternalServerError()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_target_failure")) + .andExpect(jsonPath("$.detail").value( + "The ETL target could not process the request." + )); + } + + @Test + void mapsCancellationUnexpectedFailuresWithoutLeakingMessages() throws Exception { + when(etlJobService.cancelOwned(any(UUID.class), anyString(), anyString())) + .thenThrow(new IllegalStateException("secret cancellation detail")); + + performCancellation() + .andExpect(status().isInternalServerError()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_internal_error")) + .andExpect(jsonPath("$.detail").value( + "The ETL request could not be processed." + )); + } + private org.springframework.test.web.servlet.ResultActions performSubmission() throws Exception { return mockMvc.perform(post(JOBS_PATH) .principal(PRINCIPAL) @@ -138,4 +195,10 @@ private org.springframework.test.web.servlet.ResultActions performSubmission() t private org.springframework.test.web.servlet.ResultActions performStatus() throws Exception { return mockMvc.perform(get(JOBS_PATH + "/" + UUID.randomUUID()).principal(PRINCIPAL)); } + + private org.springframework.test.web.servlet.ResultActions performCancellation() throws Exception { + return mockMvc.perform(post(JOBS_PATH + "/" + UUID.randomUUID() + "/cancellation") + .principal(PRINCIPAL) + .header("Idempotency-Key", CANCELLATION_KEY)); + } } diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java index 940d0626..469e105b 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java @@ -4,6 +4,7 @@ import com.xtrmetl.etl.controller.EtlJobController; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.test.web.servlet.MockMvc; import org.springframework.test.web.servlet.setup.MockMvcBuilders; @@ -30,6 +31,8 @@ class EtlJobControllerTest { private static final String JOBS_PATH = "/api/etl/jobs"; private static final String IDEMPOTENCY_KEY = "\"550e8400-e29b-41d4-a716-446655440000\""; + private static final String CANCELLATION_KEY = + "\"70dc8b50-e8b2-4e1a-8c5f-d84814708a77\""; private static final Principal PRINCIPAL = () -> "tenant_alpha"; private EtlJobService etlJobService; @@ -154,4 +157,82 @@ void requiresAuthenticationForJobStatus() throws Exception { verifyNoInteractions(etlJobService); } + + @Test + void cancelsAnOwnedJobAndReturnsTheTerminalStatus() throws Exception { + UUID jobRecordId = UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef"); + EtlJobSnapshot snapshot = new EtlJobSnapshot( + jobRecordId, + EtlJobStatus.CANCELLED, + 1, + null, + Instant.parse("2026-08-04T10:00:00Z"), + Instant.parse("2026-08-06T03:00:00Z") + ); + when(etlJobService.cancelOwned(jobRecordId, CANCELLATION_KEY, "tenant_alpha")) + .thenReturn(new EtlJobCancellation(snapshot, false)); + + mockMvc.perform(post(cancellationPath(jobRecordId)) + .principal(PRINCIPAL) + .header("Idempotency-Key", CANCELLATION_KEY)) + .andExpect(status().isOk()) + .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON)) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(header().string("Idempotency-Replayed", "false")) + .andExpect(header().exists(HttpHeaders.ETAG)) + .andExpect(jsonPath("$.jobRecordId").value(jobRecordId.toString())) + .andExpect(jsonPath("$.jobStatus").value("CANCELLED")) + .andExpect(jsonPath("$.attemptCount").value(1)) + .andExpect(jsonPath("$.failureCode").doesNotExist()) + .andExpect(jsonPath("$.requestPayload").doesNotExist()) + .andExpect(jsonPath("$.cancellationKeyHash").doesNotExist()) + .andExpect(jsonPath("$.cancellationCode").doesNotExist()); + + verify(etlJobService).cancelOwned(jobRecordId, CANCELLATION_KEY, "tenant_alpha"); + } + + @Test + void marksAnIdenticalCancellationAsReplayed() throws Exception { + UUID jobRecordId = UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef"); + EtlJobSnapshot snapshot = new EtlJobSnapshot( + jobRecordId, + EtlJobStatus.CANCELLED, + 0, + null, + Instant.parse("2026-08-04T10:00:00Z"), + Instant.parse("2026-08-06T03:00:00Z") + ); + when(etlJobService.cancelOwned(jobRecordId, CANCELLATION_KEY, "tenant_alpha")) + .thenReturn(new EtlJobCancellation(snapshot, true)); + + mockMvc.perform(post(cancellationPath(jobRecordId)) + .principal(PRINCIPAL) + .header("Idempotency-Key", CANCELLATION_KEY)) + .andExpect(status().isOk()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(header().string("Idempotency-Replayed", "true")) + .andExpect(jsonPath("$.jobStatus").value("CANCELLED")); + } + + @Test + void requiresAuthenticationAndAKeyBeforeCancellationServiceAccess() throws Exception { + UUID jobRecordId = UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef"); + + mockMvc.perform(post(cancellationPath(jobRecordId)) + .header("Idempotency-Key", CANCELLATION_KEY)) + .andExpect(status().isUnauthorized()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_idempotency_principal_required")); + + mockMvc.perform(post(cancellationPath(jobRecordId)).principal(PRINCIPAL)) + .andExpect(status().isBadRequest()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_job_cancellation_key_required")); + + verifyNoInteractions(etlJobService); + } + + private static String cancellationPath(UUID jobRecordId) { + return JOBS_PATH + "/" + jobRecordId + "/cancellation"; + } } diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java index 0f94c7e5..6fec9ca7 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java @@ -50,6 +50,36 @@ void migrationReservesStableWorkerStatesAndRequiresTerminalPayloadClearing() thr assertTrue(migration.contains("request_payload IS NULL")); } + @Test + void cancellationMigrationAddsOneTerminalOwnerSafeLifecycle() throws IOException { + String migration = read( + "etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql" + ).replaceAll("\\s+", " "); + + assertTrue(migration.contains("ADD COLUMN cancellation_key_hash CHAR(64)")); + assertTrue(migration.contains("ADD COLUMN cancellation_code VARCHAR(128)")); + assertTrue(migration.contains("ADD COLUMN job_cancelled_at TIMESTAMPTZ")); + assertTrue(migration.contains( + "job_status IN ('PENDING', 'RUNNING', 'SUCCEEDED', 'FAILED', 'CANCELLED')" + )); + assertTrue(migration.contains( + "job_status IN ('SUCCEEDED', 'FAILED', 'CANCELLED') AND request_payload IS NULL" + )); + assertTrue(migration.contains("job_status = 'CANCELLED'")); + assertTrue(migration.contains("cancellation_key_hash IS NOT NULL")); + assertTrue(migration.contains("cancellation_code IS NOT NULL")); + assertTrue(migration.contains("job_cancelled_at IS NOT NULL")); + assertTrue(migration.contains("job_status <> 'CANCELLED'")); + assertTrue(migration.contains("cancellation_key_hash IS NULL")); + assertTrue(migration.contains("lease_claim_id IS NULL")); + assertTrue(migration.contains("CONSTRAINT etl_job_cancellation_key_hash_format")); + assertTrue(migration.contains("CONSTRAINT etl_job_cancellation_code_format")); + assertTrue(migration.contains("CONSTRAINT etl_job_cancellation_lifecycle_check")); + assertFalse(migration.contains("principal_name")); + assertFalse(migration.contains("cancellation_key TEXT")); + assertFalse(migration.contains("cancellation_reason")); + } + @Test void paginationMigrationUsesTheOwnerAndCompleteStableOrderingKey() throws IOException { String migration = read( diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java index ad802106..358fb239 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java @@ -19,21 +19,27 @@ import org.springframework.transaction.annotation.EnableTransactionManagement; import javax.sql.DataSource; +import java.time.OffsetDateTime; import java.util.UUID; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; /** - * Defines the durable, principal-scoped asynchronous ETL job intake contract. + * Defines the durable, principal-scoped asynchronous ETL job intake and cancellation contract. */ @SpringJUnitConfig(EtlJobServiceIntegrationTest.TestConfiguration.class) class EtlJobServiceIntegrationTest { private static final String IDEMPOTENCY_KEY = "550e8400-e29b-41d4-a716-446655440000"; + private static final String CANCELLATION_KEY = "70dc8b50-e8b2-4e1a-8c5f-d84814708a77"; + private static final String SECOND_CANCELLATION_KEY = + "a52b165f-9d45-4399-ae84-1e93e8fe1e68"; private static final String PAYLOAD = "[{\"id\":\"record_alpha\",\"name\":\"accepted\"}]"; private final EtlJobService etlJobService; @@ -54,10 +60,16 @@ CREATE TABLE etl_job_records ( principal_scope_hash CHAR(64) NOT NULL, submission_key_hash CHAR(64) NOT NULL, request_digest CHAR(64) NOT NULL, - request_payload VARCHAR(8192) NOT NULL, + request_payload VARCHAR(8192), job_status VARCHAR(32) NOT NULL, attempt_count INTEGER NOT NULL DEFAULT 0, failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, CONSTRAINT etl_job_submission_scope_unique @@ -162,6 +174,184 @@ void rejectsMalformedOrOversizedPayloadsBeforePersistence() { assertEquals(0, countJobRows()); } + @Test + void cancelsPendingWorkClearsThePayloadAndReplaysTheSameSemanticKey() { + EtlJobSubmission created = etlJobService.submit(PAYLOAD, IDEMPOTENCY_KEY, "tenant_alpha"); + + EtlJobCancellation cancelled = etlJobService.cancelOwned( + created.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ); + EtlJobCancellation replayed = etlJobService.cancelOwned( + created.jobRecordId(), + "\"" + CANCELLATION_KEY + "\"", + "tenant_alpha" + ); + + assertFalse(cancelled.replayed()); + assertTrue(replayed.replayed()); + assertEquals(created.jobRecordId(), cancelled.snapshot().jobRecordId()); + assertEquals(EtlJobStatus.CANCELLED, cancelled.snapshot().jobStatus()); + assertEquals(cancelled.snapshot(), replayed.snapshot()); + assertNull(column(created.jobRecordId(), "request_payload", String.class)); + assertEquals( + EtlJobService.CANCELLED_BY_OWNER_CODE, + column(created.jobRecordId(), "cancellation_code", String.class) + ); + String keyHash = column(created.jobRecordId(), "cancellation_key_hash", String.class); + assertNotNull(keyHash); + assertEquals(64, keyHash.length()); + assertNotEquals(CANCELLATION_KEY, keyHash); + assertNotNull(column( + created.jobRecordId(), + "job_cancelled_at", + OffsetDateTime.class + )); + } + + @Test + void rejectsASecondCancellationIdentityAfterCancellation() { + EtlJobSubmission created = etlJobService.submit(PAYLOAD, IDEMPOTENCY_KEY, "tenant_alpha"); + etlJobService.cancelOwned(created.jobRecordId(), CANCELLATION_KEY, "tenant_alpha"); + + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + created.jobRecordId(), + SECOND_CANCELLATION_KEY, + "tenant_alpha" + ) + ); + + assertEquals(EtlRequestError.JOB_CANCELLATION_KEY_REUSED, exception.error()); + assertEquals(EtlJobStatus.CANCELLED, etlJobService.findOwned( + created.jobRecordId(), + "tenant_alpha" + ).jobStatus()); + } + + @Test + void keepsForeignOwnedAndMissingCancellationTargetsIndistinguishable() { + EtlJobSubmission created = etlJobService.submit(PAYLOAD, IDEMPOTENCY_KEY, "tenant_alpha"); + + EtlRequestException hidden = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + created.jobRecordId(), + CANCELLATION_KEY, + "tenant_beta" + ) + ); + EtlRequestException missing = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + UUID.randomUUID(), + CANCELLATION_KEY, + "tenant_alpha" + ) + ); + + assertEquals(EtlRequestError.JOB_NOT_FOUND, hidden.error()); + assertEquals(EtlRequestError.JOB_NOT_FOUND, missing.error()); + assertEquals(EtlJobStatus.PENDING, etlJobService.findOwned( + created.jobRecordId(), + "tenant_alpha" + ).jobStatus()); + } + + @Test + void rejectsCancellationAfterACommittedSuccessOrFailure() { + EtlJobSubmission succeeded = etlJobService.submit( + PAYLOAD, + IDEMPOTENCY_KEY, + "tenant_alpha" + ); + EtlJobSubmission failed = etlJobService.submit( + PAYLOAD, + "1d38ad67-48d8-446c-bca1-76bfe2ba8eef", + "tenant_alpha" + ); + jdbcTemplate.update( + "UPDATE etl_job_records SET job_status = 'SUCCEEDED', request_payload = NULL " + + "WHERE job_record_id = ?", + succeeded.jobRecordId() + ); + jdbcTemplate.update( + "UPDATE etl_job_records SET job_status = 'FAILED', request_payload = NULL, " + + "failure_code = 'etl_target_failure' WHERE job_record_id = ?", + failed.jobRecordId() + ); + + EtlRequestException successConflict = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + succeeded.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ) + ); + EtlRequestException failureConflict = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + failed.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ) + ); + + assertEquals(EtlRequestError.JOB_ALREADY_SUCCEEDED, successConflict.error()); + assertEquals(EtlRequestError.JOB_ALREADY_FAILED, failureConflict.error()); + } + + @Test + void cancelsRunningWorkAndInvalidatesEveryLeaseField() { + EtlJobSubmission created = etlJobService.submit(PAYLOAD, IDEMPOTENCY_KEY, "tenant_alpha"); + jdbcTemplate.update( + """ + UPDATE etl_job_records + SET job_status = 'RUNNING', + attempt_count = 1, + lease_claim_id = ?, + lease_owner_id = 'worker_alpha', + lease_expires_at = DATEADD('MINUTE', 5, CURRENT_TIMESTAMP) + WHERE job_record_id = ? + """, + UUID.fromString("7c10a65b-5791-4e0f-9fba-dadbb13971da"), + created.jobRecordId() + ); + + EtlJobCancellation cancellation = etlJobService.cancelOwned( + created.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ); + + assertEquals(EtlJobStatus.CANCELLED, cancellation.snapshot().jobStatus()); + assertEquals(1, cancellation.snapshot().attemptCount()); + assertNull(column(created.jobRecordId(), "lease_claim_id", UUID.class)); + assertNull(column(created.jobRecordId(), "lease_owner_id", String.class)); + assertNull(column(created.jobRecordId(), "lease_expires_at", OffsetDateTime.class)); + assertNull(column(created.jobRecordId(), "request_payload", String.class)); + } + + @Test + void rejectsInvalidCancellationKeysBeforeAnyTableAccess() { + jdbcTemplate.execute("DROP TABLE etl_job_records"); + + EtlRequestException absent = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned(UUID.randomUUID(), null, "tenant_alpha") + ); + EtlRequestException malformed = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned(UUID.randomUUID(), "too-short", "tenant_alpha") + ); + + assertEquals(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED, absent.error()); + assertEquals(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED, malformed.error()); + } + private int countJobRows() { Integer count = jdbcTemplate.queryForObject( "SELECT COUNT(*) FROM etl_job_records", @@ -170,8 +360,16 @@ private int countJobRows() { return count == null ? 0 : count; } + private T column(UUID jobRecordId, String columnName, Class valueType) { + return jdbcTemplate.queryForObject( + "SELECT " + columnName + " FROM etl_job_records WHERE job_record_id = ?", + (resultSet, rowNumber) -> resultSet.getObject(columnName, valueType), + jobRecordId + ); + } + /** - * Minimal transaction-enabled test context for the job intake service. + * Minimal transaction-enabled test context for durable job resource services. */ @Configuration @EnableTransactionManagement