diff --git a/CHANGELOG.md b/CHANGELOG.md index 5c8ea782..6f0b446e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Updated existing pull-request candidates now carry their captured pre-agent head into the deterministic publisher, which rejects destructive ancestry, more than 50 agent-introduced files, and any agent-introduced `.github/**` or `CODEOWNERS` change before exposing the updated pull request or authorizing checks. - The hourly OpenCode workflow now uses the current free NVIDIA `deepseek-ai/deepseek-v4-pro` endpoint for long-context coding and agentic tool use instead of the deprecated Qwen3 Coder free endpoint; model or endpoint rejection fails visibly without a non-NVIDIA, partner-only, or automatic fallback. - The managed Jackson component set now uses the patched 2.21.5 BOM, closing CVE-2026-54515, CVE-2026-59889, and GHSA-mhm7-754m-9p8w while keeping core, annotations, datatype, and module artifacts aligned. -- Durable `POST /api/etl/jobs` submissions now return RFC 9110 `202 Accepted`, a stable pending-job representation, `Location` status-monitor metadata, and explicit replay metadata without changing the synchronous `/api/etl/process` contract. The incomplete intake controller is fail-closed and requires explicit `xtrmetl.etl.jobs.intake-enabled=true` operator opt-in until worker execution and terminal payload clearing are implemented. +- Durable `POST /api/etl/jobs` submissions now return RFC 9110 `202 Accepted`, a stable pending-job representation, `Location` status-monitor metadata, and explicit replay metadata without changing the synchronous `/api/etl/process` contract. Durable intake remains independently opt-in through the preferred `mightyetl.etl.jobs.intake-enabled=true` property or supported legacy alias, while worker activation is a separate disabled-by-default control. +- Durable ETL jobs now execute through an opt-in lease-fenced worker: PostgreSQL `FOR UPDATE SKIP LOCKED` claims one eligible job per poll across replicas, exact lease-owner/token fencing prevents stale transitions, target writes plus terminal success commit transactionally, terminal states clear retained payloads, retries are bounded with stable non-sensitive failure codes, and fixed-cardinality worker telemetry records `idle`, `succeeded`, `retried`, `failed`, and `stale` outcomes. - Concurrent requests using the same authenticated-principal-scoped semantic idempotency key now return immediate RFC 9457 `409 etl_idempotency_request_in_progress` responses through PostgreSQL `pg_try_advisory_xact_lock`; retries after completion still replay the committed response. - `POST /api/etl/process` now supports optional authenticated-principal-scoped `Idempotency-Key` retries with atomic target writes, durable response replay, payload-conflict rejection, and explicit replay response metadata. - `Idempotency-Key` now prefers the quoted RFC 9651 Structured Field String representation while retaining and normalizing the legacy raw representation to the same durable ledger key. diff --git a/docs/etl/durable-job-intake.md b/docs/etl/durable-job-intake.md index 7e29a2d2..05e2b958 100644 --- a/docs/etl/durable-job-intake.md +++ b/docs/etl/durable-job-intake.md @@ -2,17 +2,21 @@ ## Scope -`POST /api/etl/jobs` creates a durable, authenticated-principal-scoped ETL job resource. This -bounded intake slice persists accepted work and exposes its status monitor; it does not execute jobs -yet. Execution, PostgreSQL `FOR UPDATE SKIP LOCKED` claiming, lease fencing, bounded attempts, -terminal payload clearing, and crash recovery belong to the following worker and lease-fencing slice. - -The incomplete intake surface is disabled by default. It is absent unless an operator explicitly -sets the preferred `mightyetl.etl.jobs.intake-enabled=true` property, its supported legacy alias -`xtrmetl.etl.jobs.intake-enabled=true`, or `ETL_JOB_INTAKE_ENABLED=true`. When both full namespaces -are supplied, `mightyetl.*` wins. Enabling intake accepts the temporary boundary that submitted -payloads remain retained in `PENDING` jobs until the worker and terminal payload-clearing slice is -implemented. Deployments that cannot accept that retention boundary must leave the setting false. +`POST /api/etl/jobs` creates a durable, authenticated-principal-scoped ETL job resource. Durable +execution is now implemented as a separate opt-in worker boundary: mightyETL executes at most one +eligible durable job per worker poll, while PostgreSQL owns cross-replica claim arbitration through +`FOR UPDATE SKIP LOCKED`, lease fencing, bounded attempts, and exact conditional lifecycle +transitions. + +Both externally reachable intake and background execution remain disabled by default. Durable job +intake is disabled by default and is absent unless an operator explicitly sets the preferred +`mightyetl.etl.jobs.intake-enabled=true` property, its supported legacy alias +`xtrmetl.etl.jobs.intake-enabled=true`, or `ETL_JOB_INTAKE_ENABLED=true`. The worker is disabled by +default and does not poll until an operator explicitly enables either the preferred +`mightyetl.etl.jobs.worker.enabled=true` property or its supported legacy alias +`xtrmetl.etl.jobs.worker.enabled=true`. When both full namespaces are supplied, `mightyetl.*` wins. +Keeping intake and execution as separate opt-ins allows an operator to stage schema and API rollout +without silently starting background target writes. The existing synchronous `POST /api/etl/process` endpoint remains unchanged. @@ -79,8 +83,58 @@ return `404 etl_job_not_found`; callers cannot use this endpoint to probe anothe existence. The response excludes the request payload, raw principal, raw submission key, and all internal -hashes. Timestamps are explicit ISO-8601 strings. Before worker execution is implemented, newly -accepted jobs remain `PENDING` with an `attemptCount` of zero. +hashes. Timestamps are explicit ISO-8601 strings. A newly accepted job begins as `PENDING`. When the +worker claims it, the database records `RUNNING`, a lease owner, an opaque lease token, lease expiry, +and the incremented attempt count. Terminal success or failure clears the retained request payload. + +## Worker execution and lease fencing + +The scheduled worker claims at most one eligible row per poll. The claim repository serializes the +selection in PostgreSQL with `FOR UPDATE SKIP LOCKED`, so concurrent replicas do not wait on or +execute the same currently claimable row. Eligibility includes new `PENDING` work and reclaimable +expired `RUNNING` work while the configured maximum attempt count has not been exhausted. + +Every mutable lifecycle transition is fenced by the exact claim identity. Success, retry release, +and terminal failure must still match the job record, lease owner, opaque lease token, and valid +lease boundary expected by the caller. A stale or superseded worker therefore cannot overwrite a +newer owner's state. Stale transitions surface as a finite `stale` worker outcome rather than being +silently accepted. + +`EtlJobExecutionService` is transactional. It validates the retained job and durable response-ledger +identity, executes or replays the target operation, and then marks the exact live lease `SUCCEEDED` +in the same transaction. If the success transition is stale, Spring rolls back the target and +response-ledger effects from that execution attempt. + +Transient Spring data-access failures are released for retry only while attempts remain. Exhausted +transient failures become `etl_target_unavailable`; non-transient data-access failures become +`etl_target_failure`; deterministic request and integrity failures retain their stable application +error codes; and unexpected runtime failures become `etl_internal_error`. The worker does not place +raw payloads, principals, submission keys, job identifiers, lease identifiers, SQL, exception class +names, or exception messages into metric labels. + +Worker telemetry uses the fixed terminal outcome vocabulary `idle`, `succeeded`, `retried`, `failed`, +and `stale`. Each completed poll records exactly one outcome and one matching duration sample, +including idle polls and database failures while persisting retry or terminal state. + +## Worker configuration + +The production worker remains fail-closed until explicitly activated. Supported keys include: + +- `mightyetl.etl.jobs.worker.enabled` / `xtrmetl.etl.jobs.worker.enabled`; +- `mightyetl.etl.jobs.worker.fixed-delay-milliseconds` / + `xtrmetl.etl.jobs.worker.fixed-delay-milliseconds`; +- `mightyetl.etl.jobs.worker.initial-delay-milliseconds` / + `xtrmetl.etl.jobs.worker.initial-delay-milliseconds`; +- `mightyetl.etl.jobs.worker.lease-duration-seconds` / + `xtrmetl.etl.jobs.worker.lease-duration-seconds`; +- `mightyetl.etl.jobs.worker.max-attempts` / `xtrmetl.etl.jobs.worker.max-attempts`; and +- `mightyetl.etl.jobs.worker.lease-owner-id` / `xtrmetl.etl.jobs.worker.lease-owner-id`. + +Defaults are bounded in production configuration, and property validation rejects non-positive delay, +lease-duration, and attempt settings as well as blank or oversized lease-owner identifiers. Operators +should assign a stable, non-secret owner identifier per worker instance and size lease duration above +normal execution latency while retaining enough margin for crash recovery through expired-lease +reclamation. ## Validation and persistence @@ -89,43 +143,51 @@ bounds used by synchronous ETL admission. The complete body must be a JSON array fields are rejected, every element must be an object with a safe textual `id`, and normalized field names must remain unique. -Flyway migration `V2__create_etl_job_records.sql` creates `etl_job_records`. All schema objects use +Flyway migration `V2__create_etl_job_records.sql` creates `etl_job_records`; later worker migrations +add lease-fencing columns and a partial eligibility index for claim scans. All schema objects use descriptive multi-word `snake_case` names. The database stores: - an opaque UUID job identifier; - SHA-256 hashes of the principal scope, semantic submission key, and exact JSON text; -- the request payload needed by the future worker while status is `PENDING` or `RUNNING`; -- status, attempt, failure, and timestamp fields. +- the request payload only while the job remains nonterminal; +- status, attempt, failure, lease-owner/token/expiry, and lifecycle timestamp fields. -The schema reserves the stable lifecycle vocabulary `PENDING`, `RUNNING`, `SUCCEEDED`, and `FAILED`. -A database check requires a non-null request payload only for the two nonterminal states and requires -that payload to be null for both terminal states. This makes terminal payload clearing an enforced -persistence invariant rather than a documentation-only convention. +The stable lifecycle vocabulary is `PENDING`, `RUNNING`, `SUCCEEDED`, and `FAILED`. Database checks +require a non-null request payload only for nonterminal states and require the payload to be null for +terminal states. Terminal payload clearing is therefore a persistence invariant, not merely an +application convention. -Raw authenticated principal names and raw idempotency keys are never persisted. The request payload -is sensitive operational data and must inherit the classification of its source records. Until the -worker slice reaches a terminal state and clears it, operators must apply database access control, -encryption, backup, and retention policy accordingly. +Raw authenticated principal names and raw idempotency keys are never persisted. The retained request +payload is sensitive operational data and inherits the classification of its source records. While a +job is `PENDING` or `RUNNING`, operators must protect it with database access control, encryption, +backup, and retention policy appropriate to the underlying records. ## Operational boundary -This slice deliberately does not advertise job completion or background execution. The controller is -disabled by default; setting an activation property to `true` is an explicit operator opt-in to -durable intake without execution. Deployments that need completed asynchronous processing must wait -for the worker and lease-fencing slice. The next slice must claim jobs safely across replicas, fence -stale lease owners, commit target effects and terminal success atomically, reclaim expired leases, -bound attempts, publish stable failure codes, and clear the stored request payload at terminal state. +Enabling intake alone still does not start background processing; enabling the worker alone does not +create externally submitted jobs. A production deployment that wants asynchronous execution must +intentionally enable both surfaces and apply the worker migrations first. Rollback is operationally +safe by disabling the worker property: existing durable rows remain in PostgreSQL and expired +`RUNNING` leases become reclaimable when a compatible worker is enabled again. Operators must not +manually rewrite lease tokens or terminal status to manufacture recovery. + +This slice establishes durable execution, lease fencing, bounded retries, terminal payload clearing, +and finite worker telemetry. Higher-level job-list pagination, polling advisories, conditional status +reads, cancellation, and replay remain separate later stack items and must not be represented as part +of this boundary until their own exact-head gates pass. ## Standards basis - RFC 9110 Section 15.3.3 defines `202 Accepted` as noncommittal and recommends that the response describe current status and point to a status monitor. -- RFC 9457 supplies the problem-details representation used by deterministic submission and lookup - failures. +- RFC 9457 supplies the problem-details representation used by deterministic submission, lookup, and + execution failures. - RFC 9651 defines the current Structured Fields String syntax accepted for `Idempotency-Key`. - The expired IETF HTTPAPI `Idempotency-Key` draft-07 is used only as work-in-progress design evidence for unique client keys, request fingerprints, `422` payload conflicts, and tenant-isolation security concerns. It expired on April 18, 2026 and is not represented as a published RFC. +- PostgreSQL row locking and `SKIP LOCKED` semantics are the database authority for concurrent claim + behavior; the worker does not attempt to replace that arbitration with process-local locking. ### References @@ -138,3 +200,5 @@ bound attempts, publish stable failure codes, and clear the stored request paylo https://www.rfc-editor.org/rfc/rfc9457 - Nottingham, M., & Kamp, P. (2024). *Structured field values for HTTP* (RFC 9651). RFC Editor. https://www.rfc-editor.org/rfc/rfc9651 +- PostgreSQL Global Development Group. (2026). *SELECT*. PostgreSQL 18 documentation. + https://www.postgresql.org/docs/18/sql-select.html diff --git a/docs/operations/durable-job-claim-index-rollout.md b/docs/operations/durable-job-claim-index-rollout.md new file mode 100644 index 00000000..79adde34 --- /dev/null +++ b/docs/operations/durable-job-claim-index-rollout.md @@ -0,0 +1,95 @@ +# Durable-job claim index rollout + +## Purpose + +The durable worker queries `etl_job_records` for the oldest eligible `PENDING` row or expired +`RUNNING` row. The descriptive `etl_job_claim_eligibility_index` supports that queue-like access +path without changing the lease-fencing state machine. + +The table can already receive job submissions when this index is introduced. PostgreSQL's ordinary +`CREATE INDEX` permits reads but blocks `INSERT`, `UPDATE`, and `DELETE` until the build completes. +For a production ETL control plane that write outage is not acceptable. Migration +`V4__add_etl_job_claim_eligibility_index.sql` therefore uses `CREATE INDEX CONCURRENTLY`. + +## Flyway execution boundary + +PostgreSQL rejects `CREATE INDEX CONCURRENTLY` inside a transaction block. The companion script +configuration file +`V4__add_etl_job_claim_eligibility_index.sql.conf` contains: + +```properties +executeInTransaction=false +``` + +Flyway's PostgreSQL transactional advisory lock is also disabled with: + +```properties +spring.flyway.postgresql.transactional-lock=false +``` + +This causes Flyway to use the PostgreSQL integration's non-transactional lock mode required for +concurrent index DDL. The transactional V3 migration remains responsible only for lease columns, +legacy-data repair, and lifecycle constraints. Isolating the index in V4 prevents an index-build +failure from partially committing those schema invariants. + +## Deployment procedure + +1. Keep durable-job intake and worker execution disabled while validating the migration package. +2. Confirm no other concurrent index build or schema migration is active on `etl_job_records`. +3. Apply V3 and verify the lease columns and lifecycle constraints. +4. Apply V4 and monitor `pg_stat_progress_create_index`, database I/O, and transaction latency. +5. Verify `pg_index.indisvalid` is true for `etl_job_claim_eligibility_index`. +6. Run the claim-selection plan against production-equivalent data and confirm the expected index is + available without forcing it through planner settings. +7. Enable one worker canary only after schema history, catalog state, and application health agree. + +`CREATE INDEX CONCURRENTLY` performs more work and can take longer than a regular build. It preserves +normal writes, but it still adds CPU, memory, and I/O load and allows only one concurrent index build +per table. + +## Failed migration and invalid-index recovery + +A failed concurrent build can leave an **invalid index** in the PostgreSQL catalog. Do not mark the +Flyway migration successful merely because an index name exists. + +Recovery is fail-closed: + +1. Keep worker execution disabled and inspect the failed Flyway record plus `pg_index.indisvalid`. +2. Preserve database and migration logs under incident-response controls. +3. Correct the underlying resource, permission, duplicate-build, or transaction-mode cause. +4. Remove an unusable index without blocking normal table access: + + ```sql + DROP INDEX CONCURRENTLY etl_job_claim_eligibility_index; + ``` + +5. Use Flyway repair only after catalog inspection and operator approval, then rerun the unchanged, + checksum-verified migration. +6. Reconfirm index validity and query-plan evidence before enabling workers. + +Do not edit an applied versioned migration or create a same-name replacement with different SQL. + +## Rollback boundary + +Application rollback does not require removing the index; an unused valid index is compatible with +older binaries, although it adds write-maintenance overhead. Remove it only after every deployed +worker version no longer relies on it and a controlled change has verified the performance impact: + +```sql +DROP INDEX CONCURRENTLY etl_job_claim_eligibility_index; +``` + +`DROP INDEX CONCURRENTLY` must also run outside a transaction block. Lease columns and constraints +require a later forward compensating migration after all compatible binaries have been removed; they +must not be rolled back by editing V3. + +## Standards and primary documentation + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Building indexes +concurrently*. https://www.postgresql.org/docs/18/sql-createindex.html#SQL-CREATEINDEX-CONCURRENTLY + +Redgate Software. (2026). *Flyway script configuration*. +https://documentation.red-gate.com/flyway/reference/script-configuration + +Redgate Software. (2026). *Flyway PostgreSQL transactional lock setting*. +https://documentation.red-gate.com/fd/flyway-postgresql-transactional-lock-setting-277579114.html diff --git a/etl-service/src/main/java/com/xtrmetl/etl/config/MightyEtlConfigAliasEnvironmentPostProcessor.java b/etl-service/src/main/java/com/xtrmetl/etl/config/MightyEtlConfigAliasEnvironmentPostProcessor.java index 5462db4b..16a9ad00 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/config/MightyEtlConfigAliasEnvironmentPostProcessor.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/config/MightyEtlConfigAliasEnvironmentPostProcessor.java @@ -28,6 +28,12 @@ public class MightyEtlConfigAliasEnvironmentPostProcessor implements Environment "etl.max-payload-bytes", "etl.max-batch-records", "etl.jobs.intake-enabled", + "etl.jobs.worker.enabled", + "etl.jobs.worker.fixed-delay-milliseconds", + "etl.jobs.worker.initial-delay-milliseconds", + "etl.jobs.worker.lease-duration-seconds", + "etl.jobs.worker.max-attempts", + "etl.jobs.worker.lease-owner-id", "connectors.databricks.enabled", "connectors.snowflake.enabled", "connectors.qlik-sense.enabled" diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobExecutionService.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobExecutionService.java new file mode 100644 index 00000000..74b0192a --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobExecutionService.java @@ -0,0 +1,59 @@ +package com.xtrmetl.etl.job; + +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.util.Objects; + +/** + * Executes one claimed ETL payload and commits ledger, target, and terminal success atomically. + * + *

