diff --git a/CHANGELOG.md b/CHANGELOG.md index d4898847..57a0b362 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- The durable job pagination index now uses PostgreSQL `CREATE INDEX CONCURRENTLY` with a migration-local Flyway `executeInTransaction=false` companion configuration, preserving production writers and documenting invalid-index recovery and concurrent rollback. +- Durable job operators can now list only their own jobs through bounded newest-first keyset pagination with canonical opaque cursors, deterministic timestamp-plus-UUID ordering, `Cache-Control: no-store`, and RFC 8288 next-page links without offset drift or cross-tenant existence leakage. - The durable-job claim eligibility index now builds in a separate PostgreSQL `CREATE INDEX CONCURRENTLY` migration with Flyway non-transactional script configuration and session-level PostgreSQL migration locking, preserving normal job writes during rollout while keeping lease columns and constraints transactional. - Durable asynchronous ETL jobs now progress from `PENDING` through lease-fenced execution to `SUCCEEDED` or `FAILED`; PostgreSQL owns cross-replica claiming, stale workers cannot commit target or lifecycle effects, and intake and execution remain independently fail-closed. - Durable-worker observability now records one terminal outcome counter and one matching duration sample for every completed poll, including idle polls and database failures while persisting retry or terminal transitions. @@ -31,6 +33,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Owner-scoped durable job list models and HTTP contract, strict cursor and page-limit validation, one-extra-row next-page detection, the descriptive `etl_job_owner_pagination_index`, deterministic tenant-isolation and equal-timestamp tests, migration rollback guidance, and APA 7th standards evidence in `docs/etl/durable-job-intake.md`. - A production rollout and invalid-index recovery runbook for the nonblocking durable-job claim index: `docs/operations/durable-job-claim-index-rollout.md`. - PostgreSQL `FOR UPDATE SKIP LOCKED` durable-job claiming, per-process and per-claim lease fencing, expiry reclaim, bounded attempts, exact-live-lease transitions, terminal payload clearing, stable failure codes, and finite-cardinality worker metrics. - Hashed durable execution identity and domain-separated reuse of `etl_idempotency_records`, coupling response replay or creation, target writes, and terminal `SUCCEEDED` in one transaction without retaining or reconstructing raw principals or raw client idempotency keys. @@ -67,6 +70,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- Job-list cursors contain only ordering keys, never authority or sensitive values; every page query independently binds the hashed authenticated principal, malformed and non-canonical cursors fail before database access, and list responses exclude payloads, principals, keys, hashes, lease identifiers, SQL, and exception text. - Durable-worker metrics and ordinary logs exclude payloads, raw principals, raw idempotency keys, hashes, job and lease identifiers, SQL, exception messages, and unbounded exception labels. - Retained payload or response-ledger identity conflicts fail closed with `etl_job_integrity_failure`; an expired or superseded lease rolls back target, ledger, and terminal-state effects. diff --git a/docs/etl/durable-job-intake.md b/docs/etl/durable-job-intake.md index dc54f365..c00121d7 100644 --- a/docs/etl/durable-job-intake.md +++ b/docs/etl/durable-job-intake.md @@ -4,9 +4,10 @@ `POST /api/etl/jobs` creates a durable, authenticated-principal-scoped ETL job resource. A separate lease-fenced worker claims accepted jobs across replicas, replays or writes the durable response -ledger, writes target rows, and commits terminal state atomically. +ledger, writes target rows, and commits terminal state atomically. Authenticated operators can also +list recent jobs in their own principal namespace through deterministic keyset pagination. -Both capabilities are fail-closed: +Both intake and execution capabilities are fail-closed: ```text mightyetl.etl.jobs.intake-enabled=false @@ -57,6 +58,56 @@ same job identifier and `Idempotency-Replayed: true`. Reusing one principal-scop JSON returns `422 etl_job_submission_key_reused`. A concurrent creation attempt that cannot acquire the transaction-level submission lock returns `409 etl_job_submission_in_progress`. +## List owned jobs + +```http +GET /api/etl/jobs?limit=50 HTTP/1.1 +Authorization: Basic +``` + +The endpoint returns only jobs owned by the same authenticated principal. It orders rows by +`created_at DESC, job_record_id DESC`; the UUID is a deterministic tie-breaker when jobs share one +database timestamp. The default page size is 50 and the accepted canonical range is 1 through 100. +Values such as `0`, `101`, `01`, signed values, whitespace-padded values, and non-decimal text fail +with `400 etl_invalid_job_page_limit` before table access. + +The service fetches one additional row beyond the requested page size. That row is not returned; it +only proves that another page exists. When a following page is available, the body includes an opaque +URL-safe cursor and the response advertises the same target through RFC 8288 Web Linking: + +```http +HTTP/1.1 200 OK +Cache-Control: no-store +Link: ; rel="next" +Content-Type: application/json + +{ + "jobs": [ + { + "jobRecordId": "cf4f083f-8c90-4f34-a8b6-b53761de44ef", + "jobStatus": "SUCCEEDED", + "attemptCount": 1, + "createdAt": "2026-08-05T01:00:00Z", + "updatedAt": "2026-08-05T01:00:05Z" + } + ], + "nextCursor": "eyJvcGFxdWUiOiJleGFtcGxlIn0" +} +``` + +The actual cursor is a canonical unpadded Base64 URL encoding of the last returned creation timestamp +and job identifier. Clients must treat it as opaque. Each following query still binds the current +principal hash and applies a strict tuple boundary equivalent to “older timestamp, or the same +timestamp with a lower UUID.” Cursor contents never grant authority and reveal no payload, principal, +submission key, or hash. Malformed, oversized, incomplete, non-canonical, or stale-format cursors fail +closed with `400 etl_invalid_job_page_cursor` before database access. A terminal or empty page omits +both `nextCursor` and the `Link` header. + +Pagination guarantees no duplicates or omissions while traversing an unchanged dataset. Concurrent +insertions are visible according to their ordering position and do not convert a cursor into a +snapshot transaction. Consumers needing a legally frozen audit set must export from an explicit +transactional or warehouse snapshot rather than treating this operational list as one. + ## Read job status ```http @@ -68,16 +119,21 @@ The query binds the current principal hash and job identifier. A malformed, miss identifier returns the same `404 etl_job_not_found`, preventing tenant-existence probing. The representation exposes only the opaque job identifier, stable lifecycle state, bounded attempt -count, stable failure code where applicable, status URL, and timestamps. It excludes request payload, -raw principal, raw submission key, internal hashes, lease identifiers, SQL, and response-ledger data. +count, stable failure code where applicable, and timestamps. It excludes request payload, raw +principal, raw submission key, internal hashes, lease identifiers, SQL, and response-ledger data. ## Lifecycle and distribution Flyway migrations create descriptive multi-word `snake_case` objects: - `V2__create_etl_job_records.sql` creates `etl_job_records` and the submission uniqueness contract; -- `V3__add_etl_job_lease_fencing.sql` adds `lease_claim_id`, `lease_owner_id`, - `lease_expires_at`, lifecycle constraints, and `etl_job_claim_eligibility_index`. +- `V3__add_etl_job_lease_fencing.sql` transactionally adds `lease_claim_id`, `lease_owner_id`, + `lease_expires_at`, legacy-data repair, and lifecycle constraints; +- `V4__add_etl_job_claim_eligibility_index.sql` concurrently adds + `etl_job_claim_eligibility_index` for oldest eligible queue-like claims without blocking writers; +- `V5__add_etl_job_owner_pagination_index.sql` concurrently adds + `etl_job_owner_pagination_index` on `principal_scope_hash`, `created_at DESC`, and + `job_record_id DESC` for the exact owner-scoped ordering contract. The stable lifecycle is `PENDING`, `RUNNING`, `SUCCEEDED`, and `FAILED`. @@ -125,18 +181,64 @@ failure, attempts exhaustion, and non-retryable failure clear the payload in the transition. Apply least privilege, encryption, backup, restore, and retention controls while data is retained. -Metrics and ordinary logs must not include payloads, principals, client keys, hashes, job or lease -identifiers, SQL, exception messages, or unbounded error classes. Operational procedures and metric -contracts are authoritative in `docs/operations/durable-job-worker.md`. +List and status representations exclude payloads, raw principals, raw keys, hashes, lease identifiers, +SQL, and exception messages. Metrics and ordinary logs must not include those values or unbounded +error classes. Operational procedures and metric contracts are authoritative in +`docs/operations/durable-job-worker.md`. Claim-index deployment and invalid-index recovery are +specified in `docs/operations/durable-job-claim-index-rollout.md`. + +## Migration and rollback + +Apply Flyway migrations in version order. V3 remains transactional so lease columns, legacy-data +repair, and lifecycle constraints commit together. V4 and V5 are isolated additive index migrations. +Both use PostgreSQL `CREATE INDEX CONCURRENTLY` so inserts, updates, and deletes remain available while +the indexes are built. Their matching `.sql.conf` files contain `executeInTransaction=false` because +PostgreSQL rejects concurrent index creation inside a transaction block. The application also sets +`spring.flyway.postgresql.transactional-lock=false`, selecting Flyway's PostgreSQL session-lock mode +required for concurrent index DDL. + +Concurrent index creation performs more work and can wait for transactions that could affect an +index. Measure duration, I/O, replication lag, and transaction age in a representative staging +environment, then schedule production rollout with explicit monitoring. If a build fails, PostgreSQL +can leave an invalid index. Inspect catalog validity rather than treating a matching object name or +schema-history row as usable evidence. The claim-index recovery procedure is documented separately; +the same fail-closed catalog inspection, concurrent removal, root-cause correction, approved Flyway +repair, and unchanged migration replay applies to the pagination index. + +Application rollback is compatible with either additional index because older binaries ignore them. +After rolling back every binary that depends on the relevant access path, run the database-only +rollback outside a transaction block: + +```sql +DROP INDEX CONCURRENTLY etl_job_owner_pagination_index; +DROP INDEX CONCURRENTLY etl_job_claim_eligibility_index; +``` + +Dropping the pagination index while the list endpoint is active preserves query correctness but can +cause an unacceptable owner-list scan cost. Dropping the claim index while workers are active can +cause expensive claim scans and contention. Do not remove either index until its traffic is withdrawn +and an execution-plan review confirms the rollback boundary. ## Standards basis - RFC 9110 Section 15.3.3 defines `202 Accepted` as noncommittal and recommends a current-status representation and status monitor. +- RFC 8288 defines the Web Linking model and the HTTP `Link` header used for the optional next-page + relationship. - RFC 9457 supplies deterministic problem-details representations. - RFC 9651 defines the accepted Structured Fields String syntax. +- PostgreSQL 18 requires explicit `ORDER BY` for guaranteed result ordering and recommends a unique + ordering when `LIMIT` is used. +- PostgreSQL 18 documents that equality constraints on leading multicolumn B-tree keys plus a range + constraint on the next key efficiently limit the scanned index portion. +- PostgreSQL 18 documents that ordinary index construction blocks writes, while concurrent index + construction preserves writes with additional scans, waits, and invalid-index recovery caveats. - PostgreSQL 18 documents `SKIP LOCKED` as unsuitable for a general consistent view but useful for avoiding contention among multiple consumers of a queue-like table. +- Flyway script configuration supports a migration-matched `.sql.conf` file and the + `executeInTransaction=false` override required for non-transactional PostgreSQL DDL. +- Flyway's PostgreSQL integration documents session-level migration locking for statements such as + `CREATE INDEX CONCURRENTLY`. - Spring fixed-delay scheduling measures each delay from completion of the preceding invocation. - OpenTelemetry SQL/PostgreSQL semantic conventions define stable database telemetry fields; raw query text and parameters remain privacy-sensitive opt-in data. @@ -146,6 +248,9 @@ contracts are authoritative in `docs/operations/durable-job-worker.md`. Fielding, R., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC 9110). RFC Editor. https://www.rfc-editor.org/rfc/rfc9110 +Nottingham, M. (2017). *Web linking* (RFC 8288). RFC Editor. +https://doi.org/10.17487/RFC8288 + Nottingham, M., & Wilde, E. (2023). *Problem details for HTTP APIs* (RFC 9457). RFC Editor. https://www.rfc-editor.org/rfc/rfc9457 @@ -156,8 +261,14 @@ OpenTelemetry Authors. (2026). *OpenTelemetry semantic conventions 1.43.0: Seman SQL databases client operations*. Cloud Native Computing Foundation. https://opentelemetry.io/docs/specs/semconv/db/sql/ -PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: SELECT*. -https://www.postgresql.org/docs/18/sql-select.html +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: CREATE INDEX*. +https://www.postgresql.org/docs/18/sql-createindex.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Introduction to indexes*. +https://www.postgresql.org/docs/18/indexes-intro.html + +Redgate Software. (2026). *Flyway PostgreSQL transactional lock setting*. +https://documentation.red-gate.com/fd/flyway-postgresql-transactional-lock-setting-277579114.html -Spring Authors. (2026). *Task execution and scheduling*. Broadcom. -https://docs.spring.io/spring-framework/reference/integration/scheduling.html +Redgate Software. (2026). *Flyway script configuration*. +https://documentation.red-gate.com/flyway/reference/script-configuration diff --git a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java index 1d467840..336e67c4 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java @@ -1,6 +1,8 @@ package com.xtrmetl.etl.controller; import com.xtrmetl.etl.job.EtlJobAcceptedResponse; +import com.xtrmetl.etl.job.EtlJobPage; +import com.xtrmetl.etl.job.EtlJobPageResponse; import com.xtrmetl.etl.job.EtlJobService; import com.xtrmetl.etl.job.EtlJobSnapshot; import com.xtrmetl.etl.job.EtlJobStatusResponse; @@ -11,6 +13,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; import org.springframework.dao.DataAccessException; import org.springframework.http.CacheControl; +import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.lang.Nullable; @@ -20,7 +23,9 @@ import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.util.UriComponentsBuilder; import java.net.URI; import java.security.Principal; @@ -28,16 +33,15 @@ import java.util.UUID; /** - * Exposes durable asynchronous ETL job submission and owner-scoped status resources. + * Exposes durable asynchronous ETL job submission, discovery, and status resources. * *

