From 4a555d62d950cf6d812866a89d1e74de1d88cb65 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:10:02 +0900 Subject: [PATCH 01/14] test(etl): re-establish pagination model RED --- .../xtrmetl/etl/job/EtlJobPageModelTest.java | 70 +++++++++++++++++++ 1 file changed, 70 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPageModelTest.java 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") + ); + } +} From 43593b2c1ce0d77f1395a7ebb7fe4b87c1c16413 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:13:29 +0900 Subject: [PATCH 02/14] feat(etl): add immutable durable job page model --- .../java/com/xtrmetl/etl/job/EtlJobPage.java | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobPage.java 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")); + } +} From e1d8f7e5abc7d9e8c3d5ad3db8dbd69160a5035e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:13:54 +0900 Subject: [PATCH 03/14] feat(etl): add durable job page response --- .../xtrmetl/etl/job/EtlJobPageResponse.java | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobPageResponse.java 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()); + } +} From cde3c707e84ad2e5cd3e46a8bd0fd2f8cb3dfd1e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:14:51 +0900 Subject: [PATCH 04/14] test(etl): re-establish pagination service RED --- ...tlJobPaginationServiceIntegrationTest.java | 253 ++++++++++++++++++ 1 file changed, 253 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPaginationServiceIntegrationTest.java 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); + } + } +} From 8c8016ba05da494d2549807d89374c975c84345f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:16:32 +0900 Subject: [PATCH 05/14] feat(etl): add pagination request errors --- .../xtrmetl/etl/service/EtlRequestError.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) 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..45fb2f6a 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 supplied durable-job list page size is not a canonical integer in the supported range. */ + 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 ETL job page limit must be a canonical integer from 1 through 100." + ), + + /** The supplied durable-job list cursor is malformed, non-canonical, or unsupported. */ + 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 ETL job page cursor is invalid or uses an unsupported opaque format." + ), + /** The requested job does not exist in the authenticated principal's namespace. */ JOB_NOT_FOUND( HttpStatus.NOT_FOUND, From d228f1bca4e3c6b4f5409a447f5c048b2b0b1aea Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:19:11 +0900 Subject: [PATCH 06/14] feat(etl): add owner-scoped cursor pagination --- .../com/xtrmetl/etl/job/EtlJobService.java | 183 +++++++++++++++++- 1 file changed, 180 insertions(+), 3 deletions(-) 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"); + } + } } From a05b9ccd19db625c02e6167fd1cb077137430a7b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:19:50 +0900 Subject: [PATCH 07/14] test(etl): re-establish pagination controller RED --- .../job/EtlJobPaginationControllerTest.java | 161 ++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobPaginationControllerTest.java 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") + ); + } +} From ff9af4c69eddb6e3c71612b580d2932d1bc82ad5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:20:36 +0900 Subject: [PATCH 08/14] feat(etl): expose owner-scoped job pagination --- .../etl/controller/EtlJobController.java | 63 +++++++++++++++++-- 1 file changed, 57 insertions(+), 6 deletions(-) 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. * From 4bd27f205b1cab1210f62e33f87d105e18183606 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:23:16 +0900 Subject: [PATCH 09/14] test(etl): require nonblocking pagination index migration --- .../job/EtlJobMigrationDocumentationTest.java | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) 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 9ea22672..1197bf69 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 runbookDocumentsAcceptedSemanticsOwnershipAndActiveWorkerBoundary() throws IOException { String runbook = read("docs/etl/durable-job-intake.md").replaceAll("\\s+", " "); From 266856b8a3e950534afe00abdb14f2b0da804e0b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:29:40 +0900 Subject: [PATCH 10/14] fix(etl): restore concurrent pagination index migration --- .../V5__add_etl_job_owner_pagination_index.sql | 10 ++++++++++ 1 file changed, 10 insertions(+) create mode 100644 etl-service/src/main/resources/db/migration/V5__add_etl_job_owner_pagination_index.sql 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 + ); From 6e9fa4eff0459ed083f628c9682bf0bde246bac9 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:29:58 +0900 Subject: [PATCH 11/14] fix(etl): run pagination index migration outside transaction --- .../db/migration/V5__add_etl_job_owner_pagination_index.sql.conf | 1 + 1 file changed, 1 insertion(+) create mode 100644 etl-service/src/main/resources/db/migration/V5__add_etl_job_owner_pagination_index.sql.conf 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 From f2ba86de2326619aaf8e6bef69497c192cb09a7c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:33:16 +0900 Subject: [PATCH 12/14] test(etl): require pagination operator contract --- .../job/EtlJobMigrationDocumentationTest.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) 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 1197bf69..b7c75107 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 @@ -114,6 +114,25 @@ void runbookDocumentsAcceptedSemanticsOwnershipAndActiveWorkerBoundary() throws assertTrue(runbook.contains("xtrmetl.etl.jobs.worker.enabled=true")); } + @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")); + } + @Test void changelogRecordsLeaseFencedDurableWorkerExecution() throws IOException { String changelog = read("CHANGELOG.md").replaceAll("\\s+", " "); From ef3c24006c4c37b4b87a28664e8b7f3a4b1d744b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:35:55 +0900 Subject: [PATCH 13/14] docs(etl): document owner-scoped pagination rollout --- docs/etl/durable-job-intake.md | 103 +++++++++++++++++++++++++++++---- 1 file changed, 91 insertions(+), 12 deletions(-) diff --git a/docs/etl/durable-job-intake.md b/docs/etl/durable-job-intake.md index 05e2b958..2ce42c45 100644 --- a/docs/etl/durable-job-intake.md +++ b/docs/etl/durable-job-intake.md @@ -3,10 +3,11 @@ ## Scope `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 +execution is 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. +transitions. Authenticated operators can list recent jobs in their own principal namespace through +deterministic keyset pagination. 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 @@ -70,6 +71,44 @@ A retry that resolves to the same durable resource returns the same job identifi transaction-level submission lock returns `409 etl_job_submission_in_progress` rather than waiting without a client-visible bound. +## 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. Rows are ordered by +`created_at DESC, job_record_id DESC`; the UUID is a deterministic tie-breaker when multiple jobs +share one database timestamp. The default page size is 50 and the canonical accepted range is 1 +through 100. Values such as `0`, `101`, `01`, signed values, whitespace-padded values, or 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 never returned; +it only proves that another page exists. When a following page exists, the response body contains an +opaque cursor and RFC 8288 Web Linking advertises the same continuation target: + +```http +HTTP/1.1 200 OK +Cache-Control: no-store +Link: ; rel="next" +Content-Type: application/json +``` + +The cursor is a canonical unpadded Base64 URL encoding of the final returned creation timestamp and +job identifier. Clients must treat it as opaque. Each next-page query independently binds the +current authenticated principal hash and applies a strict tuple boundary equivalent to “older +creation timestamp, or the same timestamp with a lower UUID.” Cursor contents never grant authority +and contain no payload, raw principal, submission key, or internal 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 the +next cursor and the `Link` header. + +Pagination guarantees no duplicate or omitted rows while traversing an unchanged dataset. Concurrent +insertions are visible according to their ordering position; an operational cursor is not a frozen +snapshot. Consumers that require a legally frozen audit set must use an explicit transactional or +warehouse snapshot. + ## Read job status ```http @@ -143,14 +182,18 @@ 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`; 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: +Flyway migrations use descriptive multi-word `snake_case` objects: -- an opaque UUID job identifier; -- SHA-256 hashes of the principal scope, semantic submission key, and exact JSON text; -- the request payload only while the job remains nonterminal; -- status, attempt, failure, lease-owner/token/expiry, and lifecycle timestamp fields. +- `V2__create_etl_job_records.sql` creates `etl_job_records` and the principal-scoped submission + uniqueness contract; +- worker migrations add the lease-fencing fields and concurrent claim-eligibility index; and +- `V5__add_etl_job_owner_pagination_index.sql` creates `etl_job_owner_pagination_index` on + `principal_scope_hash`, `created_at DESC`, and `job_record_id DESC`, matching the owner-scoped + keyset ordering contract. + +The database stores an opaque UUID job identifier; SHA-256 hashes of principal scope, semantic +submission key, and exact JSON text; the request payload only while nonterminal; and lifecycle, +attempt, failure, lease-owner/token/expiry, and timestamp fields. 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 @@ -162,6 +205,30 @@ payload is sensitive operational data and inherits the classification of its sou job is `PENDING` or `RUNNING`, operators must protect it with database access control, encryption, backup, and retention policy appropriate to the underlying records. +## Pagination migration and rollback + +The V5 owner-pagination index uses PostgreSQL `CREATE INDEX CONCURRENTLY` so inserts, updates, and +deletes remain available while PostgreSQL builds the index. Its migration-local companion file +`V5__add_etl_job_owner_pagination_index.sql.conf` contains `executeInTransaction=false` because +PostgreSQL rejects concurrent index creation inside a transaction block. + +Concurrent index creation can wait for transactions and can leave an invalid index after a failed +build. Production rollout therefore requires catalog inspection of index validity and readiness, +plus representative monitoring of duration, I/O, replication lag, and transaction age. A matching +object name alone is not proof that the index is usable. + +Older application binaries ignore this additive index. After rolling back binaries that depend on +the list access path, remove the database object outside a transaction block only after list traffic +is withdrawn and an execution-plan review confirms the operational boundary: + +```sql +DROP INDEX CONCURRENTLY etl_job_owner_pagination_index; +``` + +Dropping the index does not change query semantics, but it can turn an owner-scoped list operation +into an unacceptable scan, so removal is an explicit performance rollback rather than an emergency +schema shortcut. + ## Operational boundary Enabling intake alone still does not start background processing; enabling the worker alone does not @@ -172,7 +239,7 @@ safe by disabling the worker property: existing durable rows remain in PostgreSQ 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 +finite worker telemetry, and owner-scoped keyset 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. @@ -180,14 +247,18 @@ of this boundary until their own exact-head gates pass. - 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 8288 defines the Web Linking model and HTTP `Link` header used for the optional next-page + relationship. - 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. +- PostgreSQL 18 documents explicit ordering, row-locking semantics, multicolumn B-tree behavior, and + the availability and recovery trade-offs of concurrent index construction. +- Flyway script configuration supports migration-local transaction overrides required by PostgreSQL + DDL that cannot execute in a transaction block. ### References @@ -196,9 +267,17 @@ of this boundary until their own exact-head gates pass. - Jena, J., & Dalal, S. (2025). *The Idempotency-Key HTTP header field* (draft-ietf-httpapi-idempotency-key-header-07, expired April 18, 2026). Internet Engineering Task Force. https://datatracker.ietf.org/doc/draft-ietf-httpapi-idempotency-key-header/ +- 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 - 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). *CREATE INDEX*. PostgreSQL 18 documentation. + https://www.postgresql.org/docs/18/sql-createindex.html +- PostgreSQL Global Development Group. (2026). *Multicolumn indexes*. PostgreSQL 18 documentation. + https://www.postgresql.org/docs/18/indexes-multicolumn.html - PostgreSQL Global Development Group. (2026). *SELECT*. PostgreSQL 18 documentation. https://www.postgresql.org/docs/18/sql-select.html +- Redgate Software. (2026). *Flyway script configuration*. Flyway documentation. + https://documentation.red-gate.com/flyway/reference/script-configuration From 2a37789e651316e9bf29cbbd68902aa827a50c63 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 9 Aug 2026 12:39:19 +0900 Subject: [PATCH 14/14] docs(changelog): record owner-scoped pagination --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f0b446e..3b40aeac 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 +- 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 pagination index uses PostgreSQL `CREATE INDEX CONCURRENTLY` with migration-local Flyway `executeInTransaction=false`, preserving production writers while documenting invalid-index recovery and concurrent rollback. - Scheduled OpenCode maintenance now performs root-cause analysis, tests remediation feasibility against live authority, protection, resource, dependency, path-ownership, and writer-lease constraints, executes and verifies the best safe option available now, and continues exactly one independent bounded mightyETL slice from protected `develop` when only external blockers remain; invalid stacks and separately leased repositories stay untouched. - Container builds now pin Maven and Eclipse Temurin base-image tags to reviewed SHA-256 digests, with a fail-first contract test preventing mutable registry tags from re-entering the Dockerfile. - The model-executing hourly OpenCode maintenance job now has read-only issue access; fail-first workflow-contract coverage proves `issues: write` is unnecessary while preserving issue and roadmap inspection. @@ -34,6 +36,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`. - Test-first doctoring for external-wait progress, root-cause analysis, realistic remediation feasibility, exact post-action verification, source-actionable pull-request classification, invalid-stack isolation, and read-only dependency leases in `docs/doctoring/hourly-opencode-nonblocking-progress-evidence.md`. - Permanent fail-first exact-head workflow contracts and authoritative evidence in `docs/doctoring/exact-head-source-workflow-evidence.md`, including observed synthetic-merge checkout behavior, cross-platform source identity assertions, the rejected ignored Dependency Review ref-override experiment, corrective pull-request event-endpoint semantics, least-privilege boundaries, stack invalidation rules, rollback prohibition, and APA 7th GitHub references. - A separate fail-closed hourly OpenCode maintenance workflow pinned to OpenCode 1.18.13 and `nvidia/deepseek-ai/deepseek-v4-pro`, using only the existing `NVIDIA_NIM_API_KEY` through OpenCode's `NVIDIA_API_KEY` provider variable while preserving the independent review agent and deterministic merge-disposition workflow.