{@link EtlJobIdempotencyService} verifies the retained payload identity, acquires the durable + * execution-ledger lock, replays or writes the response ledger, and writes target rows. The + * subsequent conditional success transition must match the exact unexpired claim. If that + * transition reports a stale lease, {@link StaleEtlJobLeaseException} escapes and Spring rolls back + * every target and response-ledger write made by the same transaction.

+ */ +@Service +public class EtlJobExecutionService { + + private final EtlJobIdempotencyService idempotencyService; + private final EtlJobLeaseRepository leaseRepository; + + /** + * Creates the atomic durable-job execution boundary. + * + * @param idempotencyService hashed response-ledger and target execution service + * @param leaseRepository exact lease-fenced lifecycle persistence + */ + public EtlJobExecutionService( + EtlJobIdempotencyService idempotencyService, + EtlJobLeaseRepository leaseRepository + ) { + this.idempotencyService = Objects.requireNonNull( + idempotencyService, + "idempotencyService must not be null" + ); + this.leaseRepository = Objects.requireNonNull( + leaseRepository, + "leaseRepository must not be null" + ); + } + + /** + * Processes or replays the retained job and marks the exact live claim successful atomically. + * + * @param lease exact database claim to execute + * @throws NullPointerException when the lease is {@code null} + * @throws EtlJobIntegrityException when persisted job or ledger identity conflicts + * @throws com.xtrmetl.etl.service.EtlRequestException when the retained request is invalid + * @throws org.springframework.dao.DataAccessException when locking or a database write fails + * @throws StaleEtlJobLeaseException when the claim expires or is superseded before success + */ + @Transactional + public void execute(EtlJobLease lease) { + EtlJobLease requiredLease = Objects.requireNonNull(lease, "lease must not be null"); + idempotencyService.process(requiredLease); + leaseRepository.markSucceeded(requiredLease); + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobIdempotencyService.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobIdempotencyService.java new file mode 100644 index 00000000..cc9d3965 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobIdempotencyService.java @@ -0,0 +1,148 @@ +package com.xtrmetl.etl.job; + +import com.xtrmetl.etl.service.EtlRequestLock; +import com.xtrmetl.etl.service.EtlService; +import com.xtrmetl.etl.service.Sha256Digest; +import org.springframework.dao.CannotAcquireLockException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Service; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import java.util.List; +import java.util.Objects; + +/** + * Reuses the durable ETL response ledger with hashed job identity only. + * + *

The accepted job stores independent hashes of its authenticated principal and normalized + * submission key. This service domain-separates and hashes those values into one response-ledger + * key, verifies the exact retained payload digest, serializes execution with the existing + * transaction-level request lock, replays an existing matching response, or writes the target and + * response ledger in the caller-owned lease-fenced transaction. Raw principals and client keys are + * neither required nor reconstructed.

+ * + *

One invocation represents exactly one persisted durable attempt. This service deliberately + * creates no transaction and performs no in-process retry. It joins the transaction owned by + * {@link EtlJobExecutionService}, calls the non-retrying ETL entry point once, and lets transient + * failures escape to {@link EtlJobWorker}, which owns bounded retry accounting in + * {@code attempt_count}. Direct Spring-proxy invocation without an existing transaction fails + * before lock or JDBC access.

+ */ +@Service +public class EtlJobIdempotencyService { + + private static final String LEDGER_KEY_DOMAIN = "mightyetl:durable-job:v1:"; + private static final String SELECT_LEDGER_SQL = """ + SELECT request_digest, response_body + FROM etl_idempotency_records + WHERE idempotency_key_hash = ? + """; + private static final String INSERT_LEDGER_SQL = """ + INSERT INTO etl_idempotency_records ( + idempotency_key_hash, + request_digest, + response_body + ) VALUES (?, ?, ?) + """; + + private final JdbcTemplate jdbcTemplate; + private final EtlService etlService; + private final EtlRequestLock requestLock; + + /** + * Creates the hashed durable-job response-ledger adapter. + * + * @param jdbcTemplate parameterized response-ledger database access + * @param etlService validated ETL target writer + * @param requestLock transaction-lifetime response-ledger lock + */ + public EtlJobIdempotencyService( + JdbcTemplate jdbcTemplate, + EtlService etlService, + EtlRequestLock requestLock + ) { + this.jdbcTemplate = Objects.requireNonNull( + jdbcTemplate, + "jdbcTemplate must not be null" + ); + this.etlService = Objects.requireNonNull(etlService, "etlService must not be null"); + this.requestLock = Objects.requireNonNull(requestLock, "requestLock must not be null"); + } + + /** + * Executes or replays one durable job inside the caller's exact lease-fenced transaction. + * + *

The method has neither transaction-creation nor retry advice. The caller must establish + * the transaction that also contains terminal success fencing; otherwise execution fails + * before any request lock, target write, or response-ledger access. This prevents a direct + * proxy caller from committing durable effects without the lease-success predicate.

+ * + * @param lease exact live claim carrying hashed execution identity and retained payload + * @return newly generated or replayed stable response body + * @throws NullPointerException when the lease is {@code null} + * @throws IllegalStateException when invoked without an actual transaction + * @throws EtlJobIntegrityException when retained payload or ledger identity conflicts + * @throws CannotAcquireLockException when another transaction owns the execution ledger key + */ + public String process(EtlJobLease lease) { + EtlJobLease requiredLease = Objects.requireNonNull(lease, "lease must not be null"); + requireActiveTransaction(); + if (!Sha256Digest.digest(requiredLease.requestPayload()).equals( + requiredLease.requestDigest() + )) { + throw new EtlJobIntegrityException(); + } + + String ledgerKeyHash = Sha256Digest.digest( + LEDGER_KEY_DOMAIN + + requiredLease.principalScopeHash() + + ':' + + requiredLease.submissionKeyHash() + ); + if (!requestLock.tryLock(ledgerKeyHash)) { + throw new CannotAcquireLockException("Durable ETL execution ledger is busy"); + } + + List storedResponses = jdbcTemplate.query( + SELECT_LEDGER_SQL, + (resultSet, rowNumber) -> new StoredResponse( + resultSet.getString("request_digest"), + resultSet.getString("response_body") + ), + ledgerKeyHash + ); + if (!storedResponses.isEmpty()) { + StoredResponse storedResponse = storedResponses.getFirst(); + if (!storedResponse.requestDigest().equals(requiredLease.requestDigest())) { + throw new EtlJobIntegrityException(); + } + return storedResponse.responseBody(); + } + + String responseBody = etlService.processDataInExistingTransaction( + requiredLease.requestPayload() + ); + jdbcTemplate.update( + INSERT_LEDGER_SQL, + ledgerKeyHash, + requiredLease.requestDigest(), + responseBody + ); + return responseBody; + } + + private static void requireActiveTransaction() { + if (!TransactionSynchronizationManager.isActualTransactionActive()) { + throw new IllegalStateException( + "Durable ETL job execution requires an active transaction" + ); + } + } + + private record StoredResponse(String requestDigest, String responseBody) { + private StoredResponse { + Objects.requireNonNull(requestDigest, "requestDigest must not be null"); + Objects.requireNonNull(responseBody, "responseBody must not be null"); + } + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobIntegrityException.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobIntegrityException.java new file mode 100644 index 00000000..7aa04390 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobIntegrityException.java @@ -0,0 +1,30 @@ +package com.xtrmetl.etl.job; + +/** + * Signals that persisted durable-job execution identity no longer matches retained or ledger data. + * + *

The exception exposes only one stable machine-readable failure code and a non-sensitive + * message. It deliberately omits payloads, hashes, identifiers, SQL, timestamps, and stored + * response bodies so accidental logging does not disclose customer or operational data.

+ */ +public class EtlJobIntegrityException extends RuntimeException { + + /** Stable terminal failure code for payload or response-ledger integrity mismatches. */ + public static final String FAILURE_CODE = "etl_job_integrity_failure"; + + /** + * Creates a non-sensitive integrity failure signal. + */ + public EtlJobIntegrityException() { + super("Durable ETL job execution identity failed integrity validation"); + } + + /** + * Returns the stable machine-readable terminal failure classification. + * + * @return {@value #FAILURE_CODE} + */ + public String failureCode() { + return FAILURE_CODE; + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobLease.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobLease.java new file mode 100644 index 00000000..321bab22 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobLease.java @@ -0,0 +1,75 @@ +package com.xtrmetl.etl.job; + +import java.time.Instant; +import java.util.Objects; +import java.util.UUID; +import java.util.regex.Pattern; + +/** + * Carries one immutable, fenced database claim into durable idempotent ETL execution. + * + *

The three lowercase SHA-256 values are non-reversible persistence identifiers copied from the + * accepted job row. They let the worker reuse the durable response ledger without retaining or + * reconstructing raw authenticated principals or raw client idempotency keys.

+ * + * @param jobRecordId durable job identifier + * @param leaseClaimId unique token generated for this exact claim or reclaim + * @param leaseOwnerId non-sensitive process-lifetime worker identifier + * @param principalScopeHash SHA-256 hash of the authenticated principal namespace + * @param submissionKeyHash SHA-256 hash of the normalized durable submission key + * @param requestDigest SHA-256 digest of the exact retained request payload + * @param requestPayload validated JSON payload retained while the job is non-terminal + * @param attemptCount one-based claim attempt count after this claim was persisted + * @param leaseExpiresAt database-derived instant after which this claim is stale + */ +public record EtlJobLease( + UUID jobRecordId, + UUID leaseClaimId, + String leaseOwnerId, + String principalScopeHash, + String submissionKeyHash, + String requestDigest, + String requestPayload, + int attemptCount, + Instant leaseExpiresAt +) { + + private static final Pattern SAFE_LEASE_OWNER_PATTERN = Pattern.compile( + "[A-Za-z0-9._:-]{8,128}" + ); + private static final Pattern SHA256_HEX_PATTERN = Pattern.compile("[0-9a-f]{64}"); + + /** + * Validates every field needed for exact lease fencing and deterministic execution. + */ + public EtlJobLease { + Objects.requireNonNull(jobRecordId, "jobRecordId must not be null"); + Objects.requireNonNull(leaseClaimId, "leaseClaimId must not be null"); + String requiredOwnerId = Objects.requireNonNull( + leaseOwnerId, + "leaseOwnerId must not be null" + ); + if (!SAFE_LEASE_OWNER_PATTERN.matcher(requiredOwnerId).matches()) { + throw new IllegalArgumentException( + "leaseOwnerId must match [A-Za-z0-9._:-]{8,128}" + ); + } + requireSha256Hex(principalScopeHash, "principalScopeHash"); + requireSha256Hex(submissionKeyHash, "submissionKeyHash"); + requireSha256Hex(requestDigest, "requestDigest"); + Objects.requireNonNull(requestPayload, "requestPayload must not be null"); + if (attemptCount < 1) { + throw new IllegalArgumentException("attemptCount must be positive"); + } + Objects.requireNonNull(leaseExpiresAt, "leaseExpiresAt must not be null"); + } + + private static void requireSha256Hex(String value, String fieldName) { + String requiredValue = Objects.requireNonNull(value, fieldName + " must not be null"); + if (!SHA256_HEX_PATTERN.matcher(requiredValue).matches()) { + throw new IllegalArgumentException( + fieldName + " must be lowercase 64-character SHA-256 hex" + ); + } + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobLeaseRepository.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobLeaseRepository.java new file mode 100644 index 00000000..f88f50b6 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobLeaseRepository.java @@ -0,0 +1,387 @@ +package com.xtrmetl.etl.job; + +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; + +import java.time.Duration; +import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; +import java.util.regex.Pattern; + +/** + * Owns PostgreSQL-backed durable-job claims and exact lease-fenced state transitions. + * + *

Every claim transaction first terminalizes eligible exhausted rows, then locks at most one + * oldest eligible row with {@code FOR UPDATE SKIP LOCKED}, and finally writes a fresh claim token, + * owner, expiry, and incremented attempt count before commit. State transitions repeat the exact + * claim token, owner, running status, and database-time expiry predicates so stale workers cannot + * mutate lifecycle state. Public callers cannot create leases longer than the worker's one-day + * operational ceiling, even when they bypass Spring configuration binding.

+ * + *

Terminal success is intentionally stricter than retry or failure bookkeeping: it is accepted + * only inside the caller-owned transaction that also contains the target effects and durable + * response-ledger write. This prevents a direct repository call from publishing false success + * independently of the data it claims to have committed.

+ */ +@Repository +public class EtlJobLeaseRepository { + + /** Stable terminal code assigned when no additional claim is permitted. */ + public static final String ATTEMPTS_EXHAUSTED_FAILURE_CODE = + "etl_worker_attempts_exhausted"; + + private static final Pattern SAFE_OWNER_PATTERN = Pattern.compile( + "[A-Za-z0-9._:-]{8,128}" + ); + private static final Pattern SAFE_FAILURE_CODE_PATTERN = Pattern.compile( + "[a-z][a-z0-9_]{2,127}" + ); + + private static final String TERMINALIZE_EXHAUSTED_SQL = """ + UPDATE etl_job_records + SET job_status = 'FAILED', + request_payload = NULL, + failure_code = ?, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + updated_at = CURRENT_TIMESTAMP + WHERE attempt_count >= ? + AND ( + job_status = 'PENDING' + OR ( + job_status = 'RUNNING' + AND lease_expires_at <= CURRENT_TIMESTAMP + ) + ) + """; + + private static final String SELECT_CANDIDATE_SQL = """ + SELECT job_record_id, + principal_scope_hash, + submission_key_hash, + request_digest, + request_payload, + attempt_count, + CURRENT_TIMESTAMP AS database_now + FROM etl_job_records + WHERE attempt_count < ? + AND ( + job_status = 'PENDING' + OR ( + job_status = 'RUNNING' + AND lease_expires_at <= CURRENT_TIMESTAMP + ) + ) + ORDER BY created_at, job_record_id + FETCH FIRST 1 ROW ONLY + FOR UPDATE SKIP LOCKED + """; + + private static final String CLAIM_CANDIDATE_SQL = """ + UPDATE etl_job_records + SET job_status = 'RUNNING', + attempt_count = attempt_count + 1, + failure_code = NULL, + lease_claim_id = ?, + lease_owner_id = ?, + lease_expires_at = ?, + updated_at = CURRENT_TIMESTAMP + WHERE job_record_id = ? + AND attempt_count = ? + AND ( + job_status = 'PENDING' + OR ( + job_status = 'RUNNING' + AND lease_expires_at <= CURRENT_TIMESTAMP + ) + ) + """; + + private static final String MARK_SUCCEEDED_SQL = """ + UPDATE etl_job_records + SET job_status = 'SUCCEEDED', + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + updated_at = CURRENT_TIMESTAMP + WHERE job_record_id = ? + AND job_status = 'RUNNING' + AND lease_claim_id = ? + AND lease_owner_id = ? + AND lease_expires_at > CURRENT_TIMESTAMP + """; + + private static final String RELEASE_FOR_RETRY_SQL = """ + UPDATE etl_job_records + SET job_status = 'PENDING', + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + updated_at = CURRENT_TIMESTAMP + WHERE job_record_id = ? + AND job_status = 'RUNNING' + AND lease_claim_id = ? + AND lease_owner_id = ? + AND lease_expires_at > CURRENT_TIMESTAMP + AND attempt_count < ? + """; + + private static final String MARK_FAILED_SQL = """ + UPDATE etl_job_records + SET job_status = 'FAILED', + request_payload = NULL, + failure_code = ?, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + updated_at = CURRENT_TIMESTAMP + WHERE job_record_id = ? + AND job_status = 'RUNNING' + AND lease_claim_id = ? + AND lease_owner_id = ? + AND lease_expires_at > CURRENT_TIMESTAMP + """; + + private final JdbcTemplate jdbcTemplate; + private final TransactionTemplate transactionTemplate; + + /** + * Creates lease persistence using one JDBC adapter and one transaction authority. + * + * @param jdbcTemplate JDBC operations for the durable-job table + * @param transactionManager transaction manager that owns row locks and claim commits + */ + public EtlJobLeaseRepository( + JdbcTemplate jdbcTemplate, + PlatformTransactionManager transactionManager + ) { + this.jdbcTemplate = Objects.requireNonNull( + jdbcTemplate, + "jdbcTemplate must not be null" + ); + this.transactionTemplate = new TransactionTemplate(Objects.requireNonNull( + transactionManager, + "transactionManager must not be null" + )); + } + + /** + * Claims at most one oldest eligible job for one worker process. + * + * @param leaseOwnerId safe non-sensitive process identifier + * @param leaseDuration duration from one second through one day + * @param maxAttempts maximum permitted claim count from 1 through 100 + * @return a fresh claim, or an empty result when no row is eligible + * @throws NullPointerException when an argument is {@code null} + * @throws IllegalArgumentException when an argument violates its bounded contract + * @throws IllegalStateException when a locked candidate unexpectedly cannot be claimed + */ + public Optional claimNext( + String leaseOwnerId, + Duration leaseDuration, + int maxAttempts + ) { + String validatedOwnerId = requireSafeOwnerId(leaseOwnerId); + Duration validatedDuration = requirePositiveDuration(leaseDuration); + int validatedMaxAttempts = requireMaxAttempts(maxAttempts); + + return Objects.requireNonNull(transactionTemplate.execute(transactionStatus -> { + jdbcTemplate.update( + TERMINALIZE_EXHAUSTED_SQL, + ATTEMPTS_EXHAUSTED_FAILURE_CODE, + validatedMaxAttempts + ); + List candidates = jdbcTemplate.query( + SELECT_CANDIDATE_SQL, + (resultSet, rowNumber) -> new ClaimCandidate( + resultSet.getObject("job_record_id", UUID.class), + resultSet.getString("principal_scope_hash"), + resultSet.getString("submission_key_hash"), + resultSet.getString("request_digest"), + resultSet.getString("request_payload"), + resultSet.getInt("attempt_count"), + resultSet.getObject("database_now", OffsetDateTime.class).toInstant() + ), + validatedMaxAttempts + ); + if (candidates.isEmpty()) { + return Optional.empty(); + } + + ClaimCandidate candidate = candidates.getFirst(); + UUID leaseClaimId = UUID.randomUUID(); + Instant leaseExpiresAt = candidate.databaseNow().plus(validatedDuration); + int updatedRows = jdbcTemplate.update( + CLAIM_CANDIDATE_SQL, + leaseClaimId, + validatedOwnerId, + OffsetDateTime.ofInstant(leaseExpiresAt, ZoneOffset.UTC), + candidate.jobRecordId(), + candidate.attemptCount() + ); + if (updatedRows != 1) { + throw new IllegalStateException("Locked ETL job candidate could not be claimed"); + } + return Optional.of(new EtlJobLease( + candidate.jobRecordId(), + leaseClaimId, + validatedOwnerId, + candidate.principalScopeHash(), + candidate.submissionKeyHash(), + candidate.requestDigest(), + candidate.requestPayload(), + candidate.attemptCount() + 1, + leaseExpiresAt + )); + }), "claim transaction must return a result"); + } + + /** + * Commits terminal success only for the exact live claim in the atomic execution transaction. + * + * @param lease exact claim whose target effects completed in the same transaction + * @throws NullPointerException when the lease is {@code null} + * @throws IllegalStateException when no actual Spring transaction is active + * @throws StaleEtlJobLeaseException when the claim is expired or no longer authoritative + */ + public void markSucceeded(EtlJobLease lease) { + EtlJobLease requiredLease = Objects.requireNonNull(lease, "lease must not be null"); + requireActiveSuccessTransaction(); + requireTransition(jdbcTemplate.update( + MARK_SUCCEEDED_SQL, + requiredLease.jobRecordId(), + requiredLease.leaseClaimId(), + requiredLease.leaseOwnerId() + )); + } + + /** + * Returns a failed execution to pending only while attempts remain and the claim is exact. + * + * @param lease exact live claim to release + * @param maxAttempts maximum permitted claim count from 1 through 100 + * @throws NullPointerException when the lease is {@code null} + * @throws IllegalArgumentException when the maximum is outside the supported range + * @throws StaleEtlJobLeaseException when the claim is stale or no retry remains + */ + public void releaseForRetry(EtlJobLease lease, int maxAttempts) { + EtlJobLease requiredLease = Objects.requireNonNull(lease, "lease must not be null"); + int validatedMaxAttempts = requireMaxAttempts(maxAttempts); + requireTransition(jdbcTemplate.update( + RELEASE_FOR_RETRY_SQL, + requiredLease.jobRecordId(), + requiredLease.leaseClaimId(), + requiredLease.leaseOwnerId(), + validatedMaxAttempts + )); + } + + /** + * Commits terminal failure and clears the retained payload for the exact live claim. + * + * @param lease exact live claim to fail + * @param failureCode stable non-sensitive machine-readable failure classification + * @throws NullPointerException when an argument is {@code null} + * @throws IllegalArgumentException when the failure code is unsafe + * @throws StaleEtlJobLeaseException when the claim is expired or no longer authoritative + */ + public void markFailed(EtlJobLease lease, String failureCode) { + EtlJobLease requiredLease = Objects.requireNonNull(lease, "lease must not be null"); + String validatedFailureCode = requireSafeFailureCode(failureCode); + requireTransition(jdbcTemplate.update( + MARK_FAILED_SQL, + validatedFailureCode, + requiredLease.jobRecordId(), + requiredLease.leaseClaimId(), + requiredLease.leaseOwnerId() + )); + } + + private static String requireSafeOwnerId(String leaseOwnerId) { + String requiredOwnerId = Objects.requireNonNull( + leaseOwnerId, + "leaseOwnerId must not be null" + ); + if (!SAFE_OWNER_PATTERN.matcher(requiredOwnerId).matches()) { + throw new IllegalArgumentException( + "leaseOwnerId must match [A-Za-z0-9._:-]{8,128}" + ); + } + return requiredOwnerId; + } + + private static Duration requirePositiveDuration(Duration leaseDuration) { + Duration requiredDuration = Objects.requireNonNull( + leaseDuration, + "leaseDuration must not be null" + ); + Duration maximumDuration = Duration.ofSeconds( + EtlJobWorkerProperties.MAXIMUM_LEASE_DURATION_SECONDS + ); + if (requiredDuration.isZero() + || requiredDuration.isNegative() + || requiredDuration.compareTo(maximumDuration) > 0) { + throw new IllegalArgumentException( + "leaseDuration must be between one second and one day" + ); + } + return requiredDuration; + } + + private static int requireMaxAttempts(int maxAttempts) { + if (maxAttempts < 1 || maxAttempts > 100) { + throw new IllegalArgumentException("maxAttempts must be between 1 and 100"); + } + return maxAttempts; + } + + private static String requireSafeFailureCode(String failureCode) { + String requiredFailureCode = Objects.requireNonNull( + failureCode, + "failureCode must not be null" + ); + if (!SAFE_FAILURE_CODE_PATTERN.matcher(requiredFailureCode).matches()) { + throw new IllegalArgumentException( + "failureCode must match [a-z][a-z0-9_]{2,127}" + ); + } + return requiredFailureCode; + } + + private static void requireActiveSuccessTransaction() { + if (!TransactionSynchronizationManager.isActualTransactionActive()) { + throw new IllegalStateException( + "Durable ETL success requires an active transaction" + ); + } + } + + private static void requireTransition(int updatedRows) { + if (updatedRows != 1) { + throw new StaleEtlJobLeaseException(); + } + } + + private record ClaimCandidate( + UUID jobRecordId, + String principalScopeHash, + String submissionKeyHash, + String requestDigest, + String requestPayload, + int attemptCount, + Instant databaseNow + ) { + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobWorker.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobWorker.java new file mode 100644 index 00000000..ef3737c2 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobWorker.java @@ -0,0 +1,212 @@ +package com.xtrmetl.etl.job; + +import com.xtrmetl.etl.service.EtlRequestException; +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Timer; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.dao.DataAccessException; +import org.springframework.dao.TransientDataAccessException; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; + +/** + * Polls and executes at most one durable ETL job per fixed-delay invocation. + * + *

The database claim repository, not the scheduler, distributes work across replicas. The + * worker classifies failures into stable non-sensitive codes, retries only transient database + * failures while attempts remain, and treats every failed exact-lease transition as stale evidence. + * Metrics use a fixed terminal-outcome vocabulary and never tag payloads, principals, keys, job + * identifiers, lease identifiers, SQL, exception classes, or exception messages. Every completed + * poll records exactly one terminal outcome counter and one matching duration sample, including + * idle polls and database failures while persisting retry or terminal transitions.

+ */ +@Component +@ConditionalOnBooleanProperty( + prefix = "xtrmetl.etl.jobs.worker", + name = "enabled", + havingValue = true, + matchIfMissing = false +) +public class EtlJobWorker { + + /** Stable target-unavailability code used after transient attempts are exhausted. */ + public static final String TARGET_UNAVAILABLE_FAILURE_CODE = "etl_target_unavailable"; + + /** Stable non-transient database failure code. */ + public static final String TARGET_FAILURE_CODE = "etl_target_failure"; + + /** Stable unexpected implementation failure code. */ + public static final String INTERNAL_FAILURE_CODE = "etl_internal_error"; + + private static final String METRIC_OUTCOMES = "etl.jobs.worker.outcomes"; + private static final String METRIC_DURATION = "etl.jobs.execution.duration"; + private static final String IDLE_OUTCOME = "idle"; + private static final String SUCCEEDED_OUTCOME = "succeeded"; + private static final String RETRIED_OUTCOME = "retried"; + private static final String FAILED_OUTCOME = "failed"; + private static final String STALE_OUTCOME = "stale"; + private static final List FINITE_OUTCOMES = List.of( + IDLE_OUTCOME, + SUCCEEDED_OUTCOME, + RETRIED_OUTCOME, + FAILED_OUTCOME, + STALE_OUTCOME + ); + + private final EtlJobLeaseRepository leaseRepository; + private final EtlJobExecutionService executionService; + private final EtlJobWorkerProperties properties; + private final MeterRegistry meterRegistry; + private final Map outcomeCounters; + private final Map outcomeTimers; + + /** + * Creates one fail-closed worker and pre-registers its finite metric vocabulary. + * + * @param leaseRepository database claim and transition authority + * @param executionService atomic target-write and success boundary + * @param properties bounded worker configuration + * @param meterRegistry metrics registry for finite-cardinality evidence + */ + public EtlJobWorker( + EtlJobLeaseRepository leaseRepository, + EtlJobExecutionService executionService, + EtlJobWorkerProperties properties, + MeterRegistry meterRegistry + ) { + this.leaseRepository = Objects.requireNonNull( + leaseRepository, + "leaseRepository must not be null" + ); + this.executionService = Objects.requireNonNull( + executionService, + "executionService must not be null" + ); + this.properties = Objects.requireNonNull(properties, "properties must not be null"); + this.meterRegistry = Objects.requireNonNull( + meterRegistry, + "meterRegistry must not be null" + ); + + Map counters = new LinkedHashMap<>(); + Map timers = new LinkedHashMap<>(); + for (String outcome : FINITE_OUTCOMES) { + counters.put( + outcome, + Counter.builder(METRIC_OUTCOMES) + .description("Durable ETL worker terminal outcomes") + .tag("outcome", outcome) + .register(this.meterRegistry) + ); + timers.put( + outcome, + Timer.builder(METRIC_DURATION) + .description("Duration of one durable ETL worker poll") + .tag("outcome", outcome) + .register(this.meterRegistry) + ); + } + this.outcomeCounters = Map.copyOf(counters); + this.outcomeTimers = Map.copyOf(timers); + } + + /** + * Claims and handles at most one eligible durable job. + * + *

Fixed delay is measured after this invocation completes. A database outage during claim or + * a database failure while persisting retry or terminal state is converted into a finite failed + * outcome without copying diagnostic text into application logs or telemetry. Spring invokes the + * method only when worker activation is explicitly enabled.

+ */ + @Scheduled( + fixedDelayString = "${xtrmetl.etl.jobs.worker.fixed-delay-milliseconds:5000}", + initialDelayString = "${xtrmetl.etl.jobs.worker.initial-delay-milliseconds:5000}" + ) + public void pollOnce() { + Timer.Sample sample = Timer.start(meterRegistry); + String finalOutcome = runOnePoll(); + sample.stop(outcomeTimers.get(finalOutcome)); + } + + private String runOnePoll() { + final Optional claimedLease; + try { + claimedLease = leaseRepository.claimNext( + properties.getLeaseOwnerId(), + Duration.ofSeconds(properties.getLeaseDurationSeconds()), + properties.getMaxAttempts() + ); + } catch (DataAccessException exception) { + increment(FAILED_OUTCOME); + return FAILED_OUTCOME; + } + + if (claimedLease.isEmpty()) { + increment(IDLE_OUTCOME); + return IDLE_OUTCOME; + } + + EtlJobLease lease = claimedLease.orElseThrow(); + try { + executionService.execute(lease); + increment(SUCCEEDED_OUTCOME); + return SUCCEEDED_OUTCOME; + } catch (StaleEtlJobLeaseException exception) { + increment(STALE_OUTCOME); + return STALE_OUTCOME; + } catch (TransientDataAccessException exception) { + return handleTransientFailure(lease); + } catch (EtlJobIntegrityException exception) { + return markFailedOrStale(lease, exception.failureCode()); + } catch (EtlRequestException exception) { + return markFailedOrStale(lease, exception.error().errorCode()); + } catch (DataAccessException exception) { + return markFailedOrStale(lease, TARGET_FAILURE_CODE); + } catch (RuntimeException exception) { + return markFailedOrStale(lease, INTERNAL_FAILURE_CODE); + } + } + + private String handleTransientFailure(EtlJobLease lease) { + if (lease.attemptCount() < properties.getMaxAttempts()) { + try { + leaseRepository.releaseForRetry(lease, properties.getMaxAttempts()); + increment(RETRIED_OUTCOME); + return RETRIED_OUTCOME; + } catch (StaleEtlJobLeaseException exception) { + increment(STALE_OUTCOME); + return STALE_OUTCOME; + } catch (DataAccessException exception) { + increment(FAILED_OUTCOME); + return FAILED_OUTCOME; + } + } + return markFailedOrStale(lease, TARGET_UNAVAILABLE_FAILURE_CODE); + } + + private String markFailedOrStale(EtlJobLease lease, String failureCode) { + try { + leaseRepository.markFailed(lease, failureCode); + increment(FAILED_OUTCOME); + return FAILED_OUTCOME; + } catch (StaleEtlJobLeaseException exception) { + increment(STALE_OUTCOME); + return STALE_OUTCOME; + } catch (DataAccessException exception) { + increment(FAILED_OUTCOME); + return FAILED_OUTCOME; + } + } + + private void increment(String outcome) { + outcomeCounters.get(outcome).increment(); + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobWorkerProperties.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobWorkerProperties.java new file mode 100644 index 00000000..03047429 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobWorkerProperties.java @@ -0,0 +1,192 @@ +package com.xtrmetl.etl.job; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +import java.util.Objects; +import java.util.UUID; +import java.util.regex.Pattern; + +/** + * Holds bounded, fail-closed configuration for durable ETL job execution. + * + *

The worker is disabled unless an operator explicitly enables it. Polling delays and lease + * durations are capped at one day so a malformed environment value cannot create effectively + * permanent scheduling gaps, arithmetic overflow, or a lease that prevents timely crash recovery. + * A process-lifetime lease owner identifier is generated when no external value is supplied. The + * identifier is deliberately restricted to a short safe ASCII profile because it is persisted as + * operational metadata and must never become a free-form log or database injection surface.

+ */ +@ConfigurationProperties(prefix = "xtrmetl.etl.jobs.worker") +public class EtlJobWorkerProperties { + + /** Maximum supported fixed or initial scheduler delay: one day in milliseconds. */ + public static final long MAXIMUM_SCHEDULER_DELAY_MILLISECONDS = 86_400_000L; + + /** Maximum supported durable-job lease duration: one day in seconds. */ + public static final long MAXIMUM_LEASE_DURATION_SECONDS = 86_400L; + + private static final Pattern SAFE_LEASE_OWNER_PATTERN = Pattern.compile( + "[A-Za-z0-9._:-]{8,128}" + ); + + private boolean enabled; + private long fixedDelayMilliseconds = 5_000L; + private long initialDelayMilliseconds = 5_000L; + private long leaseDurationSeconds = 300L; + private int maxAttempts = 3; + private String leaseOwnerId = "worker-" + UUID.randomUUID(); + + /** + * Creates disabled worker configuration with bounded production-safe defaults. + */ + public EtlJobWorkerProperties() { + // Spring Boot binds through the public setters while preserving generated defaults. + } + + /** + * Reports whether scheduled durable-job execution is explicitly enabled. + * + * @return {@code true} only when an operator enabled the worker + */ + public boolean isEnabled() { + return enabled; + } + + /** + * Enables or disables scheduled durable-job execution. + * + * @param enabled whether the worker should run + */ + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + /** + * Returns the delay measured after one polling invocation completes. + * + * @return fixed delay from one millisecond through one day + */ + public long getFixedDelayMilliseconds() { + return fixedDelayMilliseconds; + } + + /** + * Sets the delay measured after one polling invocation completes. + * + * @param fixedDelayMilliseconds delay from one millisecond through one day + * @throws IllegalArgumentException when the delay is outside the supported range + */ + public void setFixedDelayMilliseconds(long fixedDelayMilliseconds) { + if (fixedDelayMilliseconds < 1L + || fixedDelayMilliseconds > MAXIMUM_SCHEDULER_DELAY_MILLISECONDS) { + throw new IllegalArgumentException( + "fixedDelayMilliseconds must be between 1 and " + + MAXIMUM_SCHEDULER_DELAY_MILLISECONDS + ); + } + this.fixedDelayMilliseconds = fixedDelayMilliseconds; + } + + /** + * Returns the delay before the first polling invocation after application startup. + * + * @return initial delay from zero milliseconds through one day + */ + public long getInitialDelayMilliseconds() { + return initialDelayMilliseconds; + } + + /** + * Sets the delay before the first polling invocation after application startup. + * + * @param initialDelayMilliseconds delay from zero milliseconds through one day + * @throws IllegalArgumentException when the delay is outside the supported range + */ + public void setInitialDelayMilliseconds(long initialDelayMilliseconds) { + if (initialDelayMilliseconds < 0L + || initialDelayMilliseconds > MAXIMUM_SCHEDULER_DELAY_MILLISECONDS) { + throw new IllegalArgumentException( + "initialDelayMilliseconds must be between 0 and " + + MAXIMUM_SCHEDULER_DELAY_MILLISECONDS + ); + } + this.initialDelayMilliseconds = initialDelayMilliseconds; + } + + /** + * Returns how long one database claim remains valid without renewal. + * + * @return lease duration from one second through one day + */ + public long getLeaseDurationSeconds() { + return leaseDurationSeconds; + } + + /** + * Sets how long one database claim remains valid without renewal. + * + * @param leaseDurationSeconds duration from one second through one day + * @throws IllegalArgumentException when the duration is outside the supported range + */ + public void setLeaseDurationSeconds(long leaseDurationSeconds) { + if (leaseDurationSeconds < 1L + || leaseDurationSeconds > MAXIMUM_LEASE_DURATION_SECONDS) { + throw new IllegalArgumentException( + "leaseDurationSeconds must be between 1 and " + + MAXIMUM_LEASE_DURATION_SECONDS + ); + } + this.leaseDurationSeconds = leaseDurationSeconds; + } + + /** + * Returns the maximum number of claims permitted before terminal failure. + * + * @return maximum attempt count from 1 through 100 + */ + public int getMaxAttempts() { + return maxAttempts; + } + + /** + * Sets the maximum number of claims permitted before terminal failure. + * + * @param maxAttempts maximum attempt count from 1 through 100 + * @throws IllegalArgumentException when the value is outside the supported range + */ + public void setMaxAttempts(int maxAttempts) { + if (maxAttempts < 1 || maxAttempts > 100) { + throw new IllegalArgumentException("maxAttempts must be between 1 and 100"); + } + this.maxAttempts = maxAttempts; + } + + /** + * Returns the non-sensitive process identifier persisted on active leases. + * + * @return safe process-lifetime lease owner identifier + */ + public String getLeaseOwnerId() { + return leaseOwnerId; + } + + /** + * Sets the non-sensitive process identifier persisted on active leases. + * + * @param leaseOwnerId 8-to-128-character safe ASCII process identifier + * @throws NullPointerException when the identifier is {@code null} + * @throws IllegalArgumentException when the identifier is too short, too long, or unsafe + */ + public void setLeaseOwnerId(String leaseOwnerId) { + String requiredOwnerId = Objects.requireNonNull( + leaseOwnerId, + "leaseOwnerId must not be null" + ); + if (!SAFE_LEASE_OWNER_PATTERN.matcher(requiredOwnerId).matches()) { + throw new IllegalArgumentException( + "leaseOwnerId must match [A-Za-z0-9._:-]{8,128}" + ); + } + this.leaseOwnerId = requiredOwnerId; + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/StaleEtlJobLeaseException.java b/etl-service/src/main/java/com/xtrmetl/etl/job/StaleEtlJobLeaseException.java new file mode 100644 index 00000000..d7320870 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/StaleEtlJobLeaseException.java @@ -0,0 +1,17 @@ +package com.xtrmetl.etl.job; + +/** + * Signals that a worker no longer owns the exact live lease required for a state transition. + * + *

The exception intentionally carries no job, claim, owner, payload, SQL, or timestamp values so + * accidental logging cannot disclose operational identifiers or retained customer data.

+ */ +public class StaleEtlJobLeaseException extends RuntimeException { + + /** + * Creates the stable non-sensitive stale-lease signal. + */ + public StaleEtlJobLeaseException() { + super("The durable ETL job lease is stale or no longer owned"); + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/service/EtlService.java b/etl-service/src/main/java/com/xtrmetl/etl/service/EtlService.java index 6bb32a15..28b86bb2 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/service/EtlService.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/service/EtlService.java @@ -38,6 +38,11 @@ * batch back rather than leaving committed prefix records. The service intentionally avoids the * JVM common pool and one-task-per-record fan-out.

* + *

Synchronous request entry points own Spring Retry outside their transaction boundaries. A + * durable worker instead uses {@link #processDataInExistingTransaction(String)} exactly once per + * persisted attempt so the lease, target rows, response ledger, and terminal state remain in one + * database transaction and retry accounting stays in the durable job record.

+ * *

Callers may optionally use {@link #processDataIdempotently(String, String, String)}. That * method attempts a principal-scoped transaction lock without waiting, replays a prior successful * response, and commits the ETL rows and durable ledger entry in the same transaction.

@@ -142,8 +147,8 @@ public EtlService( /** * Processes one JSON-array request as a prevalidated transaction-scoped batch. * - *

Only transient Spring data-access failures are retried. Typed input failures and - * deterministic target constraints fail immediately instead of repeating the same work.

+ *

Only transient Spring data-access failures are retried. Retry advice wraps the transaction + * advice so every synchronous request attempt receives a fresh transaction.

* * @param data UTF-8 JSON array payload * @return one {@code Processed: } line per record, in input order @@ -160,6 +165,26 @@ public String processData(@Nullable String data) { return processDataInCurrentTransaction(data); } + /** + * Processes one durable-job payload exactly once inside the caller's existing transaction. + * + *

This method deliberately has neither {@link Retryable} nor {@link Transactional}. The + * durable worker owns its persisted retry count and supplies the transaction that also contains + * its lease-fenced terminal transition and response-ledger write. An in-process retry inside + * that outer transaction could reuse an already failed transaction and would not represent a + * new durable attempt.

+ * + * @param data retained UTF-8 JSON array payload + * @return one {@code Processed: } line per record, in input order + * @throws IllegalStateException when the caller did not establish an actual transaction + * @throws EtlRequestException when the retained request violates a deterministic contract + * @throws org.springframework.dao.DataAccessException when the target database rejects work + */ + public String processDataInExistingTransaction(@Nullable String data) { + requireActiveTransaction("Durable ETL execution requires an active transaction"); + return processDataInCurrentTransaction(data); + } + /** * Processes or replays one principal-scoped idempotent ETL request. * @@ -200,7 +225,7 @@ public EtlIdempotencyResult processDataIdempotently( if (data == null) { throw new EtlRequestException(EtlRequestError.INVALID_JSON); } - requireActiveTransaction(); + requireActiveTransaction("Idempotent ETL processing requires an active transaction"); enforcePayloadLimit(data); String idempotencyKeyHash = sha256( @@ -288,11 +313,9 @@ private static String validatePrincipalScope(@Nullable String principalScope) { return principalScope; } - private static void requireActiveTransaction() { + private static void requireActiveTransaction(String failureMessage) { if (!TransactionSynchronizationManager.isActualTransactionActive()) { - throw new IllegalStateException( - "Idempotent ETL processing requires an active transaction" - ); + throw new IllegalStateException(failureMessage); } } diff --git a/etl-service/src/main/resources/application.properties b/etl-service/src/main/resources/application.properties new file mode 100644 index 00000000..918dd4ad --- /dev/null +++ b/etl-service/src/main/resources/application.properties @@ -0,0 +1,2 @@ +# PostgreSQL concurrent index migrations cannot use Flyway's transactional advisory lock. +spring.flyway.postgresql.transactional-lock=false diff --git a/etl-service/src/main/resources/db/migration/V3__add_etl_job_lease_fencing.sql b/etl-service/src/main/resources/db/migration/V3__add_etl_job_lease_fencing.sql new file mode 100644 index 00000000..2469acc7 --- /dev/null +++ b/etl-service/src/main/resources/db/migration/V3__add_etl_job_lease_fencing.sql @@ -0,0 +1,43 @@ +ALTER TABLE etl_job_records + ADD COLUMN lease_claim_id UUID, + ADD COLUMN lease_owner_id VARCHAR(128), + ADD COLUMN lease_expires_at TIMESTAMPTZ; + +-- Repair legacy rows before enforcing the stronger failure lifecycle invariant. +UPDATE etl_job_records +SET failure_code = 'etl_legacy_failure' +WHERE job_status = 'FAILED' + AND failure_code IS NULL; + +UPDATE etl_job_records +SET failure_code = NULL +WHERE job_status <> 'FAILED' + AND failure_code IS NOT NULL; + +ALTER TABLE etl_job_records + 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 + ) + ); diff --git a/etl-service/src/main/resources/db/migration/V4__add_etl_job_claim_eligibility_index.sql b/etl-service/src/main/resources/db/migration/V4__add_etl_job_claim_eligibility_index.sql new file mode 100644 index 00000000..a7939f34 --- /dev/null +++ b/etl-service/src/main/resources/db/migration/V4__add_etl_job_claim_eligibility_index.sql @@ -0,0 +1,11 @@ +-- Support oldest-first claim selection across pending and expired-running durable jobs. +-- CONCURRENTLY preserves inserts, updates, and deletes while PostgreSQL builds the index. +-- The companion .sql.conf disables Flyway's per-migration transaction because PostgreSQL +-- rejects CREATE INDEX CONCURRENTLY inside a transaction block. +CREATE INDEX CONCURRENTLY etl_job_claim_eligibility_index + ON etl_job_records ( + job_status, + lease_expires_at, + created_at, + job_record_id + ); diff --git a/etl-service/src/main/resources/db/migration/V4__add_etl_job_claim_eligibility_index.sql.conf b/etl-service/src/main/resources/db/migration/V4__add_etl_job_claim_eligibility_index.sql.conf new file mode 100644 index 00000000..73bd53a1 --- /dev/null +++ b/etl-service/src/main/resources/db/migration/V4__add_etl_job_claim_eligibility_index.sql.conf @@ -0,0 +1 @@ +executeInTransaction=false diff --git a/etl-service/src/test/java/com/xtrmetl/etl/config/MightyEtlConfigAliasEnvironmentPostProcessorTest.java b/etl-service/src/test/java/com/xtrmetl/etl/config/MightyEtlConfigAliasEnvironmentPostProcessorTest.java index 285b24d7..389dda7f 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/config/MightyEtlConfigAliasEnvironmentPostProcessorTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/config/MightyEtlConfigAliasEnvironmentPostProcessorTest.java @@ -45,6 +45,38 @@ void mirrorsModernDurableIntakeFlagToTheLegacyControllerCondition() { assertEquals("true", aliases.get("xtrmetl.etl.jobs.intake-enabled")); } + @Test + void mirrorsEveryModernDurableWorkerSettingToLegacyConsumers() { + MockEnvironment env = new MockEnvironment(); + env.setProperty("mightyetl.etl.jobs.worker.enabled", "true"); + env.setProperty("mightyetl.etl.jobs.worker.fixed-delay-milliseconds", "2500"); + env.setProperty("mightyetl.etl.jobs.worker.initial-delay-milliseconds", "1000"); + env.setProperty("mightyetl.etl.jobs.worker.lease-duration-seconds", "120"); + env.setProperty("mightyetl.etl.jobs.worker.max-attempts", "5"); + env.setProperty("mightyetl.etl.jobs.worker.lease-owner-id", "worker-primary"); + + Map aliases = MightyEtlConfigAliasEnvironmentPostProcessor.buildAliases(env); + + assertEquals("true", aliases.get("xtrmetl.etl.jobs.worker.enabled")); + assertEquals( + "2500", + aliases.get("xtrmetl.etl.jobs.worker.fixed-delay-milliseconds") + ); + assertEquals( + "1000", + aliases.get("xtrmetl.etl.jobs.worker.initial-delay-milliseconds") + ); + assertEquals( + "120", + aliases.get("xtrmetl.etl.jobs.worker.lease-duration-seconds") + ); + assertEquals("5", aliases.get("xtrmetl.etl.jobs.worker.max-attempts")); + assertEquals( + "worker-primary", + aliases.get("xtrmetl.etl.jobs.worker.lease-owner-id") + ); + } + @Test void mirrorsLegacyBatchLimitForModernTooling() { MockEnvironment env = new MockEnvironment(); diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobClaimIndexMigrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobClaimIndexMigrationTest.java new file mode 100644 index 00000000..cb09c0fd --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobClaimIndexMigrationTest.java @@ -0,0 +1,124 @@ +package com.xtrmetl.etl.job; + +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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Guards the production rollout contract for the durable-job claim eligibility index. + * + *

The table can already contain accepted work when lease fencing is introduced. PostgreSQL's + * regular index build blocks inserts, updates, and deletes, so the claim index must be isolated in + * a non-transactional concurrent migration while the lease columns and constraints remain in the + * transactional V3 migration.

+ */ +class EtlJobClaimIndexMigrationTest { + + private static final String V3_MIGRATION = + "etl-service/src/main/resources/db/migration/V3__add_etl_job_lease_fencing.sql"; + private static final String V4_MIGRATION = + "etl-service/src/main/resources/db/migration/V4__add_etl_job_claim_eligibility_index.sql"; + private static final String V4_CONFIGURATION = V4_MIGRATION + ".conf"; + private static final String APPLICATION_PROPERTIES = + "etl-service/src/main/resources/application.properties"; + private static final String ROLLOUT_RUNBOOK = + "docs/operations/durable-job-claim-index-rollout.md"; + + @Test + void keepsTransactionalLeaseSchemaSeparateFromConcurrentIndexBuild() throws IOException { + String leaseMigration = normalize(read(V3_MIGRATION)); + String indexMigration = normalize(read(V4_MIGRATION)); + + assertFalse( + leaseMigration.contains("CREATE INDEX"), + "the transactional lease migration must not contain a production index build" + ); + assertTrue(indexMigration.contains( + "CREATE INDEX CONCURRENTLY etl_job_claim_eligibility_index" + )); + assertTrue(indexMigration.contains( + "ON etl_job_records ( job_status, lease_expires_at, created_at, job_record_id )" + )); + } + + @Test + void disablesFlywayTransactionsAndPostgresqlTransactionalLocksForConcurrentDdl() + throws IOException { + Path configurationPath = projectRoot().resolve(V4_CONFIGURATION); + Path applicationPropertiesPath = projectRoot().resolve(APPLICATION_PROPERTIES); + + assertTrue( + Files.exists(configurationPath), + "the concurrent migration requires a matching Flyway script configuration" + ); + assertTrue( + Files.exists(applicationPropertiesPath), + "concurrent PostgreSQL Flyway DDL requires explicit non-transactional locking config" + ); + assertTrue( + Files.readString(configurationPath, StandardCharsets.UTF_8) + .contains("executeInTransaction=false") + ); + assertTrue( + Files.readString(applicationPropertiesPath, StandardCharsets.UTF_8).contains( + "spring.flyway.postgresql.transactional-lock=false" + ) + ); + } + + @Test + void runbookDocumentsConcurrentFailureRecoveryAndRollback() throws IOException { + Path runbookPath = projectRoot().resolve(ROLLOUT_RUNBOOK); + assertTrue( + Files.exists(runbookPath), + "concurrent index rollout requires an operator recovery and rollback runbook" + ); + String runbook = normalize(Files.readString(runbookPath, StandardCharsets.UTF_8)); + + assertTrue(runbook.contains("V4__add_etl_job_claim_eligibility_index.sql")); + assertTrue(runbook.contains("CREATE INDEX CONCURRENTLY")); + assertTrue(runbook.contains("invalid index")); + assertTrue(runbook.contains( + "DROP INDEX CONCURRENTLY etl_job_claim_eligibility_index" + )); + assertTrue(runbook.contains("executeInTransaction=false")); + } + + private static String read(String relativePath) throws IOException { + return Files.readString(projectRoot().resolve(relativePath), StandardCharsets.UTF_8); + } + + private static String normalize(String value) { + return value.replaceAll("\\s+", " ").trim(); + } + + /** + * Finds the reactor root from either repository-root or module-local Maven execution. + * + * @return absolute repository root containing the source under test + */ + 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/EtlJobExecutionServiceIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobExecutionServiceIntegrationTest.java new file mode 100644 index 00000000..9fd231e1 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobExecutionServiceIntegrationTest.java @@ -0,0 +1,303 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestLock; +import com.xtrmetl.etl.service.EtlService; +import com.xtrmetl.etl.service.Sha256Digest; +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 java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Proves that ledger, target writes, and exact-live-lease success commit or roll back together. + */ +@SpringJUnitConfig(EtlJobExecutionServiceIntegrationTest.TestConfiguration.class) +class EtlJobExecutionServiceIntegrationTest { + + private static final String OWNER_ID = "worker-alpha"; + private static final String PAYLOAD = """ + [{"id":"record_alpha","name":"accepted","email":"USER@EXAMPLE.COM"}] + """; + + private final EtlJobExecutionService executionService; + private final EtlJobLeaseRepository leaseRepository; + private final EtlJobIdempotencyService idempotencyService; + private final JdbcTemplate jdbcTemplate; + + @Autowired + EtlJobExecutionServiceIntegrationTest( + EtlJobExecutionService executionService, + EtlJobLeaseRepository leaseRepository, + EtlJobIdempotencyService idempotencyService, + JdbcTemplate jdbcTemplate + ) { + this.executionService = executionService; + this.leaseRepository = leaseRepository; + this.idempotencyService = idempotencyService; + this.jdbcTemplate = jdbcTemplate; + } + + @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, + 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 commitsLedgerTargetRowsAndTerminalSuccessInOneTransaction() { + UUID jobRecordId = insertPendingJob(); + EtlJobLease lease = leaseRepository.claimNext( + OWNER_ID, + Duration.ofMinutes(5), + 3 + ).orElseThrow(); + + executionService.execute(lease); + + assertEquals(jobRecordId, lease.jobRecordId()); + assertEquals(1, tableCount("processed_data")); + assertEquals(1, tableCount("etl_idempotency_records")); + assertEquals( + "ID:record_alpha,NAME:ACCEPTED,EMAIL:user@example.com,", + jdbcTemplate.queryForObject("SELECT data FROM processed_data", String.class) + ); + assertEquals("SUCCEEDED", jobStatus(jobRecordId)); + assertEquals(0, retainedPayloadCount(jobRecordId)); + } + + @Test + void rollsBackLedgerAndTargetRowsWhenTheClaimWasSuperseded() { + UUID jobRecordId = insertPendingJob(); + EtlJobLease lease = leaseRepository.claimNext( + OWNER_ID, + Duration.ofMinutes(5), + 3 + ).orElseThrow(); + jdbcTemplate.update( + "UPDATE etl_job_records SET lease_claim_id = ? WHERE job_record_id = ?", + UUID.randomUUID(), + jobRecordId + ); + + assertThrows(StaleEtlJobLeaseException.class, () -> executionService.execute(lease)); + + assertUncommittedDurableEffects(jobRecordId); + } + + @Test + void rollsBackLedgerAndTargetRowsWhenTheLeaseExpiredBeforeExecution() { + UUID jobRecordId = insertPendingJob(); + EtlJobLease lease = leaseRepository.claimNext( + OWNER_ID, + Duration.ofMinutes(5), + 3 + ).orElseThrow(); + jdbcTemplate.update( + "UPDATE etl_job_records SET lease_expires_at = ? WHERE job_record_id = ?", + Instant.now().minusSeconds(1), + jobRecordId + ); + + assertThrows(StaleEtlJobLeaseException.class, () -> executionService.execute(lease)); + + assertUncommittedDurableEffects(jobRecordId); + } + + @Test + void rejectsMissingCollaboratorsOrLease() { + assertThrows( + NullPointerException.class, + () -> new EtlJobExecutionService(null, leaseRepository) + ); + assertThrows( + NullPointerException.class, + () -> new EtlJobExecutionService(idempotencyService, null) + ); + assertThrows(NullPointerException.class, () -> executionService.execute(null)); + } + + private void assertUncommittedDurableEffects(UUID jobRecordId) { + assertEquals(0, tableCount("processed_data")); + assertEquals(0, tableCount("etl_idempotency_records")); + assertEquals("RUNNING", jobStatus(jobRecordId)); + assertEquals(1, retainedPayloadCount(jobRecordId)); + } + + private UUID insertPendingJob() { + UUID jobRecordId = UUID.randomUUID(); + Instant now = Instant.now(); + 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, + "a".repeat(64), + "b".repeat(64), + Sha256Digest.digest(PAYLOAD), + 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 jobStatus(UUID jobRecordId) { + return jdbcTemplate.queryForObject( + "SELECT job_status FROM etl_job_records WHERE job_record_id = ?", + String.class, + jobRecordId + ); + } + + private int retainedPayloadCount(UUID jobRecordId) { + Integer count = jdbcTemplate.queryForObject( + """ + SELECT COUNT(*) + FROM etl_job_records + WHERE job_record_id = ? + AND request_payload IS NOT NULL + """, + Integer.class, + jobRecordId + ); + return count == null ? 0 : count; + } + + /** + * Transaction-enabled execution context using one database for job, ledger, and target effects. + */ + @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 + EtlBatchProperties etlBatchProperties() { + return new EtlBatchProperties(); + } + + @Bean + ObjectMapper objectMapper() { + return new ObjectMapper(); + } + + @Bean + EtlRequestLock etlRequestLock() { + return idempotencyKeyHash -> true; + } + + @Bean + EtlService etlService( + JdbcTemplate jdbcTemplate, + ObjectMapper objectMapper, + EtlBatchProperties properties, + EtlRequestLock requestLock + ) { + return new EtlService(jdbcTemplate, objectMapper, properties, requestLock); + } + + @Bean + EtlJobIdempotencyService etlJobIdempotencyService( + JdbcTemplate jdbcTemplate, + EtlService etlService, + EtlRequestLock requestLock + ) { + return new EtlJobIdempotencyService(jdbcTemplate, etlService, requestLock); + } + + @Bean + EtlJobLeaseRepository etlJobLeaseRepository( + JdbcTemplate jdbcTemplate, + PlatformTransactionManager transactionManager + ) { + return new EtlJobLeaseRepository(jdbcTemplate, transactionManager); + } + + @Bean + EtlJobExecutionService etlJobExecutionService( + EtlJobIdempotencyService idempotencyService, + EtlJobLeaseRepository leaseRepository + ) { + return new EtlJobExecutionService(idempotencyService, leaseRepository); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyRetryBoundaryTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyRetryBoundaryTest.java new file mode 100644 index 00000000..cc266ede --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyRetryBoundaryTest.java @@ -0,0 +1,100 @@ +package com.xtrmetl.etl.job; + +import com.xtrmetl.etl.service.EtlRequestLock; +import com.xtrmetl.etl.service.EtlService; +import com.xtrmetl.etl.service.Sha256Digest; +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.transaction.support.TransactionTemplate; + +import java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Guards the retry boundary between durable database attempts and synchronous request retries. + * + *

A durable worker already owns bounded, persisted retries through {@code attempt_count}. It + * must therefore invoke an ETL entry point that joins the current lease transaction exactly once, + * rather than the synchronous {@code @Retryable} API whose advice is designed to wrap a fresh + * transaction for each HTTP-request attempt.

+ */ +class EtlJobIdempotencyRetryBoundaryTest { + + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + private static final String RESPONSE = "Processed: record_alpha"; + + /** + * Proves one durable attempt uses only the non-retrying current-transaction ETL entry point. + */ + @Test + void durableAttemptDoesNotInvokeTheSynchronousRetryableEntryPoint() { + EmbeddedDatabase database = new EmbeddedDatabaseBuilder() + .generateUniqueName(true) + .setType(EmbeddedDatabaseType.H2) + .build(); + try { + JdbcTemplate jdbcTemplate = new JdbcTemplate(database); + 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 + ) + """); + + EtlService etlService = mock(EtlService.class); + EtlRequestLock requestLock = mock(EtlRequestLock.class); + when(requestLock.tryLock(anyString())).thenReturn(true); + when(etlService.processDataInExistingTransaction(PAYLOAD)).thenReturn(RESPONSE); + EtlJobIdempotencyService service = new EtlJobIdempotencyService( + jdbcTemplate, + etlService, + requestLock + ); + TransactionTemplate transactionTemplate = new TransactionTemplate( + new DataSourceTransactionManager(database) + ); + + String result = transactionTemplate.execute(status -> service.process(lease())); + + assertEquals(RESPONSE, result); + verify(etlService).processDataInExistingTransaction(PAYLOAD); + verify(etlService, never()).processData(anyString()); + assertEquals( + 1, + jdbcTemplate.queryForObject( + "SELECT COUNT(*) FROM etl_idempotency_records", + Integer.class + ) + ); + } finally { + database.shutdown(); + } + } + + private static EtlJobLease lease() { + return new EtlJobLease( + UUID.randomUUID(), + UUID.randomUUID(), + "worker-alpha", + "a".repeat(64), + "b".repeat(64), + Sha256Digest.digest(PAYLOAD), + PAYLOAD, + 1, + Instant.now().plusSeconds(300) + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyServiceIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyServiceIntegrationTest.java new file mode 100644 index 00000000..adeb0867 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyServiceIntegrationTest.java @@ -0,0 +1,270 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestLock; +import com.xtrmetl.etl.service.EtlService; +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.dao.CannotAcquireLockException; +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.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.Instant; +import java.util.HexFormat; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Proves that durable jobs reuse the response ledger without retaining raw identity values. + */ +@SpringJUnitConfig(EtlJobIdempotencyServiceIntegrationTest.TestConfiguration.class) +class EtlJobIdempotencyServiceIntegrationTest { + + private static final String OWNER_ID = "worker-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 PAYLOAD = """ + [{"id":"record_alpha","name":"accepted","email":"USER@EXAMPLE.COM"}] + """; + + private final EtlJobIdempotencyService idempotencyService; + private final EtlService etlService; + private final EtlRequestLock requestLock; + private final JdbcTemplate jdbcTemplate; + + @Autowired + EtlJobIdempotencyServiceIntegrationTest( + EtlJobIdempotencyService idempotencyService, + EtlService etlService, + EtlRequestLock requestLock, + JdbcTemplate jdbcTemplate + ) { + this.idempotencyService = idempotencyService; + this.etlService = etlService; + this.requestLock = requestLock; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createTables() { + reset(requestLock); + when(requestLock.tryLock(anyString())).thenReturn(true); + jdbcTemplate.execute("DROP TABLE IF EXISTS processed_data"); + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_idempotency_records"); + 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 + @Transactional + void writesTargetAndLedgerThenReplaysWithoutASecondTargetWrite() { + EtlJobLease lease = lease(PAYLOAD, sha256(PAYLOAD)); + + String firstResponse = idempotencyService.process(lease); + String replayedResponse = idempotencyService.process(lease); + + assertEquals("Processed: record_alpha", firstResponse); + assertEquals(firstResponse, replayedResponse); + assertEquals(1, count("processed_data")); + assertEquals(1, count("etl_idempotency_records")); + verify(requestLock, org.mockito.Mockito.times(2)).tryLock(anyString()); + } + + @Test + @Transactional + void rejectsPayloadDigestMismatchBeforeLockOrWrites() { + EtlJobLease lease = lease(PAYLOAD, "c".repeat(64)); + + EtlJobIntegrityException exception = assertThrows( + EtlJobIntegrityException.class, + () -> idempotencyService.process(lease) + ); + + assertEquals("etl_job_integrity_failure", exception.failureCode()); + verify(requestLock, never()).tryLock(anyString()); + assertEquals(0, count("processed_data")); + assertEquals(0, count("etl_idempotency_records")); + } + + @Test + @Transactional + void rejectsConflictingStoredDigestWithoutAnotherTargetWrite() { + EtlJobLease lease = lease(PAYLOAD, sha256(PAYLOAD)); + idempotencyService.process(lease); + jdbcTemplate.update( + "UPDATE etl_idempotency_records SET request_digest = ?", + "d".repeat(64) + ); + + assertThrows(EtlJobIntegrityException.class, () -> idempotencyService.process(lease)); + + assertEquals(1, count("processed_data")); + assertEquals(1, count("etl_idempotency_records")); + } + + @Test + @Transactional + void reportsBusyLedgerAsTransientWithoutWrites() { + when(requestLock.tryLock(anyString())).thenReturn(false); + EtlJobLease lease = lease(PAYLOAD, sha256(PAYLOAD)); + + assertThrows(CannotAcquireLockException.class, () -> idempotencyService.process(lease)); + + assertEquals(0, count("processed_data")); + assertEquals(0, count("etl_idempotency_records")); + } + + @Test + void failsClosedWithoutARealTransaction() { + EtlJobIdempotencyService directService = new EtlJobIdempotencyService( + jdbcTemplate, + etlService, + requestLock + ); + + assertThrows(IllegalStateException.class, () -> directService.process( + lease(PAYLOAD, sha256(PAYLOAD)) + )); + verify(requestLock, never()).tryLock(anyString()); + } + + @Test + void rejectsMissingCollaboratorsAndLease() { + assertThrows( + NullPointerException.class, + () -> new EtlJobIdempotencyService(null, etlService, requestLock) + ); + assertThrows( + NullPointerException.class, + () -> new EtlJobIdempotencyService(jdbcTemplate, null, requestLock) + ); + assertThrows( + NullPointerException.class, + () -> new EtlJobIdempotencyService(jdbcTemplate, etlService, null) + ); + assertThrows(NullPointerException.class, () -> idempotencyService.process(null)); + } + + private int count(String tableName) { + Integer count = jdbcTemplate.queryForObject( + "SELECT COUNT(*) FROM " + tableName, + Integer.class + ); + return count == null ? 0 : count; + } + + private static EtlJobLease lease(String payload, String requestDigest) { + return new EtlJobLease( + UUID.randomUUID(), + UUID.randomUUID(), + OWNER_ID, + PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, + requestDigest, + payload, + 1, + Instant.now().plusSeconds(300) + ); + } + + private static String sha256(String value) { + try { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + return HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8))); + } catch (NoSuchAlgorithmException exception) { + throw new AssertionError(exception); + } + } + + /** Transaction-enabled service context backed by an isolated H2 database. */ + @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 + EtlBatchProperties etlBatchProperties() { + return new EtlBatchProperties(); + } + + @Bean + ObjectMapper objectMapper() { + return new ObjectMapper(); + } + + @Bean + EtlRequestLock etlRequestLock() { + return mock(EtlRequestLock.class); + } + + @Bean + EtlService etlService( + JdbcTemplate jdbcTemplate, + ObjectMapper objectMapper, + EtlBatchProperties properties, + EtlRequestLock requestLock + ) { + return new EtlService(jdbcTemplate, objectMapper, properties, requestLock); + } + + @Bean + EtlJobIdempotencyService etlJobIdempotencyService( + JdbcTemplate jdbcTemplate, + EtlService etlService, + EtlRequestLock requestLock + ) { + return new EtlJobIdempotencyService(jdbcTemplate, etlService, requestLock); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyTransactionAnnotationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyTransactionAnnotationTest.java new file mode 100644 index 00000000..42fe752a --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobIdempotencyTransactionAnnotationTest.java @@ -0,0 +1,36 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.Test; +import org.springframework.transaction.annotation.Transactional; + +import java.lang.reflect.Method; + +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * Guards the durable response-ledger service from creating its own execution transaction. + * + *

The lease-fenced execution service owns the transaction that contains target effects, the + * response ledger, and terminal success. A transactional annotation on the public ledger method + * would let a direct Spring-proxy caller commit target and ledger effects without the success fence.

+ */ +class EtlJobIdempotencyTransactionAnnotationTest { + + /** + * Requires the public processing method to join, rather than create, the caller transaction. + * + * @throws NoSuchMethodException when the public durable processing contract is missing + */ + @Test + void processDoesNotCreateAStandaloneTransaction() throws NoSuchMethodException { + Method processMethod = EtlJobIdempotencyService.class.getMethod( + "process", + EtlJobLease.class + ); + + assertNull( + processMethod.getAnnotation(Transactional.class), + "Durable job idempotency must not own a transaction" + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseMigrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseMigrationTest.java new file mode 100644 index 00000000..6491616c --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseMigrationTest.java @@ -0,0 +1,87 @@ +package com.xtrmetl.etl.job; + +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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Specifies the transactional Flyway contract for exact durable-job lease fencing. + */ +class EtlJobLeaseMigrationTest { + + @Test + void addsDescriptiveLeaseColumnsWithoutNonTransactionalIndexDdl() throws IOException { + String migration = readMigration(); + + assertTrue(migration.contains("ADD COLUMN lease_claim_id UUID")); + assertTrue(migration.contains("ADD COLUMN lease_owner_id VARCHAR(128)")); + assertTrue(migration.contains("ADD COLUMN lease_expires_at TIMESTAMPTZ")); + assertFalse(migration.contains("CREATE INDEX")); + assertFalse(migration.contains(" ADD COLUMN owner ")); + assertFalse(migration.contains(" ADD COLUMN lease ")); + } + + @Test + void requiresLeaseFieldsOnlyForRunningRows() throws IOException { + String migration = normalize(readMigration()); + + assertTrue(migration.contains("CONSTRAINT etl_job_lease_lifecycle_check")); + assertTrue(migration.contains( + "job_status = 'RUNNING' AND lease_claim_id IS NOT NULL AND lease_owner_id IS NOT NULL AND lease_expires_at IS NOT NULL" + )); + assertTrue(migration.contains( + "job_status <> 'RUNNING' AND lease_claim_id IS NULL AND lease_owner_id IS NULL AND lease_expires_at IS NULL" + )); + } + + @Test + void requiresFailureCodesOnlyForFailedRows() throws IOException { + String migration = normalize(readMigration()); + + assertTrue(migration.contains("CONSTRAINT etl_job_failure_lifecycle_check")); + assertTrue(migration.contains("job_status = 'FAILED' AND failure_code IS NOT NULL")); + assertTrue(migration.contains("job_status <> 'FAILED' AND failure_code IS NULL")); + } + + private static String readMigration() throws IOException { + return Files.readString( + projectRoot().resolve( + "etl-service/src/main/resources/db/migration/" + + "V3__add_etl_job_lease_fencing.sql" + ), + StandardCharsets.UTF_8 + ); + } + + private static String normalize(String value) { + return value.replaceAll("\\s+", " ").trim(); + } + + /** + * Finds the reactor root from either repository-root or module-local Maven execution. + */ + 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/EtlJobLeaseModelTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseModelTest.java new file mode 100644 index 00000000..b221d95c --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseModelTest.java @@ -0,0 +1,147 @@ +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.assertThrows; + +/** + * Specifies the immutable value contract carried from a database claim into execution. + */ +class EtlJobLeaseModelTest { + + private static final UUID JOB_RECORD_ID = UUID.randomUUID(); + private static final UUID LEASE_CLAIM_ID = UUID.randomUUID(); + private static final String OWNER_ID = "worker-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 static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + private static final Instant EXPIRY = Instant.parse("2026-08-05T01:00:00Z"); + + @Test + void retainsEveryValidatedClaimField() { + EtlJobLease lease = validLease(); + + assertEquals(JOB_RECORD_ID, lease.jobRecordId()); + assertEquals(LEASE_CLAIM_ID, lease.leaseClaimId()); + assertEquals(OWNER_ID, lease.leaseOwnerId()); + assertEquals(PRINCIPAL_SCOPE_HASH, lease.principalScopeHash()); + assertEquals(SUBMISSION_KEY_HASH, lease.submissionKeyHash()); + assertEquals(REQUEST_DIGEST, lease.requestDigest()); + assertEquals(PAYLOAD, lease.requestPayload()); + assertEquals(2, lease.attemptCount()); + assertEquals(EXPIRY, lease.leaseExpiresAt()); + } + + @Test + void rejectsMissingUnsafeOrImpossibleFields() { + assertThrows( + NullPointerException.class, + () -> lease(null, LEASE_CLAIM_ID, OWNER_ID, PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, REQUEST_DIGEST, PAYLOAD, 1, EXPIRY) + ); + assertThrows( + NullPointerException.class, + () -> lease(JOB_RECORD_ID, null, OWNER_ID, PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, REQUEST_DIGEST, PAYLOAD, 1, EXPIRY) + ); + assertThrows( + NullPointerException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, null, PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, REQUEST_DIGEST, PAYLOAD, 1, EXPIRY) + ); + assertThrows( + IllegalArgumentException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, "unsafe owner", + PRINCIPAL_SCOPE_HASH, SUBMISSION_KEY_HASH, REQUEST_DIGEST, + PAYLOAD, 1, EXPIRY) + ); + assertThrows( + NullPointerException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, null, + SUBMISSION_KEY_HASH, REQUEST_DIGEST, PAYLOAD, 1, EXPIRY) + ); + assertThrows( + IllegalArgumentException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, "A".repeat(64), + SUBMISSION_KEY_HASH, REQUEST_DIGEST, PAYLOAD, 1, EXPIRY) + ); + assertThrows( + NullPointerException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, PRINCIPAL_SCOPE_HASH, + null, REQUEST_DIGEST, PAYLOAD, 1, EXPIRY) + ); + assertThrows( + IllegalArgumentException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, PRINCIPAL_SCOPE_HASH, + "short", REQUEST_DIGEST, PAYLOAD, 1, EXPIRY) + ); + assertThrows( + NullPointerException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, null, PAYLOAD, 1, EXPIRY) + ); + assertThrows( + IllegalArgumentException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, "g".repeat(64), PAYLOAD, 1, EXPIRY) + ); + assertThrows( + NullPointerException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, REQUEST_DIGEST, null, 1, EXPIRY) + ); + assertThrows( + IllegalArgumentException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, REQUEST_DIGEST, PAYLOAD, 0, EXPIRY) + ); + assertThrows( + NullPointerException.class, + () -> lease(JOB_RECORD_ID, LEASE_CLAIM_ID, OWNER_ID, PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, REQUEST_DIGEST, PAYLOAD, 1, null) + ); + } + + private static EtlJobLease validLease() { + return lease( + JOB_RECORD_ID, + LEASE_CLAIM_ID, + OWNER_ID, + PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, + REQUEST_DIGEST, + PAYLOAD, + 2, + EXPIRY + ); + } + + private static EtlJobLease lease( + UUID jobRecordId, + UUID leaseClaimId, + String leaseOwnerId, + String principalScopeHash, + String submissionKeyHash, + String requestDigest, + String requestPayload, + int attemptCount, + Instant leaseExpiresAt + ) { + return new EtlJobLease( + jobRecordId, + leaseClaimId, + leaseOwnerId, + principalScopeHash, + submissionKeyHash, + requestDigest, + requestPayload, + attemptCount, + leaseExpiresAt + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseRepositoryIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseRepositoryIntegrationTest.java new file mode 100644 index 00000000..231d2340 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseRepositoryIntegrationTest.java @@ -0,0 +1,404 @@ +package com.xtrmetl.etl.job; + +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 org.springframework.transaction.annotation.Transactional; + +import javax.sql.DataSource; +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import static org.junit.jupiter.api.Assertions.assertEquals; +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; + +/** + * Proves exclusive durable-job claims and exact-live-lease state transitions against SQL. + */ +@SpringJUnitConfig(EtlJobLeaseRepositoryIntegrationTest.TestConfiguration.class) +class EtlJobLeaseRepositoryIntegrationTest { + + private static final Duration LEASE_DURATION = Duration.ofMinutes(5); + private static final String OWNER_ALPHA = "worker-alpha"; + private static final String OWNER_BETA = "worker-beta"; + private static final String PRINCIPAL_SCOPE_HASH = "a".repeat(64); + private static final String REQUEST_DIGEST = "b".repeat(64); + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + + private final EtlJobLeaseRepository repository; + private final JdbcTemplate jdbcTemplate; + private ExecutorService executorService; + + @Autowired + EtlJobLeaseRepositoryIntegrationTest( + EtlJobLeaseRepository repository, + JdbcTemplate jdbcTemplate + ) { + this.repository = repository; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createJobTable() { + executorService = Executors.newFixedThreadPool(2); + 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, + created_at TIMESTAMP WITH TIME ZONE NOT NULL, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL + ) + """); + } + + @AfterEach + void closeExecutor() { + executorService.close(); + } + + @Test + void claimsTheOldestEligibleJobAndIncrementsItsAttempt() { + UUID newerJobId = insertPending(Instant.parse("2026-08-05T00:01:00Z"), 0); + UUID olderJobId = insertPending(Instant.parse("2026-08-05T00:00:00Z"), 1); + + EtlJobLease lease = repository.claimNext(OWNER_ALPHA, LEASE_DURATION, 3).orElseThrow(); + + assertEquals(olderJobId, lease.jobRecordId()); + assertEquals(OWNER_ALPHA, lease.leaseOwnerId()); + assertEquals(PRINCIPAL_SCOPE_HASH, lease.principalScopeHash()); + assertTrue(lease.submissionKeyHash().matches("[0-9a-f]{64}")); + assertEquals(REQUEST_DIGEST, lease.requestDigest()); + assertEquals(PAYLOAD, lease.requestPayload()); + assertEquals(2, lease.attemptCount()); + assertNotNull(lease.leaseClaimId()); + assertTrue(lease.leaseExpiresAt().isAfter(Instant.now())); + assertEquals("RUNNING", textColumn(olderJobId, "job_status")); + assertEquals(2, integerColumn(olderJobId, "attempt_count")); + assertEquals("PENDING", textColumn(newerJobId, "job_status")); + } + + @Test + void skipsLiveClaimsAndReclaimsExpiredClaimsWithFreshFencing() { + UUID liveJobId = insertRunning( + Instant.parse("2026-08-05T00:00:00Z"), + 1, + OWNER_ALPHA, + UUID.randomUUID(), + Instant.now().plusSeconds(300) + ); + UUID priorClaimId = UUID.randomUUID(); + UUID expiredJobId = insertRunning( + Instant.parse("2026-08-05T00:01:00Z"), + 1, + OWNER_ALPHA, + priorClaimId, + Instant.now().minusSeconds(60) + ); + + EtlJobLease reclaimed = repository.claimNext(OWNER_BETA, LEASE_DURATION, 3).orElseThrow(); + + assertEquals(expiredJobId, reclaimed.jobRecordId()); + assertEquals(OWNER_BETA, reclaimed.leaseOwnerId()); + assertNotEquals(priorClaimId, reclaimed.leaseClaimId()); + assertEquals(2, reclaimed.attemptCount()); + assertEquals("RUNNING", textColumn(liveJobId, "job_status")); + assertEquals(OWNER_ALPHA, textColumn(liveJobId, "lease_owner_id")); + } + + @Test + void returnsEmptyWhenEveryClaimIsLive() { + insertRunning( + Instant.now(), + 1, + OWNER_ALPHA, + UUID.randomUUID(), + Instant.now().plusSeconds(300) + ); + + Optional lease = repository.claimNext(OWNER_BETA, LEASE_DURATION, 3); + + assertTrue(lease.isEmpty()); + } + + @Test + void onlyOneConcurrentWorkerCanClaimOnePendingJob() throws Exception { + UUID jobRecordId = insertPending(Instant.now(), 0); + CountDownLatch start = new CountDownLatch(1); + + Future> alpha = executorService.submit(() -> { + start.await(); + return repository.claimNext(OWNER_ALPHA, LEASE_DURATION, 3); + }); + Future> beta = executorService.submit(() -> { + start.await(); + return repository.claimNext(OWNER_BETA, LEASE_DURATION, 3); + }); + start.countDown(); + + List claims = List.of(alpha.get(), beta.get()).stream() + .flatMap(Optional::stream) + .toList(); + + assertEquals(1, claims.size()); + assertEquals(jobRecordId, claims.getFirst().jobRecordId()); + assertEquals(1, integerColumn(jobRecordId, "attempt_count")); + } + + @Test + void terminalizesExhaustedEligibleRowsBeforeLookingForWork() { + UUID pendingJobId = insertPending(Instant.now(), 3); + UUID expiredJobId = insertRunning( + Instant.now().plusSeconds(1), + 3, + OWNER_ALPHA, + UUID.randomUUID(), + Instant.now().minusSeconds(60) + ); + + Optional lease = repository.claimNext(OWNER_BETA, LEASE_DURATION, 3); + + assertTrue(lease.isEmpty()); + assertTerminalExhaustion(pendingJobId); + assertTerminalExhaustion(expiredJobId); + } + + @Test + @Transactional + void exactLiveLeaseCanSucceedRetryOrFailAndClearsTheRightFields() { + UUID successJobId = insertPending(Instant.parse("2026-08-05T00:00:00Z"), 0); + EtlJobLease successLease = repository.claimNext(OWNER_ALPHA, LEASE_DURATION, 3) + .orElseThrow(); + repository.markSucceeded(successLease); + assertEquals(successJobId, successLease.jobRecordId()); + assertEquals("SUCCEEDED", textColumn(successJobId, "job_status")); + assertNull(textColumn(successJobId, "request_payload")); + assertNull(textColumn(successJobId, "failure_code")); + assertNull(textColumn(successJobId, "lease_owner_id")); + + UUID retryJobId = insertPending(Instant.parse("2026-08-05T00:01:00Z"), 0); + EtlJobLease retryLease = repository.claimNext(OWNER_ALPHA, LEASE_DURATION, 3) + .orElseThrow(); + repository.releaseForRetry(retryLease, 3); + assertEquals(retryJobId, retryLease.jobRecordId()); + assertEquals("PENDING", textColumn(retryJobId, "job_status")); + assertEquals(PAYLOAD, textColumn(retryJobId, "request_payload")); + assertNull(textColumn(retryJobId, "failure_code")); + assertNull(textColumn(retryJobId, "lease_owner_id")); + + UUID failedJobId = insertPending(Instant.parse("2026-08-05T00:00:30Z"), 2); + EtlJobLease failedLease = repository.claimNext(OWNER_ALPHA, LEASE_DURATION, 3) + .orElseThrow(); + repository.markFailed(failedLease, "etl_target_failure"); + assertEquals(failedJobId, failedLease.jobRecordId()); + assertEquals("FAILED", textColumn(failedJobId, "job_status")); + assertNull(textColumn(failedJobId, "request_payload")); + assertEquals("etl_target_failure", textColumn(failedJobId, "failure_code")); + assertNull(textColumn(failedJobId, "lease_owner_id")); + } + + @Test + @Transactional + void rejectsExpiredSupersededOrExhaustedTransitions() { + UUID jobRecordId = insertPending(Instant.now(), 0); + EtlJobLease lease = repository.claimNext(OWNER_ALPHA, LEASE_DURATION, 1).orElseThrow(); + jdbcTemplate.update( + "UPDATE etl_job_records SET lease_expires_at = ? WHERE job_record_id = ?", + Instant.now().minusSeconds(1), + jobRecordId + ); + + assertThrows(StaleEtlJobLeaseException.class, () -> repository.markSucceeded(lease)); + assertThrows( + StaleEtlJobLeaseException.class, + () -> repository.releaseForRetry(lease, 1) + ); + assertThrows( + StaleEtlJobLeaseException.class, + () -> repository.markFailed(lease, "etl_target_failure") + ); + + jdbcTemplate.update( + """ + UPDATE etl_job_records + SET lease_claim_id = ?, lease_expires_at = ? + WHERE job_record_id = ? + """, + UUID.randomUUID(), + Instant.now().plusSeconds(300), + jobRecordId + ); + assertThrows(StaleEtlJobLeaseException.class, () -> repository.markSucceeded(lease)); + } + + @Test + void rejectsInvalidPublicArgumentsBeforeSqlExecution() { + assertThrows(NullPointerException.class, () -> repository.claimNext(null, LEASE_DURATION, 3)); + assertThrows(NullPointerException.class, () -> repository.claimNext(OWNER_ALPHA, null, 3)); + assertThrows(IllegalArgumentException.class, () -> repository.claimNext("short", LEASE_DURATION, 3)); + assertThrows(IllegalArgumentException.class, () -> repository.claimNext(OWNER_ALPHA, Duration.ZERO, 3)); + assertThrows(IllegalArgumentException.class, () -> repository.claimNext(OWNER_ALPHA, Duration.ofSeconds(-1), 3)); + assertThrows(IllegalArgumentException.class, () -> repository.claimNext(OWNER_ALPHA, LEASE_DURATION, 0)); + assertThrows(IllegalArgumentException.class, () -> repository.claimNext(OWNER_ALPHA, LEASE_DURATION, 101)); + assertThrows(NullPointerException.class, () -> repository.markSucceeded(null)); + assertThrows(NullPointerException.class, () -> repository.releaseForRetry(null, 3)); + assertThrows(IllegalArgumentException.class, () -> repository.releaseForRetry(sampleLease(), 0)); + assertThrows(NullPointerException.class, () -> repository.markFailed(null, "etl_target_failure")); + assertThrows(NullPointerException.class, () -> repository.markFailed(sampleLease(), null)); + assertThrows(IllegalArgumentException.class, () -> repository.markFailed(sampleLease(), "UNSAFE FAILURE")); + } + + private UUID insertPending(Instant createdAt, int attemptCount) { + UUID jobRecordId = UUID.randomUUID(); + 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', ?, ?, ?) + """, + jobRecordId, + PRINCIPAL_SCOPE_HASH, + UUID.randomUUID().toString().replace("-", "").repeat(2), + REQUEST_DIGEST, + PAYLOAD, + attemptCount, + createdAt, + createdAt + ); + return jobRecordId; + } + + private UUID insertRunning( + Instant createdAt, + int attemptCount, + String ownerId, + UUID claimId, + Instant expiresAt + ) { + UUID jobRecordId = UUID.randomUUID(); + jdbcTemplate.update( + """ + INSERT INTO etl_job_records ( + job_record_id, principal_scope_hash, submission_key_hash, + request_digest, request_payload, job_status, attempt_count, + lease_claim_id, lease_owner_id, lease_expires_at, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, 'RUNNING', ?, ?, ?, ?, ?, ?) + """, + jobRecordId, + "c".repeat(64), + UUID.randomUUID().toString().replace("-", "").repeat(2), + "d".repeat(64), + PAYLOAD, + attemptCount, + claimId, + ownerId, + expiresAt, + createdAt, + createdAt + ); + return jobRecordId; + } + + private void assertTerminalExhaustion(UUID jobRecordId) { + assertEquals("FAILED", textColumn(jobRecordId, "job_status")); + assertEquals("etl_worker_attempts_exhausted", textColumn(jobRecordId, "failure_code")); + assertNull(textColumn(jobRecordId, "request_payload")); + assertNull(textColumn(jobRecordId, "lease_owner_id")); + } + + private String textColumn(UUID jobRecordId, String columnName) { + return jdbcTemplate.queryForObject( + "SELECT " + columnName + " FROM etl_job_records WHERE job_record_id = ?", + String.class, + jobRecordId + ); + } + + private int integerColumn(UUID jobRecordId, String columnName) { + Integer value = jdbcTemplate.queryForObject( + "SELECT " + columnName + " FROM etl_job_records WHERE job_record_id = ?", + Integer.class, + jobRecordId + ); + return value == null ? -1 : value; + } + + private static EtlJobLease sampleLease() { + return new EtlJobLease( + UUID.randomUUID(), + UUID.randomUUID(), + OWNER_ALPHA, + PRINCIPAL_SCOPE_HASH, + "e".repeat(64), + REQUEST_DIGEST, + PAYLOAD, + 1, + Instant.now().plusSeconds(300) + ); + } + + /** Minimal transaction-enabled SQL context for durable lease persistence 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/EtlJobLeaseRepositoryValidationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseRepositoryValidationTest.java new file mode 100644 index 00000000..b56f85ae --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseRepositoryValidationTest.java @@ -0,0 +1,92 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionStatus; + +import java.sql.ResultSet; +import java.time.Duration; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** + * Proves that the lease repository rejects unsafe arguments and impossible claim transitions. + */ +class EtlJobLeaseRepositoryValidationTest { + + @Test + void rejectsLeaseDurationsAboveTheOperationalSafetyCeilingBeforeDatabaseAccess() { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + PlatformTransactionManager transactionManager = mock(PlatformTransactionManager.class); + EtlJobLeaseRepository repository = new EtlJobLeaseRepository( + jdbcTemplate, + transactionManager + ); + + assertThrows( + IllegalArgumentException.class, + () -> repository.claimNext( + "worker-alpha", + Duration.ofSeconds( + EtlJobWorkerProperties.MAXIMUM_LEASE_DURATION_SECONDS + 1L + ), + 3 + ) + ); + verifyNoInteractions(jdbcTemplate, transactionManager); + } + + @Test + void failsClosedWhenALockedCandidateCannotBeUpdated() throws Exception { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + PlatformTransactionManager transactionManager = mock(PlatformTransactionManager.class); + TransactionStatus transactionStatus = mock(TransactionStatus.class); + ResultSet resultSet = mock(ResultSet.class); + UUID jobRecordId = UUID.randomUUID(); + + when(transactionManager.getTransaction(any(TransactionDefinition.class))) + .thenReturn(transactionStatus); + when(jdbcTemplate.update(anyString(), any(Object[].class))).thenReturn(0); + when(resultSet.getObject("job_record_id", UUID.class)).thenReturn(jobRecordId); + when(resultSet.getString("principal_scope_hash")).thenReturn("a".repeat(64)); + when(resultSet.getString("submission_key_hash")).thenReturn("b".repeat(64)); + when(resultSet.getString("request_digest")).thenReturn("c".repeat(64)); + when(resultSet.getString("request_payload")) + .thenReturn("[{\"id\":\"record_alpha\"}]"); + when(resultSet.getInt("attempt_count")).thenReturn(0); + when(resultSet.getObject("database_now", OffsetDateTime.class)) + .thenReturn(OffsetDateTime.of(2026, 8, 5, 0, 0, 0, 0, ZoneOffset.UTC)); + doAnswer(invocation -> { + @SuppressWarnings("unchecked") + RowMapper rowMapper = (RowMapper) invocation.getArgument(1); + return List.of(rowMapper.mapRow(resultSet, 0)); + }).when(jdbcTemplate).query( + anyString(), + any(RowMapper.class), + any(Object[].class) + ); + + EtlJobLeaseRepository repository = new EtlJobLeaseRepository( + jdbcTemplate, + transactionManager + ); + + assertThrows( + IllegalStateException.class, + () -> repository.claimNext("worker-alpha", Duration.ofMinutes(5), 3) + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseSuccessTransactionBoundaryTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseSuccessTransactionBoundaryTest.java new file mode 100644 index 00000000..a1f740b0 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobLeaseSuccessTransactionBoundaryTest.java @@ -0,0 +1,61 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.transaction.PlatformTransactionManager; + +import java.time.Instant; +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; + +/** + * Guards the atomicity boundary of the durable-job success transition. + * + *

Marking a job successful is valid only in the database transaction that also contains the + * target effects and response-ledger write. A direct repository call without an actual transaction + * could otherwise publish a false terminal success after unrelated or absent target work.

+ */ +class EtlJobLeaseSuccessTransactionBoundaryTest { + + /** + * Proves success fails closed before JDBC when no caller-owned transaction is active. + */ + @Test + void refusesSuccessWithoutAnActiveTransactionBeforeDatabaseAccess() { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + PlatformTransactionManager transactionManager = mock(PlatformTransactionManager.class); + EtlJobLeaseRepository repository = new EtlJobLeaseRepository( + jdbcTemplate, + transactionManager + ); + + IllegalStateException exception = assertThrows( + IllegalStateException.class, + () -> repository.markSucceeded(sampleLease()) + ); + + assertEquals( + "Durable ETL success requires an active transaction", + exception.getMessage() + ); + verifyNoInteractions(jdbcTemplate, transactionManager); + } + + private static EtlJobLease sampleLease() { + return new EtlJobLease( + UUID.randomUUID(), + UUID.randomUUID(), + "worker-alpha", + "a".repeat(64), + "b".repeat(64), + "c".repeat(64), + "[{\"id\":\"record_alpha\"}]", + 1, + Instant.now().plusSeconds(300) + ); + } +} 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 46aa44bc..9ea22672 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 @@ -51,21 +51,36 @@ void migrationReservesStableWorkerStatesAndRequiresTerminalPayloadClearing() thr } @Test - void runbookDocumentsAcceptedSemanticsOwnershipAndTheWorkerBoundary() throws IOException { + void runbookDocumentsAcceptedSemanticsOwnershipAndActiveWorkerBoundary() throws IOException { String runbook = read("docs/etl/durable-job-intake.md").replaceAll("\\s+", " "); assertTrue(runbook.contains("202 Accepted")); assertTrue(runbook.contains("Location: /api/etl/jobs/{job_record_id}")); assertTrue(runbook.contains("same authenticated principal")); assertTrue(runbook.contains("byte-for-byte same JSON text")); - assertTrue(runbook.contains("does not execute jobs yet")); + assertFalse(runbook.contains("does not execute jobs yet")); + assertTrue(runbook.contains("executes at most one eligible durable job per worker poll")); + assertTrue(runbook.contains("FOR UPDATE SKIP LOCKED")); + assertTrue(runbook.contains("lease fencing")); assertTrue(runbook.contains("request payload")); - assertTrue(runbook.contains("worker and lease-fencing slice")); assertTrue(runbook.contains("Cache-Control: no-store")); assertTrue(runbook.contains("422 etl_job_submission_key_reused")); - assertTrue(runbook.contains("disabled by default")); + assertTrue(runbook.contains("intake is disabled by default")); + assertTrue(runbook.contains("worker is disabled by default")); assertTrue(runbook.contains("mightyetl.etl.jobs.intake-enabled=true")); assertTrue(runbook.contains("xtrmetl.etl.jobs.intake-enabled=true")); + assertTrue(runbook.contains("xtrmetl.etl.jobs.worker.enabled=true")); + } + + @Test + void changelogRecordsLeaseFencedDurableWorkerExecution() throws IOException { + String changelog = read("CHANGELOG.md").replaceAll("\\s+", " "); + + assertTrue(changelog.contains( + "Durable ETL jobs now execute through an opt-in lease-fenced worker" + )); + assertTrue(changelog.contains("FOR UPDATE SKIP LOCKED")); + assertTrue(changelog.contains("stable non-sensitive failure codes")); } private static String read(String relativePath) throws IOException { diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobWorkerPropertiesTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobWorkerPropertiesTest.java new file mode 100644 index 00000000..8a5239f3 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobWorkerPropertiesTest.java @@ -0,0 +1,119 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.Test; + +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Specifies fail-closed activation and bounded durable-job worker configuration. + */ +class EtlJobWorkerPropertiesTest { + + private static final long MAXIMUM_SCHEDULER_DELAY_MILLISECONDS = 86_400_000L; + private static final long MAXIMUM_LEASE_DURATION_SECONDS = 86_400L; + + @Test + void defaultsToDisabledBoundedPollingWithGeneratedSafeOwner() { + EtlJobWorkerProperties properties = new EtlJobWorkerProperties(); + + assertFalse(properties.isEnabled()); + assertEquals(5_000L, properties.getFixedDelayMilliseconds()); + assertEquals(5_000L, properties.getInitialDelayMilliseconds()); + assertEquals(300L, properties.getLeaseDurationSeconds()); + assertEquals(3, properties.getMaxAttempts()); + assertTrue(properties.getLeaseOwnerId().matches("[A-Za-z0-9._:-]{8,128}")); + + EtlJobWorkerProperties another = new EtlJobWorkerProperties(); + assertNotEquals(properties.getLeaseOwnerId(), another.getLeaseOwnerId()); + } + + @Test + void acceptsEverySupportedBoundary() { + EtlJobWorkerProperties properties = new EtlJobWorkerProperties(); + + properties.setEnabled(true); + properties.setFixedDelayMilliseconds(1L); + properties.setInitialDelayMilliseconds(0L); + properties.setLeaseDurationSeconds(1L); + properties.setMaxAttempts(1); + properties.setLeaseOwnerId("worker-01"); + + assertTrue(properties.isEnabled()); + assertEquals(1L, properties.getFixedDelayMilliseconds()); + assertEquals(0L, properties.getInitialDelayMilliseconds()); + assertEquals(1L, properties.getLeaseDurationSeconds()); + assertEquals(1, properties.getMaxAttempts()); + assertEquals("worker-01", properties.getLeaseOwnerId()); + + properties.setFixedDelayMilliseconds(MAXIMUM_SCHEDULER_DELAY_MILLISECONDS); + properties.setInitialDelayMilliseconds(MAXIMUM_SCHEDULER_DELAY_MILLISECONDS); + properties.setLeaseDurationSeconds(MAXIMUM_LEASE_DURATION_SECONDS); + properties.setMaxAttempts(100); + properties.setLeaseOwnerId("w".repeat(128)); + assertEquals(MAXIMUM_SCHEDULER_DELAY_MILLISECONDS, properties.getFixedDelayMilliseconds()); + assertEquals(MAXIMUM_SCHEDULER_DELAY_MILLISECONDS, properties.getInitialDelayMilliseconds()); + assertEquals(MAXIMUM_LEASE_DURATION_SECONDS, properties.getLeaseDurationSeconds()); + assertEquals(100, properties.getMaxAttempts()); + assertEquals(128, properties.getLeaseOwnerId().length()); + } + + @Test + void rejectsUnsafeNumericConfiguration() { + EtlJobWorkerProperties properties = new EtlJobWorkerProperties(); + + assertThrows( + IllegalArgumentException.class, + () -> properties.setFixedDelayMilliseconds(0L) + ); + assertThrows( + IllegalArgumentException.class, + () -> properties.setFixedDelayMilliseconds( + MAXIMUM_SCHEDULER_DELAY_MILLISECONDS + 1L + ) + ); + assertThrows( + IllegalArgumentException.class, + () -> properties.setInitialDelayMilliseconds(-1L) + ); + assertThrows( + IllegalArgumentException.class, + () -> properties.setInitialDelayMilliseconds( + MAXIMUM_SCHEDULER_DELAY_MILLISECONDS + 1L + ) + ); + assertThrows( + IllegalArgumentException.class, + () -> properties.setLeaseDurationSeconds(0L) + ); + assertThrows( + IllegalArgumentException.class, + () -> properties.setLeaseDurationSeconds(MAXIMUM_LEASE_DURATION_SECONDS + 1L) + ); + assertThrows(IllegalArgumentException.class, () -> properties.setMaxAttempts(0)); + assertThrows(IllegalArgumentException.class, () -> properties.setMaxAttempts(101)); + } + + @Test + void rejectsMissingShortLongOrUnsafeOwnerIdentifiers() { + EtlJobWorkerProperties properties = new EtlJobWorkerProperties(); + + assertThrows(NullPointerException.class, () -> properties.setLeaseOwnerId(null)); + assertThrows(IllegalArgumentException.class, () -> properties.setLeaseOwnerId("short")); + assertThrows( + IllegalArgumentException.class, + () -> properties.setLeaseOwnerId("w".repeat(129)) + ); + assertThrows( + IllegalArgumentException.class, + () -> properties.setLeaseOwnerId("worker identifier") + ); + assertThrows( + IllegalArgumentException.class, + () -> properties.setLeaseOwnerId("worker/identifier") + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobWorkerTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobWorkerTest.java new file mode 100644 index 00000000..eaf0fbfa --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobWorkerTest.java @@ -0,0 +1,315 @@ +package com.xtrmetl.etl.job; + +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.dao.CannotAcquireLockException; +import org.springframework.dao.DataIntegrityViolationException; + +import java.time.Instant; +import java.util.Optional; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Specifies one-job polling, bounded retries, stable failures, stale fencing, and safe metrics. + */ +@ExtendWith(MockitoExtension.class) +class EtlJobWorkerTest { + + private static final String OWNER_ID = "worker-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 static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + + @Mock + private EtlJobLeaseRepository leaseRepository; + + @Mock + private EtlJobExecutionService executionService; + + private EtlJobWorkerProperties properties; + private SimpleMeterRegistry meterRegistry; + private EtlJobWorker worker; + + @BeforeEach + void createWorker() { + properties = new EtlJobWorkerProperties(); + properties.setEnabled(true); + properties.setLeaseOwnerId(OWNER_ID); + properties.setLeaseDurationSeconds(120L); + properties.setMaxAttempts(3); + meterRegistry = new SimpleMeterRegistry(); + worker = new EtlJobWorker( + leaseRepository, + executionService, + properties, + meterRegistry + ); + } + + @Test + void recordsIdleWithoutExecutingWhenNoJobIsEligible() { + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.empty()); + + worker.pollOnce(); + + verify(executionService, never()).execute(any()); + assertMetric("idle", 1.0, 1L); + } + + @Test + void executesAtMostOneClaimAndRecordsSucceeded() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + + worker.pollOnce(); + + verify(executionService).execute(lease); + assertMetric("succeeded", 1.0, 1L); + } + + @Test + void releasesTransientFailureWhenAttemptsRemain() { + EtlJobLease lease = lease(2); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new CannotAcquireLockException("temporary")) + .when(executionService).execute(lease); + + worker.pollOnce(); + + verify(leaseRepository).releaseForRetry(lease, 3); + verify(leaseRepository, never()).markFailed(any(), anyString()); + assertMetric("retried", 1.0, 1L); + } + + @Test + void terminalizesTransientFailureAtTheAttemptLimit() { + EtlJobLease lease = lease(3); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new CannotAcquireLockException("temporary")) + .when(executionService).execute(lease); + + worker.pollOnce(); + + verify(leaseRepository).markFailed(lease, "etl_target_unavailable"); + verify(leaseRepository, never()).releaseForRetry(any(), anyInt()); + assertMetric("failed", 1.0, 1L); + } + + @Test + void terminalizesIntegrityFailureWithItsStableCode() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new EtlJobIntegrityException()).when(executionService).execute(lease); + + worker.pollOnce(); + + verify(leaseRepository).markFailed(lease, "etl_job_integrity_failure"); + assertMetric("failed", 1.0, 1L); + } + + @Test + void terminalizesDeterministicRequestFailureWithItsStableCode() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new EtlRequestException(EtlRequestError.INVALID_JSON)) + .when(executionService).execute(lease); + + worker.pollOnce(); + + verify(leaseRepository).markFailed(lease, "etl_invalid_json"); + assertMetric("failed", 1.0, 1L); + } + + @Test + void terminalizesNonTransientTargetFailureWithoutDiagnosticText() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new DataIntegrityViolationException("sensitive SQL")) + .when(executionService).execute(lease); + + worker.pollOnce(); + + verify(leaseRepository).markFailed(lease, "etl_target_failure"); + assertMetric("failed", 1.0, 1L); + } + + @Test + void terminalizesUnexpectedRuntimeFailureWithGenericCode() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new IllegalStateException("sensitive implementation detail")) + .when(executionService).execute(lease); + + worker.pollOnce(); + + verify(leaseRepository).markFailed(lease, "etl_internal_error"); + assertMetric("failed", 1.0, 1L); + } + + @Test + void treatsAnExecutionFenceFailureAsStaleEvidence() { + EtlJobLease executionStaleLease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(executionStaleLease)); + doThrow(new StaleEtlJobLeaseException()) + .when(executionService).execute(executionStaleLease); + + worker.pollOnce(); + + verify(leaseRepository, never()).markFailed(any(), anyString()); + verify(leaseRepository, never()).releaseForRetry(any(), anyInt()); + assertMetric("stale", 1.0, 1L); + } + + @Test + void treatsAStaleRetryTransitionAsStaleEvidence() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new CannotAcquireLockException("temporary")) + .when(executionService).execute(lease); + doThrow(new StaleEtlJobLeaseException()) + .when(leaseRepository).releaseForRetry(lease, 3); + + worker.pollOnce(); + + assertMetric("stale", 1.0, 1L); + assertMetric("retried", 0.0, 0L); + } + + @Test + void recordsRetryTransitionDatabaseFailureAsFailedEvidence() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new CannotAcquireLockException("temporary")) + .when(executionService).execute(lease); + doThrow(new DataIntegrityViolationException("sensitive SQL")) + .when(leaseRepository).releaseForRetry(lease, 3); + + worker.pollOnce(); + + assertMetric("failed", 1.0, 1L); + assertMetric("retried", 0.0, 0L); + assertMetric("stale", 0.0, 0L); + } + + @Test + void treatsAStaleTerminalTransitionAsStaleEvidence() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new EtlRequestException(EtlRequestError.INVALID_JSON)) + .when(executionService).execute(lease); + doThrow(new StaleEtlJobLeaseException()) + .when(leaseRepository).markFailed(lease, "etl_invalid_json"); + + worker.pollOnce(); + + assertMetric("stale", 1.0, 1L); + assertMetric("failed", 0.0, 0L); + } + + @Test + void recordsTerminalTransitionDatabaseFailureAsFailedEvidence() { + EtlJobLease lease = lease(1); + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenReturn(Optional.of(lease)); + doThrow(new EtlRequestException(EtlRequestError.INVALID_JSON)) + .when(executionService).execute(lease); + doThrow(new DataIntegrityViolationException("sensitive SQL")) + .when(leaseRepository).markFailed(lease, "etl_invalid_json"); + + worker.pollOnce(); + + assertMetric("failed", 1.0, 1L); + assertMetric("stale", 0.0, 0L); + } + + @Test + void recordsClaimDatabaseFailureWithoutLeakingOrExecuting() { + when(leaseRepository.claimNext(anyString(), any(), anyInt())) + .thenThrow(new CannotAcquireLockException("sensitive SQL")); + + worker.pollOnce(); + + verify(executionService, never()).execute(any()); + assertMetric("failed", 1.0, 1L); + } + + @Test + void rejectsMissingCollaborators() { + assertThrows( + NullPointerException.class, + () -> new EtlJobWorker(null, executionService, properties, meterRegistry) + ); + assertThrows( + NullPointerException.class, + () -> new EtlJobWorker(leaseRepository, null, properties, meterRegistry) + ); + assertThrows( + NullPointerException.class, + () -> new EtlJobWorker(leaseRepository, executionService, null, meterRegistry) + ); + assertThrows( + NullPointerException.class, + () -> new EtlJobWorker(leaseRepository, executionService, properties, null) + ); + } + + private void assertMetric(String outcome, double counterCount, long timerCount) { + assertEquals( + counterCount, + meterRegistry.find("etl.jobs.worker.outcomes") + .tag("outcome", outcome) + .counter() + .count() + ); + assertEquals( + timerCount, + meterRegistry.find("etl.jobs.execution.duration") + .tag("outcome", outcome) + .timer() + .count() + ); + } + + private static EtlJobLease lease(int attemptCount) { + return new EtlJobLease( + UUID.randomUUID(), + UUID.randomUUID(), + OWNER_ID, + PRINCIPAL_SCOPE_HASH, + SUBMISSION_KEY_HASH, + REQUEST_DIGEST, + PAYLOAD, + attemptCount, + Instant.now().plusSeconds(300) + ); + } +}