Submission requires authentication and an {@code Idempotency-Key}. The accepted response is * intentionally noncommittal under RFC 9110: it reports the durable pending state and supplies a - * status-monitor resource through both the representation and {@code Location} header. This intake - * slice does not claim that worker execution has started.

+ * status-monitor resource through both the representation and {@code Location} header.

* - *

Because this bounded slice retains payloads but does not yet execute jobs or clear terminal - * payloads, the controller is disabled by default. Operators must explicitly set - * {@code xtrmetl.etl.jobs.intake-enabled=true} after accepting that temporary lifecycle boundary.

+ *

Job discovery is owner-scoped and uses an opaque keyset cursor. A next-page link is emitted + * under RFC 8288 only when another page exists. The service independently binds every list query to + * the authenticated principal hash, so cursor contents never grant authority.

* *

Success and covered failure responses use {@code Cache-Control: no-store}. Malformed, absent, * and foreign-owned job identifiers use the same owner-safe not-found classification so the status @@ -56,6 +60,8 @@ public class EtlJobController { /** Response header indicating whether a prior durable submission was replayed. */ public static final String IDEMPOTENCY_REPLAYED_HEADER = "Idempotency-Replayed"; + private static final String DEFAULT_JOB_PAGE_LIMIT_TEXT = "50"; + private final EtlJobService etlJobService; /** @@ -123,6 +129,51 @@ public ResponseEntity submit( .body(responseBody); } + /** + * Lists one deterministic page of jobs in the authenticated principal namespace. + * + * @param cursor opaque next-page cursor, or {@code null} for the newest page + * @param limit canonical decimal page size from 1 through 100, or {@code null} for 50 + * @param principal authenticated principal namespace + * @return owner-scoped page with an RFC 8288 next link only when another page exists + */ + @GetMapping + @Observed(name = "etl.jobs.list", contextualName = "etl-job-list") + public ResponseEntity list( + @RequestParam(value = "cursor", required = false) @Nullable String cursor, + @RequestParam(value = "limit", required = false) @Nullable String limit, + @Nullable Principal principal + ) { + if (principal == null) { + throw new EtlRequestException(EtlRequestError.IDEMPOTENCY_PRINCIPAL_REQUIRED); + } + + final EtlJobPage page; + try { + page = etlJobService.listOwned(principal.getName(), cursor, limit); + } catch (EtlRequestException | DataAccessException exception) { + throw exception; + } catch (RuntimeException exception) { + throw new EtlUnexpectedException(exception); + } + + ResponseEntity.BodyBuilder responseBuilder = ResponseEntity.ok() + .cacheControl(CacheControl.noStore()); + if (page.nextCursor() != null) { + String effectiveLimit = limit == null ? DEFAULT_JOB_PAGE_LIMIT_TEXT : limit; + String nextTarget = UriComponentsBuilder.fromPath("/api/etl/jobs") + .queryParam("limit", effectiveLimit) + .queryParam("cursor", page.nextCursor()) + .build() + .toUriString(); + responseBuilder.header( + HttpHeaders.LINK, + "<" + nextTarget + ">; rel=\"next\"" + ); + } + return responseBuilder.body(EtlJobPageResponse.from(page)); + } + /** * Returns one status resource only within the authenticated principal namespace. * diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobPage.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobPage.java new file mode 100644 index 00000000..3f91e4cb --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobPage.java @@ -0,0 +1,32 @@ +package com.xtrmetl.etl.job; + +import org.springframework.lang.Nullable; + +import java.util.List; +import java.util.Objects; + +/** + * Immutable owner-scoped page of operator-safe durable ETL job snapshots. + * + *

The list is defensively copied so callers cannot mutate a page after the service has derived + * its next-cursor boundary. The optional cursor is opaque to clients and identifies the last item + * returned by this page; it is absent when the current page is terminal.

+ * + * @param jobs immutable operator-safe job snapshots in deterministic newest-first order + * @param nextCursor opaque cursor for the following page, or {@code null} when no page follows + */ +public record EtlJobPage( + List jobs, + @Nullable String nextCursor +) { + + /** + * Validates and defensively copies the immutable page. + * + * @param jobs non-null snapshots without null elements + * @param nextCursor opaque following-page cursor, or {@code null} + */ + public EtlJobPage { + jobs = List.copyOf(Objects.requireNonNull(jobs, "jobs must not be null")); + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobPageResponse.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobPageResponse.java new file mode 100644 index 00000000..042c26d8 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobPageResponse.java @@ -0,0 +1,47 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.annotation.JsonInclude; +import org.springframework.lang.Nullable; + +import java.util.List; +import java.util.Objects; + +/** + * Client-visible owner-scoped page of durable ETL job status resources. + * + *

Each item is converted through {@link EtlJobStatusResponse}, so retained payloads, raw + * principals, idempotency keys, hashes, SQL, and exception text cannot enter the page response. + * The next cursor is omitted from JSON when the page is terminal.

+ * + * @param jobs immutable client-safe job status representations + * @param nextCursor opaque following-page cursor, omitted when no page follows + */ +public record EtlJobPageResponse( + List jobs, + @JsonInclude(JsonInclude.Include.NON_NULL) @Nullable String nextCursor +) { + + /** + * Validates and defensively copies the immutable response page. + * + * @param jobs non-null client-safe statuses without null elements + * @param nextCursor opaque following-page cursor, or {@code null} + */ + public EtlJobPageResponse { + jobs = List.copyOf(Objects.requireNonNull(jobs, "jobs must not be null")); + } + + /** + * Converts an internal owner-scoped page into the public wire representation. + * + * @param page immutable internal job page + * @return client-safe page response + */ + public static EtlJobPageResponse from(EtlJobPage page) { + EtlJobPage requiredPage = Objects.requireNonNull(page, "page must not be null"); + List responses = requiredPage.jobs().stream() + .map(EtlJobStatusResponse::from) + .toList(); + return new EtlJobPageResponse(responses, requiredPage.nextCursor()); + } +} diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java index e1992d17..e1fde8be 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java @@ -20,6 +20,8 @@ import java.nio.charset.StandardCharsets; import java.sql.Timestamp; import java.time.Instant; +import java.time.format.DateTimeParseException; +import java.util.Base64; import java.util.HashSet; import java.util.List; import java.util.Locale; @@ -29,18 +31,23 @@ import java.util.regex.Pattern; /** - * Creates and reads durable principal-scoped asynchronous ETL job resources. + * Creates, reads, and lists durable principal-scoped asynchronous ETL job resources. * *

The intake path validates the complete bounded JSON batch before persistence. Raw * authentication principals and idempotency keys are never stored. Instead, independent SHA-256 * hashes namespace the submission, while the exact JSON text is retained only because a later - * worker slice must execute the accepted job. The status representation never exposes that payload - * or either internal hash.

+ * worker slice must execute the accepted job. Status and list representations never expose that + * payload or either internal hash.

* *

A PostgreSQL transaction-level try-lock serializes creation of one principal-scoped * submission key. The table-level unique constraint remains a second integrity boundary. Replaying * byte-identical JSON returns the original job identifier; reusing the key with different JSON text * returns a deterministic conflict.

+ * + *

Job discovery uses owner-scoped keyset pagination ordered by creation time and UUID. The + * opaque cursor is non-authoritative: it contains only the last returned ordering key, while the + * principal hash remains an independent mandatory query predicate. Malformed or non-canonical + * cursors fail closed before database access.

*/ @Service public class EtlJobService { @@ -70,8 +77,31 @@ INSERT INTO etl_job_records ( WHERE job_record_id = ? AND principal_scope_hash = ? """; + private static final String SELECT_OWNED_JOB_PAGE_SQL = """ + SELECT job_record_id, job_status, attempt_count, + failure_code, created_at, updated_at + FROM etl_job_records + WHERE principal_scope_hash = ? + ORDER BY created_at DESC, job_record_id DESC + LIMIT ? + """; + private static final String SELECT_OWNED_JOB_PAGE_AFTER_CURSOR_SQL = """ + SELECT job_record_id, job_status, attempt_count, + failure_code, created_at, updated_at + FROM etl_job_records + WHERE principal_scope_hash = ? + AND ( + created_at < ? + OR (created_at = ? AND job_record_id < ?) + ) + ORDER BY created_at DESC, job_record_id DESC + LIMIT ? + """; private static final int MAX_PRINCIPAL_SCOPE_CODE_POINTS = 512; private static final int MAX_RECORD_ID_CODE_POINTS = 256; + private static final int DEFAULT_JOB_PAGE_SIZE = 50; + private static final int MAX_JOB_PAGE_SIZE = 100; + private static final int MAX_JOB_PAGE_CURSOR_CHARACTERS = 192; private static final Pattern OUTER_IDENTIFIER_WHITESPACE = Pattern.compile( "^\\s|\\s$", Pattern.UNICODE_CHARACTER_CLASS @@ -83,6 +113,7 @@ INSERT INTO etl_job_records ( private static final Pattern IDEMPOTENCY_KEY_STRUCTURED_FIELD_PROFILE = Pattern.compile( "\"(" + IDEMPOTENCY_KEY_VALUE_EXPRESSION + ")\"" ); + private static final Pattern JOB_PAGE_LIMIT_PROFILE = Pattern.compile("[1-9][0-9]{0,2}"); private final JdbcTemplate jdbcTemplate; private final ObjectMapper objectMapper; @@ -237,6 +268,62 @@ public EtlJobSnapshot findOwned( return jobs.getFirst(); } + /** + * Lists one deterministic owner-scoped page of durable ETL jobs. + * + *

The query fetches one more row than the requested page size. That extra row is never + * returned; it only proves whether another page exists. The cursor records the final returned + * row's creation timestamp and UUID, while every query independently requires the authenticated + * principal hash.

+ * + * @param principalScope authenticated principal namespace + * @param cursor opaque following-page cursor, or {@code null} for the newest page + * @param pageSizeText canonical decimal page size, or {@code null} for the default of 50 + * @return immutable operator-safe page + * @throws EtlRequestException when principal, cursor, or page size validation fails + */ + @Transactional(readOnly = true) + public EtlJobPage listOwned( + @Nullable String principalScope, + @Nullable String cursor, + @Nullable String pageSizeText + ) { + String principalScopeHash = Sha256Digest.digest(validatePrincipalScope(principalScope)); + int pageSize = validatePageSize(pageSizeText); + PageCursor pageCursor = decodeCursor(cursor); + int fetchLimit = pageSize + 1; + + List queriedJobs; + if (pageCursor == null) { + queriedJobs = jdbcTemplate.query( + SELECT_OWNED_JOB_PAGE_SQL, + EtlJobService::mapSnapshotRow, + principalScopeHash, + fetchLimit + ); + } else { + Timestamp cursorTimestamp = Timestamp.from(pageCursor.createdAt()); + queriedJobs = jdbcTemplate.query( + SELECT_OWNED_JOB_PAGE_AFTER_CURSOR_SQL, + EtlJobService::mapSnapshotRow, + principalScopeHash, + cursorTimestamp, + cursorTimestamp, + pageCursor.jobRecordId(), + fetchLimit + ); + } + + boolean hasNextPage = queriedJobs.size() > pageSize; + List pageJobs = hasNextPage + ? List.copyOf(queriedJobs.subList(0, pageSize)) + : List.copyOf(queriedJobs); + String nextCursor = hasNextPage + ? encodeCursor(pageJobs.getLast()) + : null; + return new EtlJobPage(pageJobs, nextCursor); + } + private StoredJobRecord findSubmission(String principalScopeHash, String submissionKeyHash) { List jobs = jdbcTemplate.query( SELECT_SUBMISSION_SQL, @@ -257,6 +344,20 @@ private StoredJobRecord findSubmission(String principalScopeHash, String submiss return jobs.isEmpty() ? null : jobs.getFirst(); } + private static EtlJobSnapshot mapSnapshotRow( + java.sql.ResultSet resultSet, + int rowNumber + ) throws java.sql.SQLException { + return mapSnapshot( + resultSet.getObject("job_record_id", UUID.class), + resultSet.getString("job_status"), + resultSet.getInt("attempt_count"), + resultSet.getString("failure_code"), + resultSet.getTimestamp("created_at"), + resultSet.getTimestamp("updated_at") + ); + } + private static EtlJobSnapshot mapSnapshot( UUID jobRecordId, String jobStatus, @@ -283,6 +384,75 @@ private static EtlJobSnapshot mapSnapshot( ); } + private static int validatePageSize(@Nullable String pageSizeText) { + if (pageSizeText == null) { + return DEFAULT_JOB_PAGE_SIZE; + } + if (!JOB_PAGE_LIMIT_PROFILE.matcher(pageSizeText).matches()) { + throw new EtlRequestException(EtlRequestError.INVALID_JOB_PAGE_LIMIT); + } + int pageSize = Integer.parseInt(pageSizeText); + if (pageSize > MAX_JOB_PAGE_SIZE) { + throw new EtlRequestException(EtlRequestError.INVALID_JOB_PAGE_LIMIT); + } + return pageSize; + } + + @Nullable + private static PageCursor decodeCursor(@Nullable String cursor) { + if (cursor == null) { + return null; + } + if (cursor.isEmpty() || cursor.length() > MAX_JOB_PAGE_CURSOR_CHARACTERS) { + throw new EtlRequestException(EtlRequestError.INVALID_JOB_PAGE_CURSOR); + } + + final String decoded; + try { + byte[] cursorBytes = Base64.getUrlDecoder().decode(cursor); + decoded = new String(cursorBytes, StandardCharsets.UTF_8); + } catch (IllegalArgumentException exception) { + throw new EtlRequestException(EtlRequestError.INVALID_JOB_PAGE_CURSOR, exception); + } + + String[] cursorParts = decoded.split("\\|", -1); + if (cursorParts.length != 2) { + throw new EtlRequestException(EtlRequestError.INVALID_JOB_PAGE_CURSOR); + } + + final PageCursor pageCursor; + try { + pageCursor = new PageCursor( + Instant.parse(cursorParts[0]), + UUID.fromString(cursorParts[1]) + ); + } catch (DateTimeParseException | IllegalArgumentException exception) { + throw new EtlRequestException(EtlRequestError.INVALID_JOB_PAGE_CURSOR, exception); + } + if (!cursor.equals(encodeCursor(pageCursor.createdAt(), pageCursor.jobRecordId()))) { + throw new EtlRequestException(EtlRequestError.INVALID_JOB_PAGE_CURSOR); + } + return pageCursor; + } + + private static String encodeCursor(EtlJobSnapshot snapshot) { + EtlJobSnapshot requiredSnapshot = Objects.requireNonNull( + snapshot, + "snapshot must not be null" + ); + return encodeCursor(requiredSnapshot.createdAt(), requiredSnapshot.jobRecordId()); + } + + private static String encodeCursor(Instant createdAt, UUID jobRecordId) { + String cursorPayload = Objects.requireNonNull( + createdAt, + "createdAt must not be null" + ) + "|" + Objects.requireNonNull(jobRecordId, "jobRecordId must not be null"); + return Base64.getUrlEncoder() + .withoutPadding() + .encodeToString(cursorPayload.getBytes(StandardCharsets.UTF_8)); + } + private String validatePayload(@Nullable String requestPayload) { if (requestPayload == null) { throw new EtlRequestException(EtlRequestError.INVALID_JSON); @@ -386,4 +556,11 @@ private record StoredJobRecord(String requestDigest, EtlJobSnapshot snapshot) { Objects.requireNonNull(snapshot, "snapshot must not be null"); } } + + private record PageCursor(Instant createdAt, UUID jobRecordId) { + private PageCursor { + Objects.requireNonNull(createdAt, "createdAt must not be null"); + Objects.requireNonNull(jobRecordId, "jobRecordId must not be null"); + } + } } diff --git a/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java b/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java index 090afa87..a7786ee6 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java @@ -104,6 +104,24 @@ public enum EtlRequestError { "A durable ETL job with the same principal-scoped Idempotency-Key is being created." ), + /** The durable job page limit is absent from mightyETL's bounded canonical profile. */ + INVALID_JOB_PAGE_LIMIT( + HttpStatus.BAD_REQUEST, + "etl_invalid_job_page_limit", + "urn:mightyetl:problem:etl-invalid-job-page-limit", + "Invalid ETL job page limit", + "The job page limit must be a canonical integer from 1 through 100." + ), + + /** The durable job cursor is malformed, oversized, incomplete, or non-canonical. */ + INVALID_JOB_PAGE_CURSOR( + HttpStatus.BAD_REQUEST, + "etl_invalid_job_page_cursor", + "urn:mightyetl:problem:etl-invalid-job-page-cursor", + "Invalid ETL job page cursor", + "The job page cursor is invalid or no longer follows the supported opaque format." + ), + /** The requested job does not exist in the authenticated principal's namespace. */ JOB_NOT_FOUND( HttpStatus.NOT_FOUND, diff --git a/etl-service/src/main/resources/db/migration/V5__add_etl_job_owner_pagination_index.sql b/etl-service/src/main/resources/db/migration/V5__add_etl_job_owner_pagination_index.sql new file mode 100644 index 00000000..66802e4c --- /dev/null +++ b/etl-service/src/main/resources/db/migration/V5__add_etl_job_owner_pagination_index.sql @@ -0,0 +1,10 @@ +-- Support deterministic newest-first keyset pagination inside one hashed principal namespace. +-- 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_owner_pagination_index + ON etl_job_records ( + principal_scope_hash, + created_at DESC, + job_record_id DESC + ); diff --git a/etl-service/src/main/resources/db/migration/V5__add_etl_job_owner_pagination_index.sql.conf b/etl-service/src/main/resources/db/migration/V5__add_etl_job_owner_pagination_index.sql.conf new file mode 100644 index 00000000..73bd53a1 --- /dev/null +++ b/etl-service/src/main/resources/db/migration/V5__add_etl_job_owner_pagination_index.sql.conf @@ -0,0 +1 @@ +executeInTransaction=false 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 698acc9c..0f94c7e5 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java @@ -50,6 +50,48 @@ void migrationReservesStableWorkerStatesAndRequiresTerminalPayloadClearing() thr assertTrue(migration.contains("request_payload IS NULL")); } + @Test + void paginationMigrationUsesTheOwnerAndCompleteStableOrderingKey() throws IOException { + String migration = read( + "etl-service/src/main/resources/db/migration/" + + "V5__add_etl_job_owner_pagination_index.sql" + ).replaceAll("\\s+", " "); + + assertTrue(migration.contains( + "CREATE INDEX CONCURRENTLY etl_job_owner_pagination_index" + )); + assertTrue(migration.contains( + "ON etl_job_records ( principal_scope_hash, created_at DESC, job_record_id DESC )" + )); + assertFalse(migration.contains(" OFFSET ")); + assertFalse(migration.contains("principal_name")); + } + + @Test + void paginationIndexMigrationDoesNotBlockProductionWriters() throws IOException { + String migrationPath = "etl-service/src/main/resources/db/migration/" + + "V5__add_etl_job_owner_pagination_index.sql"; + String configurationPath = migrationPath + ".conf"; + String migration = read(migrationPath).replaceAll("\\s+", " "); + Path configuration = projectRoot().resolve(configurationPath); + String configurationText = Files.exists(configuration) + ? Files.readString(configuration, StandardCharsets.UTF_8).trim() + : ""; + + assertTrue( + migration.contains("CREATE INDEX CONCURRENTLY etl_job_owner_pagination_index"), + "the production pagination index must not block concurrent inserts or updates" + ); + assertTrue( + Files.exists(configuration), + "Flyway requires a per-script configuration for non-transactional PostgreSQL DDL" + ); + assertTrue( + configurationText.contains("executeInTransaction=false"), + "CREATE INDEX CONCURRENTLY cannot run inside Flyway's default transaction" + ); + } + @Test void runbookDocumentsAcceptedSemanticsOwnershipAndLeaseFencedExecution() throws IOException { String runbook = read("docs/etl/durable-job-intake.md").replaceAll("\\s+", " "); @@ -70,10 +112,34 @@ void runbookDocumentsAcceptedSemanticsOwnershipAndLeaseFencedExecution() throws assertTrue(runbook.contains("xtrmetl.*")); } + @Test + void runbookDocumentsOwnerScopedKeysetPaginationAndRollback() throws IOException { + String runbook = read("docs/etl/durable-job-intake.md").replaceAll("\\s+", " "); + + assertTrue(runbook.contains("GET /api/etl/jobs?limit=50")); + assertTrue(runbook.contains("created_at DESC, job_record_id DESC")); + assertTrue(runbook.contains("strict tuple boundary")); + assertTrue(runbook.contains("Link: <")); + assertTrue(runbook.contains("rel=\"next\"")); + assertTrue(runbook.contains("etl_invalid_job_page_limit")); + assertTrue(runbook.contains("etl_invalid_job_page_cursor")); + assertTrue(runbook.contains("V5__add_etl_job_owner_pagination_index.sql")); + assertTrue(runbook.contains("executeInTransaction=false")); + assertTrue(runbook.contains( + "DROP INDEX CONCURRENTLY etl_job_owner_pagination_index" + )); + assertTrue(runbook.contains("RFC 8288")); + } + private static String read(String relativePath) throws IOException { return Files.readString(projectRoot().resolve(relativePath), StandardCharsets.UTF_8); } + /** + * Finds the reactor root from either repository-root or module-local Maven execution. + * + * @return repository root containing the Maven reactor + */ private static Path projectRoot() { Path current = Paths.get(System.getProperty("user.dir")).toAbsolutePath(); Path lastPomParent = null; diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPageModelTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPageModelTest.java new file mode 100644 index 00000000..73958628 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPageModelTest.java @@ -0,0 +1,70 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Defines the immutable operator-safe page model used by durable job discovery. + */ +class EtlJobPageModelTest { + + @Test + void copiesSnapshotsAndMapsOnlyOperatorSafeStatusFields() { + EtlJobSnapshot snapshot = snapshot(); + List mutableSnapshots = new ArrayList<>(); + mutableSnapshots.add(snapshot); + + EtlJobPage page = new EtlJobPage(mutableSnapshots, "opaque_cursor"); + mutableSnapshots.clear(); + EtlJobPageResponse response = EtlJobPageResponse.from(page); + + assertEquals(List.of(snapshot), page.jobs()); + assertNotSame(mutableSnapshots, page.jobs()); + assertEquals("opaque_cursor", page.nextCursor()); + assertEquals(1, response.jobs().size()); + assertEquals(snapshot.jobRecordId(), response.jobs().getFirst().jobRecordId()); + assertEquals(snapshot.jobStatus(), response.jobs().getFirst().jobStatus()); + assertEquals(snapshot.attemptCount(), response.jobs().getFirst().attemptCount()); + assertNull(response.jobs().getFirst().failureCode()); + assertEquals(snapshot.createdAt(), response.jobs().getFirst().createdAt()); + assertEquals(snapshot.updatedAt(), response.jobs().getFirst().updatedAt()); + assertEquals("opaque_cursor", response.nextCursor()); + assertThrows(UnsupportedOperationException.class, () -> page.jobs().clear()); + assertThrows(UnsupportedOperationException.class, () -> response.jobs().clear()); + } + + @Test + void rejectsNullCollectionsAndNullPageInputs() { + assertThrows(NullPointerException.class, () -> new EtlJobPage(null, null)); + assertThrows(NullPointerException.class, () -> new EtlJobPageResponse(null, null)); + assertThrows(NullPointerException.class, () -> EtlJobPageResponse.from(null)); + assertThrows( + NullPointerException.class, + () -> new EtlJobPage(List.of((EtlJobSnapshot) null), null) + ); + assertThrows( + NullPointerException.class, + () -> new EtlJobPageResponse(List.of((EtlJobStatusResponse) null), null) + ); + } + + private static EtlJobSnapshot snapshot() { + return new EtlJobSnapshot( + UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef"), + EtlJobStatus.SUCCEEDED, + 2, + null, + Instant.parse("2026-08-05T01:00:00Z"), + Instant.parse("2026-08-05T01:00:05Z") + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPaginationControllerTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPaginationControllerTest.java new file mode 100644 index 00000000..2efb43cb --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPaginationControllerTest.java @@ -0,0 +1,161 @@ +package com.xtrmetl.etl.job; + +import com.xtrmetl.etl.controller.EtlApiProblemHandler; +import com.xtrmetl.etl.controller.EtlJobController; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +import java.security.Principal; +import java.time.Instant; +import java.util.List; +import java.util.UUID; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +/** + * Defines the authenticated HTTP contract for owner-scoped durable job pagination. + */ +class EtlJobPaginationControllerTest { + + private static final String JOBS_PATH = "/api/etl/jobs"; + private static final Principal PRINCIPAL = () -> "tenant_alpha"; + + private EtlJobService etlJobService; + private MockMvc mockMvc; + + @BeforeEach + void setUp() { + etlJobService = mock(EtlJobService.class); + mockMvc = MockMvcBuilders + .standaloneSetup(new EtlJobController(etlJobService)) + .setControllerAdvice(new EtlApiProblemHandler()) + .build(); + } + + @Test + void listsOwnedJobsAndAdvertisesOnlyTheExistingNextPage() throws Exception { + EtlJobSnapshot snapshot = snapshot(); + when(etlJobService.listOwned("tenant_alpha", "current_cursor", "2")) + .thenReturn(new EtlJobPage(List.of(snapshot), "next_cursor")); + + mockMvc.perform(get(JOBS_PATH) + .principal(PRINCIPAL) + .queryParam("cursor", "current_cursor") + .queryParam("limit", "2")) + .andExpect(status().isOk()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(header().string( + "Link", + "; rel=\"next\"" + )) + .andExpect(jsonPath("$.jobs.length()").value(1)) + .andExpect(jsonPath("$.jobs[0].jobRecordId").value( + snapshot.jobRecordId().toString() + )) + .andExpect(jsonPath("$.jobs[0].jobStatus").value("SUCCEEDED")) + .andExpect(jsonPath("$.jobs[0].attemptCount").value(2)) + .andExpect(jsonPath("$.jobs[0].failureCode").doesNotExist()) + .andExpect(jsonPath("$.jobs[0].createdAt").value("2026-08-05T01:00:00Z")) + .andExpect(jsonPath("$.jobs[0].updatedAt").value("2026-08-05T01:00:05Z")) + .andExpect(jsonPath("$.jobs[0].requestPayload").doesNotExist()) + .andExpect(jsonPath("$.jobs[0].principalScopeHash").doesNotExist()) + .andExpect(jsonPath("$.nextCursor").value("next_cursor")); + + verify(etlJobService).listOwned("tenant_alpha", "current_cursor", "2"); + } + + @Test + void usesTheDocumentedDefaultLimitInTheNextPageLink() throws Exception { + when(etlJobService.listOwned("tenant_alpha", null, null)) + .thenReturn(new EtlJobPage(List.of(snapshot()), "next_cursor")); + + mockMvc.perform(get(JOBS_PATH).principal(PRINCIPAL)) + .andExpect(status().isOk()) + .andExpect(header().string( + "Link", + "; rel=\"next\"" + )); + + verify(etlJobService).listOwned("tenant_alpha", null, null); + } + + @Test + void omitsTheNextLinkAndCursorForTheTerminalPage() throws Exception { + when(etlJobService.listOwned("tenant_alpha", null, null)) + .thenReturn(new EtlJobPage(List.of(), null)); + + mockMvc.perform(get(JOBS_PATH).principal(PRINCIPAL)) + .andExpect(status().isOk()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(header().doesNotExist("Link")) + .andExpect(jsonPath("$.jobs.length()").value(0)) + .andExpect(jsonPath("$.nextCursor").doesNotExist()); + + verify(etlJobService).listOwned("tenant_alpha", null, null); + } + + @Test + void preservesTypedListValidationFailures() throws Exception { + when(etlJobService.listOwned("tenant_alpha", null, "0")) + .thenThrow(new EtlRequestException(EtlRequestError.INVALID_JOB_PAGE_LIMIT)); + + mockMvc.perform(get(JOBS_PATH) + .principal(PRINCIPAL) + .queryParam("limit", "0")) + .andExpect(status().isBadRequest()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_invalid_job_page_limit")); + + verify(etlJobService).listOwned("tenant_alpha", null, "0"); + } + + @Test + void mapsUnexpectedListFailuresWithoutLeakingMessages() throws Exception { + when(etlJobService.listOwned("tenant_alpha", null, null)) + .thenThrow(new IllegalStateException("secret runtime detail")); + + mockMvc.perform(get(JOBS_PATH).principal(PRINCIPAL)) + .andExpect(status().isInternalServerError()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_internal_error")) + .andExpect(jsonPath("$.detail").value( + "The ETL request could not be processed." + )); + + verify(etlJobService).listOwned("tenant_alpha", null, null); + } + + @Test + void requiresAuthenticationBeforeListingJobs() throws Exception { + mockMvc.perform(get(JOBS_PATH)) + .andExpect(status().isUnauthorized()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value( + "etl_idempotency_principal_required" + )); + + verifyNoInteractions(etlJobService); + } + + private static EtlJobSnapshot snapshot() { + return new EtlJobSnapshot( + UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef"), + EtlJobStatus.SUCCEEDED, + 2, + null, + Instant.parse("2026-08-05T01:00:00Z"), + Instant.parse("2026-08-05T01:00:05Z") + ); + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPaginationServiceIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPaginationServiceIntegrationTest.java new file mode 100644 index 00000000..57fb6646 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPaginationServiceIntegrationTest.java @@ -0,0 +1,253 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import com.xtrmetl.etl.service.EtlRequestLock; +import 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.nio.charset.StandardCharsets; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Defines stable keyset traversal, bounded validation, and tenant isolation for job discovery. + */ +@SpringJUnitConfig(EtlJobPaginationServiceIntegrationTest.TestConfiguration.class) +class EtlJobPaginationServiceIntegrationTest { + + private static final UUID OLDEST_JOB = UUID.fromString( + "00000000-0000-0000-0000-000000000000" + ); + private static final UUID TIED_LOWER_JOB = UUID.fromString( + "00000000-0000-0000-0000-000000000001" + ); + private static final UUID TIED_HIGHER_JOB = UUID.fromString( + "00000000-0000-0000-0000-000000000002" + ); + private static final UUID NEWEST_JOB = UUID.fromString( + "00000000-0000-0000-0000-000000000003" + ); + private static final UUID FOREIGN_JOB = UUID.fromString( + "00000000-0000-0000-0000-000000000004" + ); + + private final EtlJobService etlJobService; + private final JdbcTemplate jdbcTemplate; + + @Autowired + EtlJobPaginationServiceIntegrationTest( + EtlJobService etlJobService, + JdbcTemplate jdbcTemplate + ) { + this.etlJobService = etlJobService; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createJobTable() { + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_job_records"); + jdbcTemplate.execute(""" + CREATE TABLE etl_job_records ( + job_record_id UUID PRIMARY KEY, + principal_scope_hash CHAR(64) NOT NULL, + submission_key_hash CHAR(64) NOT NULL, + request_digest CHAR(64) NOT NULL, + request_payload VARCHAR(8192), + job_status VARCHAR(32) NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + failure_code VARCHAR(128), + created_at TIMESTAMP WITH TIME ZONE NOT NULL, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL, + CONSTRAINT etl_job_submission_scope_unique + UNIQUE (principal_scope_hash, submission_key_hash) + ) + """); + } + + @Test + void traversesAnUnchangedTenantDatasetWithoutDuplicatesOrOmissions() { + insertJob(OLDEST_JOB, "tenant_alpha", Instant.parse("2026-08-05T01:00:00Z")); + insertJob(TIED_LOWER_JOB, "tenant_alpha", Instant.parse("2026-08-05T02:00:00Z")); + insertJob(TIED_HIGHER_JOB, "tenant_alpha", Instant.parse("2026-08-05T02:00:00Z")); + insertJob(NEWEST_JOB, "tenant_alpha", Instant.parse("2026-08-05T03:00:00Z")); + insertJob(FOREIGN_JOB, "tenant_beta", Instant.parse("2026-08-05T04:00:00Z")); + + EtlJobPage firstPage = etlJobService.listOwned("tenant_alpha", null, "2"); + assertEquals(List.of(NEWEST_JOB, TIED_HIGHER_JOB), ids(firstPage)); + assertNotNull(firstPage.nextCursor()); + + EtlJobPage secondPage = etlJobService.listOwned( + "tenant_alpha", + firstPage.nextCursor(), + "2" + ); + assertEquals(List.of(TIED_LOWER_JOB, OLDEST_JOB), ids(secondPage)); + assertNull(secondPage.nextCursor()); + + List traversed = new ArrayList<>(ids(firstPage)); + traversed.addAll(ids(secondPage)); + assertEquals( + List.of(NEWEST_JOB, TIED_HIGHER_JOB, TIED_LOWER_JOB, OLDEST_JOB), + traversed + ); + assertFalse(traversed.contains(FOREIGN_JOB)); + } + + @Test + void usesTheBoundedDefaultAndReturnsAnEmptyTerminalPage() { + EtlJobPage page = etlJobService.listOwned("tenant_alpha", null, null); + + assertTrue(page.jobs().isEmpty()); + assertNull(page.nextCursor()); + } + + @Test + void rejectsMalformedLimitsAndCursorsBeforeDatabaseAccess() { + jdbcTemplate.execute("DROP TABLE etl_job_records"); + + for (String invalidLimit : List.of("0", "-1", "101", "abc", "01", " 2")) { + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> etlJobService.listOwned("tenant_alpha", null, invalidLimit) + ); + assertEquals(EtlRequestError.INVALID_JOB_PAGE_LIMIT, exception.error()); + } + + List invalidCursors = List.of( + "", + "!", + encode("missing_separator"), + encode("not-an-instant|00000000-0000-0000-0000-000000000000"), + encode("2026-08-05T01:00:00Z|not-a-uuid"), + encode("2026-08-05T01:00:00Z|00000000-0000-0000-0000-000000000000") + "=", + encode("2026-08-05T01:00:00.000Z|00000000-0000-0000-0000-000000000000"), + "a".repeat(193) + ); + for (String invalidCursor : invalidCursors) { + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> etlJobService.listOwned("tenant_alpha", invalidCursor, "2") + ); + assertEquals(EtlRequestError.INVALID_JOB_PAGE_CURSOR, exception.error()); + } + + EtlRequestException missingPrincipal = assertThrows( + EtlRequestException.class, + () -> etlJobService.listOwned(null, null, "2") + ); + assertEquals(EtlRequestError.IDEMPOTENCY_PRINCIPAL_REQUIRED, missingPrincipal.error()); + } + + private void insertJob(UUID jobRecordId, String principalScope, Instant createdAt) { + String identity = jobRecordId.toString(); + 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, + Sha256Digest.digest(principalScope), + Sha256Digest.digest("submission:" + identity), + Sha256Digest.digest("payload:" + identity), + "[{\"id\":\"" + identity + "\"}]", + Timestamp.from(createdAt), + Timestamp.from(createdAt) + ); + } + + private static List ids(EtlJobPage page) { + return page.jobs().stream().map(EtlJobSnapshot::jobRecordId).toList(); + } + + private static String encode(String cursorPayload) { + return Base64.getUrlEncoder() + .withoutPadding() + .encodeToString(cursorPayload.getBytes(StandardCharsets.UTF_8)); + } + + /** + * Minimal transaction-enabled test context for job pagination. + */ + @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 + EtlJobService etlJobService( + JdbcTemplate jdbcTemplate, + ObjectMapper objectMapper, + EtlBatchProperties properties, + EtlRequestLock requestLock + ) { + return new EtlJobService(jdbcTemplate, objectMapper, properties, requestLock); + } + } +}