From f50be481b2b19e6fb566e5430f2d6aac6437fdd4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 07:37:59 +0900 Subject: [PATCH 01/18] test(etl): prove cancellation result boundary on repaired stack --- .../etl/job/EtlJobCancellationTest.java | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationTest.java diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationTest.java new file mode 100644 index 00000000..7d23140d --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationTest.java @@ -0,0 +1,45 @@ +package com.xtrmetl.etl.job; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Covers immutable cancellation-result validation and replay semantics. + */ +class EtlJobCancellationTest { + + @Test + void acceptsOnlyCancelledOperatorSafeSnapshots() { + EtlJobSnapshot cancelledSnapshot = snapshot(EtlJobStatus.CANCELLED); + + EtlJobCancellation first = new EtlJobCancellation(cancelledSnapshot, false); + EtlJobCancellation replay = new EtlJobCancellation(cancelledSnapshot, true); + + assertEquals(cancelledSnapshot, first.snapshot()); + assertFalse(first.replayed()); + assertTrue(replay.replayed()); + assertThrows(NullPointerException.class, () -> new EtlJobCancellation(null, false)); + assertThrows( + IllegalArgumentException.class, + () -> new EtlJobCancellation(snapshot(EtlJobStatus.PENDING), false) + ); + } + + private static EtlJobSnapshot snapshot(EtlJobStatus status) { + return new EtlJobSnapshot( + UUID.fromString("75f61ec2-4f96-49e4-bf8e-66e2b75fb175"), + status, + 1, + null, + Instant.parse("2026-08-06T00:00:00Z"), + Instant.parse("2026-08-06T00:00:01Z") + ); + } +} From 7840701ea88a5c83f2b1acbbb70625d27f3ed69c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 07:40:44 +0900 Subject: [PATCH 02/18] feat(etl): add cancellation result contract --- .../xtrmetl/etl/job/EtlJobCancellation.java | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobCancellation.java diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobCancellation.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobCancellation.java new file mode 100644 index 00000000..9f8bf074 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobCancellation.java @@ -0,0 +1,32 @@ +package com.xtrmetl.etl.job; + +import java.util.Objects; + +/** + * Reports one committed or replayed owner-scoped durable job cancellation. + * + *

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

+ * + * @param snapshot cancelled owner-authorized job representation + * @param replayed {@code true} when the same normalized cancellation key had already committed + */ +public record EtlJobCancellation(EtlJobSnapshot snapshot, boolean replayed) { + + /** + * Validates the immutable cancellation result. + * + * @param snapshot cancelled owner-authorized job representation + * @param replayed whether this response proves an earlier identical cancellation + */ + public EtlJobCancellation { + EtlJobSnapshot requiredSnapshot = Objects.requireNonNull( + snapshot, + "snapshot must not be null" + ); + if (requiredSnapshot.jobStatus() != EtlJobStatus.CANCELLED) { + throw new IllegalArgumentException("snapshot must be in CANCELLED state"); + } + } +} From 759a365bd068e11bfdcc318b192179c81d97738f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 07:41:08 +0900 Subject: [PATCH 03/18] feat(etl): add terminal cancelled status --- .../main/java/com/xtrmetl/etl/job/EtlJobStatus.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java index 8e6fdd2c..8ae84f44 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobStatus.java @@ -3,10 +3,9 @@ /** * Stable lifecycle states exposed by the asynchronous ETL job resource. * - *

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

+ *

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

*/ public enum EtlJobStatus { @@ -20,5 +19,8 @@ public enum EtlJobStatus { SUCCEEDED, /** The job reached a terminal failure and the retained payload was cleared. */ - FAILED + FAILED, + + /** The authenticated owner cancelled the job and invalidated any active lease. */ + CANCELLED } From 9f1743278b85bbe33736aae50e79c444729ecd4b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 07:44:20 +0900 Subject: [PATCH 04/18] test(etl): prove cancellation service boundary on repaired stack --- ...EtlJobCancellationServiceBoundaryTest.java | 170 ++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationServiceBoundaryTest.java diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationServiceBoundaryTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationServiceBoundaryTest.java new file mode 100644 index 00000000..c955b0b1 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationServiceBoundaryTest.java @@ -0,0 +1,170 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** + * Covers fail-closed cancellation validation, transaction, and race classification boundaries. + */ +class EtlJobCancellationServiceBoundaryTest { + + private static final UUID JOB_RECORD_ID = UUID.fromString( + "2f4e2926-03dd-461f-873a-a70ad2256680" + ); + private static final String CANCELLATION_KEY = + "5d09cd43-50d1-4b82-a94f-a43f5bc6e56b"; + + @AfterEach + void clearSyntheticTransactionState() { + TransactionSynchronizationManager.clear(); + } + + @Test + void refusesCancellationWithoutAnActualTransactionBeforeDatabaseWork() { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + EtlJobService service = service(jdbcTemplate); + + IllegalStateException exception = assertThrows( + IllegalStateException.class, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, "tenant_alpha") + ); + + assertEquals( + "Durable ETL job cancellation requires an active transaction", + exception.getMessage() + ); + verifyNoInteractions(jdbcTemplate); + } + + @Test + void rejectsInvalidCancellationIdentityBeforeDatabaseWork() { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + EtlJobService service = service(jdbcTemplate); + + assertThrows( + NullPointerException.class, + () -> service.cancelOwned(null, CANCELLATION_KEY, "tenant_alpha") + ); + assertError( + EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED, + () -> service.cancelOwned(JOB_RECORD_ID, null, "tenant_alpha") + ); + assertError( + EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED, + () -> service.cancelOwned(JOB_RECORD_ID, "unsafe key", "tenant_alpha") + ); + assertError( + EtlRequestError.IDEMPOTENCY_PRINCIPAL_REQUIRED, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, null) + ); + assertError( + EtlRequestError.IDEMPOTENCY_PRINCIPAL_REQUIRED, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, " ") + ); + verifyNoInteractions(jdbcTemplate); + } + + @Test + void reportsAnActiveRowThatDidNotTransitionAsCancellationInProgress() { + TransactionSynchronizationManager.setActualTransactionActive(true); + EtlJobService service = service(new UnchangedPendingJobJdbcTemplate(JOB_RECORD_ID)); + + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, "tenant_alpha") + ); + + assertEquals(EtlRequestError.JOB_CANCELLATION_IN_PROGRESS, exception.error()); + } + + @Test + void failsClosedWhenUpdateCountClaimsTransitionButReturnedRowIsStillActive() { + TransactionSynchronizationManager.setActualTransactionActive(true); + EtlJobService service = service(new UnchangedPendingJobJdbcTemplate(JOB_RECORD_ID, 1)); + + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> service.cancelOwned(JOB_RECORD_ID, CANCELLATION_KEY, "tenant_alpha") + ); + + assertEquals(EtlRequestError.JOB_CANCELLATION_IN_PROGRESS, exception.error()); + } + + private static EtlJobService service(JdbcTemplate jdbcTemplate) { + return new EtlJobService( + jdbcTemplate, + new ObjectMapper(), + new EtlBatchProperties(), + idempotencyKeyHash -> true + ); + } + + private static void assertError(EtlRequestError expected, Runnable invocation) { + EtlRequestException exception = assertThrows(EtlRequestException.class, invocation::run); + assertEquals(expected, exception.error()); + } + + /** + * Deterministic JDBC double for concurrency boundaries that return an unchanged PENDING row. + */ + private static final class UnchangedPendingJobJdbcTemplate extends JdbcTemplate { + + private final UUID jobRecordId; + private final int updatedRows; + + private UnchangedPendingJobJdbcTemplate(UUID jobRecordId) { + this(jobRecordId, 0); + } + + private UnchangedPendingJobJdbcTemplate(UUID jobRecordId, int updatedRows) { + this.jobRecordId = jobRecordId; + this.updatedRows = updatedRows; + } + + @Override + public int update(String sql, Object... args) { + return updatedRows; + } + + @Override + public List query(String sql, RowMapper rowMapper, Object... args) { + ResultSet resultSet = mock(ResultSet.class); + try { + when(resultSet.getObject("job_record_id", UUID.class)).thenReturn(jobRecordId); + when(resultSet.getString("job_status")).thenReturn("PENDING"); + when(resultSet.getInt("attempt_count")).thenReturn(0); + when(resultSet.getString("failure_code")).thenReturn(null); + when(resultSet.getString("cancellation_key_hash")).thenReturn(null); + when(resultSet.getTimestamp("created_at")).thenReturn( + Timestamp.from(Instant.parse("2026-08-06T00:00:00Z")) + ); + when(resultSet.getTimestamp("updated_at")).thenReturn( + Timestamp.from(Instant.parse("2026-08-06T00:00:01Z")) + ); + return List.of(rowMapper.mapRow(resultSet, 0)); + } catch (SQLException exception) { + throw new AssertionError("row mapper unexpectedly rejected the test row", exception); + } + } + } +} From da52c2073e948823055c17f654ea82b6f374f1b7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 07:47:35 +0900 Subject: [PATCH 05/18] feat(etl): add cancellation request classifications --- .../xtrmetl/etl/service/EtlRequestError.java | 45 +++++++++++++++++++ 1 file changed, 45 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 45fb2f6a..d28aa33d 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/service/EtlRequestError.java @@ -122,6 +122,51 @@ public enum EtlRequestError { "The ETL job page cursor is invalid or uses an unsupported opaque format." ), + /** The cancellation key is absent or outside the bounded safe idempotency profile. */ + JOB_CANCELLATION_KEY_REQUIRED( + HttpStatus.BAD_REQUEST, + "etl_job_cancellation_key_required", + "urn:mightyetl:problem:etl-job-cancellation-key-required", + "ETL job cancellation key required", + "Cancellation requires a supported principal-scoped Idempotency-Key." + ), + + /** The durable job was already cancelled with a different cancellation key. */ + JOB_CANCELLATION_KEY_REUSED( + HttpStatus.UNPROCESSABLE_ENTITY, + "etl_job_cancellation_key_reused", + "urn:mightyetl:problem:etl-job-cancellation-key-reused", + "ETL job cancellation key reused", + "The durable job was already cancelled with a different Idempotency-Key." + ), + + /** An eligible job remained active after the authoritative cancellation update. */ + JOB_CANCELLATION_IN_PROGRESS( + HttpStatus.CONFLICT, + "etl_job_cancellation_in_progress", + "urn:mightyetl:problem:etl-job-cancellation-in-progress", + "ETL job cancellation in progress", + "The durable job cancellation could not yet establish a terminal outcome." + ), + + /** Durable success committed before the cancellation transition. */ + JOB_ALREADY_SUCCEEDED( + HttpStatus.CONFLICT, + "etl_job_already_succeeded", + "urn:mightyetl:problem:etl-job-already-succeeded", + "ETL job already succeeded", + "The durable job succeeded before cancellation could commit." + ), + + /** Durable failure committed before the cancellation transition. */ + JOB_ALREADY_FAILED( + HttpStatus.CONFLICT, + "etl_job_already_failed", + "urn:mightyetl:problem:etl-job-already-failed", + "ETL job already failed", + "The durable job failed before cancellation could commit." + ), + /** The requested job does not exist in the authenticated principal's namespace. */ JOB_NOT_FOUND( HttpStatus.NOT_FOUND, From de72f6e31a0e11b677fdd551caa4e01d20386159 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 07:48:58 +0900 Subject: [PATCH 06/18] feat(etl): implement owner-scoped durable job cancellation --- .../com/xtrmetl/etl/job/EtlJobService.java | 160 +++++++++++++++++- 1 file changed, 159 insertions(+), 1 deletion(-) diff --git a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java index e1fde8be..3a524df6 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/job/EtlJobService.java @@ -31,7 +31,7 @@ import java.util.regex.Pattern; /** - * Creates, reads, and lists durable principal-scoped asynchronous ETL job resources. + * Creates, reads, lists, and cancels durable principal-scoped asynchronous ETL job resources. * *

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

+ * + *

Cancellation is owner-scoped and idempotent. Only pending or running jobs are eligible for + * the atomic transition; an active lease is cleared by the same database update so a stale worker + * cannot retain ownership after cancellation commits. Raw cancellation keys are never persisted.

*/ @Service public class EtlJobService { @@ -97,6 +101,31 @@ INSERT INTO etl_job_records ( ORDER BY created_at DESC, job_record_id DESC LIMIT ? """; + private static final String SELECT_OWNED_CANCELLATION_SQL = """ + SELECT job_record_id, job_status, attempt_count, failure_code, + cancellation_key_hash, created_at, updated_at + FROM etl_job_records + WHERE job_record_id = ? + AND principal_scope_hash = ? + """; + private static final String CANCEL_OWNED_JOB_SQL = """ + UPDATE etl_job_records + SET job_status = ?, + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + cancellation_key_hash = ?, + cancellation_code = ?, + job_cancelled_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE job_record_id = ? + AND principal_scope_hash = ? + AND job_status IN (?, ?) + """; + static final String CANCELLED_BY_OWNER_CODE = "etl_job_cancelled_by_owner"; + private static final String CANCELLATION_KEY_DOMAIN = "mightyetl:durable-job-cancellation:v1:"; private static final 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; @@ -268,6 +297,80 @@ public EtlJobSnapshot findOwned( return jobs.getFirst(); } + /** + * Cancels one owner-scoped pending or running durable job. + * + *

The state transition, payload clearing, lease invalidation, and cancellation identity + * persistence occur in one conditional database update. The cancellation key is validated with + * the same bounded profile as other idempotency keys, then domain-separated before hashing so + * its digest cannot collide semantically with submission-key storage.

+ * + * @param jobRecordId opaque owner-scoped durable job identifier + * @param cancellationKey principal-scoped cancellation idempotency key + * @param principalScope authenticated principal namespace + * @return committed or replayed cancellation result + * @throws EtlRequestException when cancellation validation, ownership, or state contracts fail + * @throws IllegalStateException when invoked without an actual transaction + */ + @Transactional + public EtlJobCancellation cancelOwned( + UUID jobRecordId, + @Nullable String cancellationKey, + @Nullable String principalScope + ) { + UUID validatedJobId = Objects.requireNonNull(jobRecordId, "jobRecordId must not be null"); + String validatedKey = validateCancellationKey(cancellationKey); + String validatedScope = validatePrincipalScope(principalScope); + requireActiveCancellationTransaction(); + + String principalScopeHash = Sha256Digest.digest(validatedScope); + String cancellationKeyHash = Sha256Digest.digest( + CANCELLATION_KEY_DOMAIN + + principalScopeHash + + ":" + + validatedJobId + + ":" + + validatedKey + ); + int updatedRows = jdbcTemplate.update( + CANCEL_OWNED_JOB_SQL, + EtlJobStatus.CANCELLED.name(), + cancellationKeyHash, + CANCELLED_BY_OWNER_CODE, + validatedJobId, + principalScopeHash, + EtlJobStatus.PENDING.name(), + EtlJobStatus.RUNNING.name() + ); + + StoredCancellation storedCancellation = findOwnedCancellation( + validatedJobId, + principalScopeHash + ); + if (storedCancellation == null) { + throw new EtlRequestException(EtlRequestError.JOB_NOT_FOUND); + } + if (updatedRows == 1) { + if (storedCancellation.snapshot().jobStatus() != EtlJobStatus.CANCELLED) { + throw new EtlRequestException(EtlRequestError.JOB_CANCELLATION_IN_PROGRESS); + } + return new EtlJobCancellation(storedCancellation.snapshot(), false); + } + return switch (storedCancellation.snapshot().jobStatus()) { + case CANCELLED -> { + if (!cancellationKeyHash.equals(storedCancellation.cancellationKeyHash())) { + throw new EtlRequestException(EtlRequestError.JOB_CANCELLATION_KEY_REUSED); + } + yield new EtlJobCancellation(storedCancellation.snapshot(), true); + } + case SUCCEEDED -> throw new EtlRequestException(EtlRequestError.JOB_ALREADY_SUCCEEDED); + case FAILED -> throw new EtlRequestException(EtlRequestError.JOB_ALREADY_FAILED); + case PENDING, RUNNING -> throw new EtlRequestException( + EtlRequestError.JOB_CANCELLATION_IN_PROGRESS + ); + }; + } + /** * Lists one deterministic owner-scoped page of durable ETL jobs. * @@ -344,6 +447,30 @@ private StoredJobRecord findSubmission(String principalScopeHash, String submiss return jobs.isEmpty() ? null : jobs.getFirst(); } + @Nullable + private StoredCancellation findOwnedCancellation( + UUID jobRecordId, + String principalScopeHash + ) { + List jobs = jdbcTemplate.query( + SELECT_OWNED_CANCELLATION_SQL, + (resultSet, rowNumber) -> new StoredCancellation( + mapSnapshot( + resultSet.getObject("job_record_id", UUID.class), + resultSet.getString("job_status"), + resultSet.getInt("attempt_count"), + resultSet.getString("failure_code"), + resultSet.getTimestamp("created_at"), + resultSet.getTimestamp("updated_at") + ), + resultSet.getString("cancellation_key_hash") + ), + jobRecordId, + principalScopeHash + ); + return jobs.isEmpty() ? null : jobs.getFirst(); + } + private static EtlJobSnapshot mapSnapshotRow( java.sql.ResultSet resultSet, int rowNumber @@ -532,6 +659,20 @@ private static String validateIdempotencyKey(@Nullable String idempotencyKey) { throw new EtlRequestException(EtlRequestError.INVALID_IDEMPOTENCY_KEY); } + private static String validateCancellationKey(@Nullable String cancellationKey) { + if (cancellationKey == null) { + throw new EtlRequestException(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED); + } + var structuredFieldMatcher = IDEMPOTENCY_KEY_STRUCTURED_FIELD_PROFILE.matcher(cancellationKey); + if (structuredFieldMatcher.matches()) { + return structuredFieldMatcher.group(1); + } + if (IDEMPOTENCY_KEY_VALUE_PROFILE.matcher(cancellationKey).matches()) { + return cancellationKey; + } + throw new EtlRequestException(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED); + } + private static String validatePrincipalScope(@Nullable String principalScope) { if (principalScope == null || principalScope.isBlank() @@ -550,6 +691,14 @@ private static void requireActiveTransaction() { } } + private static void requireActiveCancellationTransaction() { + if (!TransactionSynchronizationManager.isActualTransactionActive()) { + throw new IllegalStateException( + "Durable ETL job cancellation requires an active transaction" + ); + } + } + private record StoredJobRecord(String requestDigest, EtlJobSnapshot snapshot) { private StoredJobRecord { Objects.requireNonNull(requestDigest, "requestDigest must not be null"); @@ -557,6 +706,15 @@ private record StoredJobRecord(String requestDigest, EtlJobSnapshot snapshot) { } } + private record StoredCancellation( + EtlJobSnapshot snapshot, + @Nullable String cancellationKeyHash + ) { + private StoredCancellation { + Objects.requireNonNull(snapshot, "snapshot must not be null"); + } + } + private record PageCursor(Instant createdAt, UUID jobRecordId) { private PageCursor { Objects.requireNonNull(createdAt, "createdAt must not be null"); From c3055c3e543e09913e14e0badaf810e0b3a8718b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 07:52:17 +0900 Subject: [PATCH 07/18] test(etl): cover cancellation state and privacy branches --- .../etl/job/EtlJobServiceIntegrationTest.java | 204 +++++++++++++++++- 1 file changed, 201 insertions(+), 3 deletions(-) diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java index ad802106..0404f2ba 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobServiceIntegrationTest.java @@ -19,21 +19,27 @@ import org.springframework.transaction.annotation.EnableTransactionManagement; import javax.sql.DataSource; +import java.time.OffsetDateTime; import java.util.UUID; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; /** - * Defines the durable, principal-scoped asynchronous ETL job intake contract. + * Defines the durable, principal-scoped asynchronous ETL job intake and cancellation contract. */ @SpringJUnitConfig(EtlJobServiceIntegrationTest.TestConfiguration.class) class EtlJobServiceIntegrationTest { private static final String IDEMPOTENCY_KEY = "550e8400-e29b-41d4-a716-446655440000"; + private static final String CANCELLATION_KEY = "70dc8b50-e8b2-4e1a-8c5f-d84814708a77"; + private static final String SECOND_CANCELLATION_KEY = + "a52b165f-9d45-4399-ae84-1e93e8fe1e68"; private static final String PAYLOAD = "[{\"id\":\"record_alpha\",\"name\":\"accepted\"}]"; private final EtlJobService etlJobService; @@ -54,10 +60,16 @@ CREATE TABLE etl_job_records ( principal_scope_hash CHAR(64) NOT NULL, submission_key_hash CHAR(64) NOT NULL, request_digest CHAR(64) NOT NULL, - request_payload VARCHAR(8192) NOT NULL, + request_payload VARCHAR(8192), job_status VARCHAR(32) NOT NULL, attempt_count INTEGER NOT NULL DEFAULT 0, failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, CONSTRAINT etl_job_submission_scope_unique @@ -162,6 +174,184 @@ void rejectsMalformedOrOversizedPayloadsBeforePersistence() { assertEquals(0, countJobRows()); } + @Test + void cancelsPendingWorkClearsThePayloadAndReplaysTheSameSemanticKey() { + EtlJobSubmission created = etlJobService.submit(PAYLOAD, IDEMPOTENCY_KEY, "tenant_alpha"); + + EtlJobCancellation cancelled = etlJobService.cancelOwned( + created.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ); + EtlJobCancellation replayed = etlJobService.cancelOwned( + created.jobRecordId(), + "\"" + CANCELLATION_KEY + "\"", + "tenant_alpha" + ); + + assertFalse(cancelled.replayed()); + assertTrue(replayed.replayed()); + assertEquals(created.jobRecordId(), cancelled.snapshot().jobRecordId()); + assertEquals(EtlJobStatus.CANCELLED, cancelled.snapshot().jobStatus()); + assertEquals(cancelled.snapshot(), replayed.snapshot()); + assertNull(column(created.jobRecordId(), "request_payload", String.class)); + assertEquals( + EtlJobService.CANCELLED_BY_OWNER_CODE, + column(created.jobRecordId(), "cancellation_code", String.class) + ); + String keyHash = column(created.jobRecordId(), "cancellation_key_hash", String.class); + assertNotNull(keyHash); + assertEquals(64, keyHash.length()); + assertNotEquals(CANCELLATION_KEY, keyHash); + assertNotNull(column( + created.jobRecordId(), + "job_cancelled_at", + OffsetDateTime.class + )); + } + + @Test + void rejectsASecondCancellationIdentityAfterCancellation() { + EtlJobSubmission created = etlJobService.submit(PAYLOAD, IDEMPOTENCY_KEY, "tenant_alpha"); + etlJobService.cancelOwned(created.jobRecordId(), CANCELLATION_KEY, "tenant_alpha"); + + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + created.jobRecordId(), + SECOND_CANCELLATION_KEY, + "tenant_alpha" + ) + ); + + assertEquals(EtlRequestError.JOB_CANCELLATION_KEY_REUSED, exception.error()); + assertEquals(EtlJobStatus.CANCELLED, etlJobService.findOwned( + created.jobRecordId(), + "tenant_alpha" + ).jobStatus()); + } + + @Test + void keepsForeignOwnedAndMissingCancellationTargetsIndistinguishable() { + EtlJobSubmission created = etlJobService.submit(PAYLOAD, IDEMPOTENCY_KEY, "tenant_alpha"); + + EtlRequestException hidden = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + created.jobRecordId(), + CANCELLATION_KEY, + "tenant_beta" + ) + ); + EtlRequestException missing = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + UUID.randomUUID(), + CANCELLATION_KEY, + "tenant_alpha" + ) + ); + + assertEquals(EtlRequestError.JOB_NOT_FOUND, hidden.error()); + assertEquals(EtlRequestError.JOB_NOT_FOUND, missing.error()); + assertEquals(EtlJobStatus.PENDING, etlJobService.findOwned( + created.jobRecordId(), + "tenant_alpha" + ).jobStatus()); + } + + @Test + void rejectsCancellationAfterACommittedSuccessOrFailure() { + EtlJobSubmission succeeded = etlJobService.submit( + PAYLOAD, + IDEMPOTENCY_KEY, + "tenant_alpha" + ); + EtlJobSubmission failed = etlJobService.submit( + PAYLOAD, + "1d38ad67-48d8-446c-bca1-76bfe2ba8eef", + "tenant_alpha" + ); + jdbcTemplate.update( + "UPDATE etl_job_records SET job_status = 'SUCCEEDED', request_payload = NULL " + + "WHERE job_record_id = ?", + succeeded.jobRecordId() + ); + jdbcTemplate.update( + "UPDATE etl_job_records SET job_status = 'FAILED', request_payload = NULL, " + + "failure_code = 'etl_target_failure' WHERE job_record_id = ?", + failed.jobRecordId() + ); + + EtlRequestException successConflict = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + succeeded.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ) + ); + EtlRequestException failureConflict = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned( + failed.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ) + ); + + assertEquals("etl_job_already_succeeded", successConflict.error().errorCode()); + assertEquals("etl_job_already_failed", failureConflict.error().errorCode()); + } + + @Test + void cancelsRunningWorkAndInvalidatesEveryLeaseField() { + EtlJobSubmission created = etlJobService.submit(PAYLOAD, IDEMPOTENCY_KEY, "tenant_alpha"); + jdbcTemplate.update( + """ + UPDATE etl_job_records + SET job_status = 'RUNNING', + attempt_count = 1, + lease_claim_id = ?, + lease_owner_id = 'worker_alpha', + lease_expires_at = DATEADD('MINUTE', 5, CURRENT_TIMESTAMP) + WHERE job_record_id = ? + """, + UUID.fromString("7c10a65b-5791-4e0f-9fba-dadbb13971da"), + created.jobRecordId() + ); + + EtlJobCancellation cancellation = etlJobService.cancelOwned( + created.jobRecordId(), + CANCELLATION_KEY, + "tenant_alpha" + ); + + assertEquals(EtlJobStatus.CANCELLED, cancellation.snapshot().jobStatus()); + assertEquals(1, cancellation.snapshot().attemptCount()); + assertNull(column(created.jobRecordId(), "lease_claim_id", UUID.class)); + assertNull(column(created.jobRecordId(), "lease_owner_id", String.class)); + assertNull(column(created.jobRecordId(), "lease_expires_at", OffsetDateTime.class)); + assertNull(column(created.jobRecordId(), "request_payload", String.class)); + } + + @Test + void rejectsInvalidCancellationKeysBeforeAnyTableAccess() { + jdbcTemplate.execute("DROP TABLE etl_job_records"); + + EtlRequestException absent = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned(UUID.randomUUID(), null, "tenant_alpha") + ); + EtlRequestException malformed = assertThrows( + EtlRequestException.class, + () -> etlJobService.cancelOwned(UUID.randomUUID(), "too-short", "tenant_alpha") + ); + + assertEquals(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED, absent.error()); + assertEquals(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED, malformed.error()); + } + private int countJobRows() { Integer count = jdbcTemplate.queryForObject( "SELECT COUNT(*) FROM etl_job_records", @@ -170,8 +360,16 @@ private int countJobRows() { return count == null ? 0 : count; } + private T column(UUID jobRecordId, String columnName, Class valueType) { + return jdbcTemplate.queryForObject( + "SELECT " + columnName + " FROM etl_job_records WHERE job_record_id = ?", + (resultSet, rowNumber) -> resultSet.getObject(columnName, valueType), + jobRecordId + ); + } + /** - * Minimal transaction-enabled test context for the job intake service. + * Minimal transaction-enabled test context for durable job resource services. */ @Configuration @EnableTransactionManagement From 7c97ea9d40f67bd8478f3a35103a9d773c601aab Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:08:14 +0900 Subject: [PATCH 08/18] test(etl): require cancellation migration lifecycle --- .../job/EtlJobMigrationDocumentationTest.java | 30 +++++++++++++++++++ 1 file changed, 30 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 b7c75107..42827513 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java @@ -50,6 +50,36 @@ void migrationReservesStableWorkerStatesAndRequiresTerminalPayloadClearing() thr assertTrue(migration.contains("request_payload IS NULL")); } + @Test + void cancellationMigrationAddsOneTerminalOwnerSafeLifecycle() throws IOException { + String migration = read( + "etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql" + ).replaceAll("\\s+", " "); + + assertTrue(migration.contains("ADD COLUMN cancellation_key_hash CHAR(64)")); + assertTrue(migration.contains("ADD COLUMN cancellation_code VARCHAR(128)")); + assertTrue(migration.contains("ADD COLUMN job_cancelled_at TIMESTAMPTZ")); + assertTrue(migration.contains( + "job_status IN ('PENDING', 'RUNNING', 'SUCCEEDED', 'FAILED', 'CANCELLED')" + )); + assertTrue(migration.contains( + "job_status IN ('SUCCEEDED', 'FAILED', 'CANCELLED') AND request_payload IS NULL" + )); + assertTrue(migration.contains("job_status = 'CANCELLED'")); + assertTrue(migration.contains("cancellation_key_hash IS NOT NULL")); + assertTrue(migration.contains("cancellation_code IS NOT NULL")); + assertTrue(migration.contains("job_cancelled_at IS NOT NULL")); + assertTrue(migration.contains("job_status <> 'CANCELLED'")); + assertTrue(migration.contains("cancellation_key_hash IS NULL")); + assertTrue(migration.contains("lease_claim_id IS NULL")); + assertTrue(migration.contains("CONSTRAINT etl_job_cancellation_key_hash_format")); + assertTrue(migration.contains("CONSTRAINT etl_job_cancellation_code_format")); + assertTrue(migration.contains("CONSTRAINT etl_job_cancellation_lifecycle_check")); + assertFalse(migration.contains("principal_name")); + assertFalse(migration.contains("cancellation_key TEXT")); + assertFalse(migration.contains("cancellation_reason")); + } + @Test void paginationMigrationUsesTheOwnerAndCompleteStableOrderingKey() throws IOException { String migration = read( From 0eba8f8515d5ac9f98b35243d85b278ccf066c4b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:09:31 +0900 Subject: [PATCH 09/18] feat(etl): add cancellation lifecycle migration --- .../V6__add_etl_job_cancellation.sql | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) create mode 100644 etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql diff --git a/etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql b/etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql new file mode 100644 index 00000000..b7a1b7af --- /dev/null +++ b/etl-service/src/main/resources/db/migration/V6__add_etl_job_cancellation.sql @@ -0,0 +1,81 @@ +-- Add owner-scoped durable job cancellation without retaining raw principals or keys. +ALTER TABLE etl_job_records + ADD COLUMN cancellation_key_hash CHAR(64), + ADD COLUMN cancellation_code VARCHAR(128), + ADD COLUMN job_cancelled_at TIMESTAMPTZ; + +-- Replace lifecycle constraints so clean installations and upgrades converge on one state machine. +ALTER TABLE etl_job_records + DROP CONSTRAINT etl_job_status_value_check, + DROP CONSTRAINT etl_job_payload_lifecycle_check, + DROP CONSTRAINT etl_job_lease_lifecycle_check, + DROP CONSTRAINT etl_job_failure_lifecycle_check; + +ALTER TABLE etl_job_records + ADD CONSTRAINT etl_job_status_value_check CHECK ( + job_status IN ('PENDING', 'RUNNING', 'SUCCEEDED', 'FAILED', 'CANCELLED') + ), + ADD CONSTRAINT etl_job_payload_lifecycle_check CHECK ( + ( + job_status IN ('PENDING', 'RUNNING') + AND request_payload IS NOT NULL + ) + OR + ( + job_status IN ('SUCCEEDED', 'FAILED', 'CANCELLED') + AND request_payload IS NULL + ) + ), + ADD CONSTRAINT etl_job_lease_lifecycle_check CHECK ( + ( + job_status = 'RUNNING' + AND lease_claim_id IS NOT NULL + AND lease_owner_id IS NOT NULL + AND lease_expires_at IS NOT NULL + ) + OR + ( + job_status <> 'RUNNING' + AND lease_claim_id IS NULL + AND lease_owner_id IS NULL + AND lease_expires_at IS NULL + ) + ), + ADD CONSTRAINT etl_job_failure_lifecycle_check CHECK ( + ( + job_status = 'FAILED' + AND failure_code IS NOT NULL + ) + OR + ( + job_status <> 'FAILED' + AND failure_code IS NULL + ) + ), + ADD CONSTRAINT etl_job_cancellation_key_hash_format CHECK ( + cancellation_key_hash IS NULL + OR cancellation_key_hash ~ '^[0-9a-f]{64}$' + ), + ADD CONSTRAINT etl_job_cancellation_code_format CHECK ( + cancellation_code IS NULL + OR cancellation_code ~ '^[a-z][a-z0-9_]{2,127}$' + ), + ADD CONSTRAINT etl_job_cancellation_code_value CHECK ( + cancellation_code IS NULL + OR cancellation_code = 'etl_job_cancelled_by_owner' + ), + ADD CONSTRAINT etl_job_cancellation_lifecycle_check CHECK ( + ( + job_status = 'CANCELLED' + AND cancellation_key_hash IS NOT NULL + AND cancellation_code IS NOT NULL + AND job_cancelled_at IS NOT NULL + ) + OR + ( + job_status <> 'CANCELLED' + AND cancellation_key_hash IS NULL + AND cancellation_code IS NULL + AND job_cancelled_at IS NULL + ) + ); From 0c51835069c9ac2fdc4a7e92b3029ed76ef9bc80 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:11:21 +0900 Subject: [PATCH 10/18] test(etl): expose cancellation HTTP contract --- .../xtrmetl/etl/job/EtlJobControllerTest.java | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java index 940d0626..469e105b 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerTest.java @@ -4,6 +4,7 @@ import com.xtrmetl.etl.controller.EtlJobController; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.test.web.servlet.MockMvc; import org.springframework.test.web.servlet.setup.MockMvcBuilders; @@ -30,6 +31,8 @@ class EtlJobControllerTest { private static final String JOBS_PATH = "/api/etl/jobs"; private static final String IDEMPOTENCY_KEY = "\"550e8400-e29b-41d4-a716-446655440000\""; + private static final String CANCELLATION_KEY = + "\"70dc8b50-e8b2-4e1a-8c5f-d84814708a77\""; private static final Principal PRINCIPAL = () -> "tenant_alpha"; private EtlJobService etlJobService; @@ -154,4 +157,82 @@ void requiresAuthenticationForJobStatus() throws Exception { verifyNoInteractions(etlJobService); } + + @Test + void cancelsAnOwnedJobAndReturnsTheTerminalStatus() throws Exception { + UUID jobRecordId = UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef"); + EtlJobSnapshot snapshot = new EtlJobSnapshot( + jobRecordId, + EtlJobStatus.CANCELLED, + 1, + null, + Instant.parse("2026-08-04T10:00:00Z"), + Instant.parse("2026-08-06T03:00:00Z") + ); + when(etlJobService.cancelOwned(jobRecordId, CANCELLATION_KEY, "tenant_alpha")) + .thenReturn(new EtlJobCancellation(snapshot, false)); + + mockMvc.perform(post(cancellationPath(jobRecordId)) + .principal(PRINCIPAL) + .header("Idempotency-Key", CANCELLATION_KEY)) + .andExpect(status().isOk()) + .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON)) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(header().string("Idempotency-Replayed", "false")) + .andExpect(header().exists(HttpHeaders.ETAG)) + .andExpect(jsonPath("$.jobRecordId").value(jobRecordId.toString())) + .andExpect(jsonPath("$.jobStatus").value("CANCELLED")) + .andExpect(jsonPath("$.attemptCount").value(1)) + .andExpect(jsonPath("$.failureCode").doesNotExist()) + .andExpect(jsonPath("$.requestPayload").doesNotExist()) + .andExpect(jsonPath("$.cancellationKeyHash").doesNotExist()) + .andExpect(jsonPath("$.cancellationCode").doesNotExist()); + + verify(etlJobService).cancelOwned(jobRecordId, CANCELLATION_KEY, "tenant_alpha"); + } + + @Test + void marksAnIdenticalCancellationAsReplayed() throws Exception { + UUID jobRecordId = UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef"); + EtlJobSnapshot snapshot = new EtlJobSnapshot( + jobRecordId, + EtlJobStatus.CANCELLED, + 0, + null, + Instant.parse("2026-08-04T10:00:00Z"), + Instant.parse("2026-08-06T03:00:00Z") + ); + when(etlJobService.cancelOwned(jobRecordId, CANCELLATION_KEY, "tenant_alpha")) + .thenReturn(new EtlJobCancellation(snapshot, true)); + + mockMvc.perform(post(cancellationPath(jobRecordId)) + .principal(PRINCIPAL) + .header("Idempotency-Key", CANCELLATION_KEY)) + .andExpect(status().isOk()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(header().string("Idempotency-Replayed", "true")) + .andExpect(jsonPath("$.jobStatus").value("CANCELLED")); + } + + @Test + void requiresAuthenticationAndAKeyBeforeCancellationServiceAccess() throws Exception { + UUID jobRecordId = UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef"); + + mockMvc.perform(post(cancellationPath(jobRecordId)) + .header("Idempotency-Key", CANCELLATION_KEY)) + .andExpect(status().isUnauthorized()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_idempotency_principal_required")); + + mockMvc.perform(post(cancellationPath(jobRecordId)).principal(PRINCIPAL)) + .andExpect(status().isBadRequest()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_job_cancellation_key_required")); + + verifyNoInteractions(etlJobService); + } + + private static String cancellationPath(UUID jobRecordId) { + return JOBS_PATH + "/" + jobRecordId + "/cancellation"; + } } From 8e05eb1c69221be3806832eaad6ac1e2c1bcb79c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:14:26 +0900 Subject: [PATCH 11/18] feat(etl): expose owner-scoped cancellation endpoint --- .../etl/controller/EtlJobController.java | 70 +++++++++++++++++-- 1 file changed, 65 insertions(+), 5 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 bb2c1e0e..8f5b6b08 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobController.java @@ -1,6 +1,7 @@ package com.xtrmetl.etl.controller; import com.xtrmetl.etl.job.EtlJobAcceptedResponse; +import com.xtrmetl.etl.job.EtlJobCancellation; import com.xtrmetl.etl.job.EtlJobPage; import com.xtrmetl.etl.job.EtlJobPageResponse; import com.xtrmetl.etl.job.EtlJobService; @@ -35,7 +36,7 @@ import java.util.UUID; /** - * Exposes durable asynchronous ETL job submission, discovery, and status resources. + * Exposes durable asynchronous ETL job submission, discovery, status, and cancellation resources. * *

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

* + *

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

+ * *

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

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

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

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

+ * + * @param jobRecordIdText opaque durable job identifier text + * @param idempotencyKey required client-generated cancellation key + * @param principal authenticated principal namespace + * @return cancelled operator-safe status and replay evidence + */ + @PostMapping("/{jobRecordId}/cancellation") + @Observed(name = "etl.jobs.cancel", contextualName = "etl-job-cancellation") + public ResponseEntity cancel( + @PathVariable("jobRecordId") String jobRecordIdText, + @RequestHeader(value = "Idempotency-Key", required = false) + @Nullable String idempotencyKey, + @Nullable Principal principal + ) { + if (principal == null) { + throw new EtlRequestException(EtlRequestError.IDEMPOTENCY_PRINCIPAL_REQUIRED); + } + if (idempotencyKey == null) { + throw new EtlRequestException(EtlRequestError.JOB_CANCELLATION_KEY_REQUIRED); + } + + UUID jobRecordId = parseJobRecordId(jobRecordIdText); + final EtlJobCancellation cancellation; + try { + cancellation = etlJobService.cancelOwned( + jobRecordId, + idempotencyKey, + principal.getName() + ); + } catch (EtlRequestException | DataAccessException exception) { + throw exception; + } catch (RuntimeException exception) { + throw new EtlUnexpectedException(exception); + } + + EtlJobStatusResponse responseBody = EtlJobStatusResponse.from(cancellation.snapshot()); + return ResponseEntity.ok() + .cacheControl(CacheControl.noStore()) + .eTag(statusEntityTag(responseBody)) + .header( + IDEMPOTENCY_REPLAYED_HEADER, + Boolean.toString(cancellation.replayed()) + ) + .contentType(MediaType.APPLICATION_JSON) + .body(responseBody); + } + /** * Builds an opaque weak validator from the complete status representation. * From 5fba2d3fae7e3d2f5bf627ee5e8040072c7ef939 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:16:25 +0900 Subject: [PATCH 12/18] test(etl): close cancellation controller coverage --- .../etl/job/EtlJobControllerFailureTest.java | 67 ++++++++++++++++++- 1 file changed, 65 insertions(+), 2 deletions(-) diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java index 0418ebaa..30a2f420 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobControllerFailureTest.java @@ -26,13 +26,15 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; /** - * Covers typed data-access, malformed resource identifiers, and unexpected failures at both - * durable job controller boundaries. + * Covers typed data-access, malformed resource identifiers, and unexpected failures at durable job + * submission, status, and cancellation boundaries. */ class EtlJobControllerFailureTest { private static final String JOBS_PATH = "/api/etl/jobs"; private static final String IDEMPOTENCY_KEY = "\"550e8400-e29b-41d4-a716-446655440000\""; + private static final String CANCELLATION_KEY = + "\"70dc8b50-e8b2-4e1a-8c5f-d84814708a77\""; private static final Principal PRINCIPAL = () -> "tenant_alpha"; private EtlJobService etlJobService; @@ -127,6 +129,61 @@ void mapsStatusUnexpectedFailuresWithoutLeakingMessages() throws Exception { .andExpect(jsonPath("$.errorCode").value("etl_internal_error")); } + @Test + void preservesTypedCancellationConflicts() throws Exception { + when(etlJobService.cancelOwned(any(UUID.class), anyString(), anyString())) + .thenThrow(new EtlRequestException(EtlRequestError.JOB_ALREADY_SUCCEEDED)); + + performCancellation() + .andExpect(status().isConflict()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_job_already_succeeded")) + .andExpect(jsonPath("$.detail").value( + "The durable job succeeded before cancellation could commit." + )); + } + + @Test + void treatsMalformedCancellationIdentifiersAsOwnerSafeNotFound() throws Exception { + mockMvc.perform(post(JOBS_PATH + "/not-a-uuid/cancellation") + .principal(PRINCIPAL) + .header("Idempotency-Key", CANCELLATION_KEY)) + .andExpect(status().isNotFound()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_job_not_found")); + + verifyNoInteractions(etlJobService); + } + + @Test + void mapsCancellationDatabaseFailuresWithoutLeakingMessages() throws Exception { + DataAccessException databaseFailure = new DataAccessException("secret database detail") { }; + when(etlJobService.cancelOwned(any(UUID.class), anyString(), anyString())) + .thenThrow(databaseFailure); + + performCancellation() + .andExpect(status().isInternalServerError()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_target_failure")) + .andExpect(jsonPath("$.detail").value( + "The ETL target could not process the request." + )); + } + + @Test + void mapsCancellationUnexpectedFailuresWithoutLeakingMessages() throws Exception { + when(etlJobService.cancelOwned(any(UUID.class), anyString(), anyString())) + .thenThrow(new IllegalStateException("secret cancellation detail")); + + performCancellation() + .andExpect(status().isInternalServerError()) + .andExpect(header().string("Cache-Control", "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_internal_error")) + .andExpect(jsonPath("$.detail").value( + "The ETL request could not be processed." + )); + } + private org.springframework.test.web.servlet.ResultActions performSubmission() throws Exception { return mockMvc.perform(post(JOBS_PATH) .principal(PRINCIPAL) @@ -138,4 +195,10 @@ private org.springframework.test.web.servlet.ResultActions performSubmission() t private org.springframework.test.web.servlet.ResultActions performStatus() throws Exception { return mockMvc.perform(get(JOBS_PATH + "/" + UUID.randomUUID()).principal(PRINCIPAL)); } + + private org.springframework.test.web.servlet.ResultActions performCancellation() throws Exception { + return mockMvc.perform(post(JOBS_PATH + "/" + UUID.randomUUID() + "/cancellation") + .principal(PRINCIPAL) + .header("Idempotency-Key", CANCELLATION_KEY)); + } } From f842acb18f5a36cbd5e08e118c95d93dbe631cdc Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:17:57 +0900 Subject: [PATCH 13/18] test(etl): require cancelled polling terminality --- .../java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java index e21dd479..e51a4bc1 100644 --- a/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java +++ b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java @@ -71,6 +71,7 @@ void terminalJobsRemoveAnyRetryAfterSuggestion() { assertTerminalHeaderRemoved(advice, EtlJobStatus.SUCCEEDED); assertTerminalHeaderRemoved(advice, EtlJobStatus.FAILED); + assertTerminalHeaderRemoved(advice, EtlJobStatus.CANCELLED); } @Test From 123bb5bfaf0ac1e5881ff65ec7463888268906df Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:20:36 +0900 Subject: [PATCH 14/18] fix(etl): treat cancelled jobs as polling terminal --- .../xtrmetl/etl/controller/EtlJobPollingAdvice.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java index 3973f581..998139e5 100644 --- a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java +++ b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java @@ -20,9 +20,9 @@ * *

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

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

* *

The header is only client guidance. PostgreSQL lease fencing, principal-scoped authorization, * lifecycle validation, rate limiting, and client-side backoff remain independent correctness and @@ -124,7 +124,9 @@ public Object beforeBodyWrite( response.getHeaders().remove(HttpHeaders.RETRY_AFTER); } } - case SUCCEEDED, FAILED -> response.getHeaders().remove(HttpHeaders.RETRY_AFTER); + case SUCCEEDED, FAILED, CANCELLED -> response.getHeaders().remove( + HttpHeaders.RETRY_AFTER + ); } } return body; From 15db978eee8649116ac7f9203482813175ddd6d9 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:22:14 +0900 Subject: [PATCH 15/18] test(etl): prove concurrent cancellation convergence --- ...ancellationConcurrencyIntegrationTest.java | 257 ++++++++++++++++++ 1 file changed, 257 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConcurrencyIntegrationTest.java diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConcurrencyIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConcurrencyIntegrationTest.java new file mode 100644 index 00000000..86090241 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationConcurrencyIntegrationTest.java @@ -0,0 +1,257 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import com.xtrmetl.etl.service.EtlRequestLock; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; + +import javax.sql.DataSource; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Proves concurrent owner cancellation requests converge on one authoritative terminal transition. + */ +@SpringJUnitConfig(EtlJobCancellationConcurrencyIntegrationTest.TestConfiguration.class) +class EtlJobCancellationConcurrencyIntegrationTest { + + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + private static final String SUBMISSION_KEY = "550e8400-e29b-41d4-a716-446655440000"; + private static final String CANCELLATION_KEY = "70dc8b50-e8b2-4e1a-8c5f-d84814708a77"; + private static final String OTHER_CANCELLATION_KEY = + "a52b165f-9d45-4399-ae84-1e93e8fe1e68"; + + private final EtlJobService jobService; + private final JdbcTemplate jdbcTemplate; + private final ExecutorService executor = Executors.newFixedThreadPool(2); + + @Autowired + EtlJobCancellationConcurrencyIntegrationTest( + EtlJobService jobService, + JdbcTemplate jdbcTemplate + ) { + this.jobService = jobService; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createJobTable() { + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_job_records"); + jdbcTemplate.execute(""" + CREATE TABLE etl_job_records ( + job_record_id UUID PRIMARY KEY, + principal_scope_hash CHAR(64) NOT NULL, + submission_key_hash CHAR(64) NOT NULL, + request_digest CHAR(64) NOT NULL, + request_payload CLOB, + job_status VARCHAR(32) NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT etl_job_submission_scope_unique + UNIQUE (principal_scope_hash, submission_key_hash) + ) + """); + } + + @AfterEach + void stopExecutor() throws InterruptedException { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + + @Test + void concurrentIdenticalRequestsProduceOneTransitionAndOneReplay() throws Exception { + UUID jobRecordId = submitPendingJob(); + List> futures = startTogether( + jobRecordId, + List.of(CANCELLATION_KEY, CANCELLATION_KEY) + ); + + List results = List.of( + futures.get(0).get(10, TimeUnit.SECONDS), + futures.get(1).get(10, TimeUnit.SECONDS) + ); + + assertEquals(1, results.stream().filter(result -> !result.replayed()).count()); + assertEquals(1, results.stream().filter(EtlJobCancellation::replayed).count()); + assertTrue(results.stream().allMatch( + result -> result.snapshot().jobStatus() == EtlJobStatus.CANCELLED + )); + assertEquals(1, cancelledRowCount(jobRecordId)); + } + + @Test + void concurrentDifferentKeysProduceOneTransitionAndOneStableConflict() throws Exception { + UUID jobRecordId = submitPendingJob(); + List> futures = startTogether( + jobRecordId, + List.of(CANCELLATION_KEY, OTHER_CANCELLATION_KEY) + ); + + int successes = 0; + int keyConflicts = 0; + for (Future future : futures) { + try { + EtlJobCancellation result = future.get(10, TimeUnit.SECONDS); + assertFalse(result.replayed()); + successes++; + } catch (ExecutionException exception) { + Throwable cause = exception.getCause(); + EtlRequestException requestException = assertThrows( + EtlRequestException.class, + () -> { + throw cause; + } + ); + assertEquals( + EtlRequestError.JOB_CANCELLATION_KEY_REUSED, + requestException.error() + ); + keyConflicts++; + } + } + + assertEquals(1, successes); + assertEquals(1, keyConflicts); + assertEquals(1, cancelledRowCount(jobRecordId)); + } + + private UUID submitPendingJob() { + return jobService.submit(PAYLOAD, SUBMISSION_KEY, "tenant_alpha").jobRecordId(); + } + + private List> startTogether( + UUID jobRecordId, + List cancellationKeys + ) throws InterruptedException { + CountDownLatch ready = new CountDownLatch(cancellationKeys.size()); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (String cancellationKey : cancellationKeys) { + futures.add(executor.submit(() -> { + ready.countDown(); + if (!start.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("concurrent cancellation start timed out"); + } + return jobService.cancelOwned( + jobRecordId, + cancellationKey, + "tenant_alpha" + ); + })); + } + assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + return List.copyOf(futures); + } + + private int cancelledRowCount(UUID jobRecordId) { + Integer count = jdbcTemplate.queryForObject( + """ + SELECT COUNT(*) + FROM etl_job_records + WHERE job_record_id = ? + AND job_status = 'CANCELLED' + AND request_payload IS NULL + AND lease_claim_id IS NULL + AND lease_owner_id IS NULL + AND lease_expires_at IS NULL + AND cancellation_key_hash IS NOT NULL + AND cancellation_code = ? + AND job_cancelled_at IS NOT NULL + """, + Integer.class, + jobRecordId, + EtlJobService.CANCELLED_BY_OWNER_CODE + ); + return count == null ? 0 : count; + } + + /** Minimal transaction-enabled context for concurrent cancellation integration. */ + @Configuration + @EnableTransactionManagement + static class TestConfiguration { + + @Bean + DataSource dataSource() { + return new EmbeddedDatabaseBuilder() + .generateUniqueName(true) + .setType(EmbeddedDatabaseType.H2) + .build(); + } + + @Bean + JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + PlatformTransactionManager transactionManager(DataSource dataSource) { + return new DataSourceTransactionManager(dataSource); + } + + @Bean + ObjectMapper objectMapper() { + return new ObjectMapper(); + } + + @Bean + EtlBatchProperties etlBatchProperties() { + return new EtlBatchProperties(); + } + + @Bean + EtlRequestLock etlRequestLock() { + return lockHash -> true; + } + + @Bean + EtlJobService etlJobService( + JdbcTemplate jdbcTemplate, + ObjectMapper objectMapper, + EtlBatchProperties batchProperties, + EtlRequestLock requestLock + ) { + return new EtlJobService( + jdbcTemplate, + objectMapper, + batchProperties, + requestLock + ); + } + } +} From 1f86c4e438bc2b653a829183b2989e562fa698ed Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:26:22 +0900 Subject: [PATCH 16/18] test(etl): prove cancellation key domain separation --- ...bCancellationKeyDomainIntegrationTest.java | 209 ++++++++++++++++++ 1 file changed, 209 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationKeyDomainIntegrationTest.java diff --git a/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationKeyDomainIntegrationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationKeyDomainIntegrationTest.java new file mode 100644 index 00000000..7a457de1 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobCancellationKeyDomainIntegrationTest.java @@ -0,0 +1,209 @@ +package com.xtrmetl.etl.job; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestLock; +import com.xtrmetl.etl.service.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.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Proves durable cancellation idempotency hashes are purpose-, owner-, and job-bound without + * persisting the raw cancellation key. + */ +@SpringJUnitConfig(EtlJobCancellationKeyDomainIntegrationTest.TestConfiguration.class) +class EtlJobCancellationKeyDomainIntegrationTest { + + private static final String CANCELLATION_DOMAIN = + "mightyetl:durable-job-cancellation:v1:"; + private static final String PAYLOAD = "[{\"id\":\"record_alpha\"}]"; + private static final String CANCELLATION_KEY = + "70dc8b50-e8b2-4e1a-8c5f-d84814708a77"; + private static final String SUBMISSION_KEY_ALPHA = + "550e8400-e29b-41d4-a716-446655440000"; + private static final String SUBMISSION_KEY_BETA = + "1d38ad67-48d8-446c-bca1-76bfe2ba8eef"; + private static final String SUBMISSION_KEY_OTHER_OWNER = + "11cf0982-4fe9-4a35-b981-58390596163f"; + + private final EtlJobService jobService; + private final JdbcTemplate jdbcTemplate; + + @Autowired + EtlJobCancellationKeyDomainIntegrationTest( + EtlJobService jobService, + JdbcTemplate jdbcTemplate + ) { + this.jobService = jobService; + this.jdbcTemplate = jdbcTemplate; + } + + @BeforeEach + void createJobTable() { + jdbcTemplate.execute("DROP TABLE IF EXISTS etl_job_records"); + jdbcTemplate.execute(""" + CREATE TABLE etl_job_records ( + job_record_id UUID PRIMARY KEY, + principal_scope_hash CHAR(64) NOT NULL, + submission_key_hash CHAR(64) NOT NULL, + request_digest CHAR(64) NOT NULL, + request_payload CLOB, + job_status VARCHAR(32) NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + failure_code VARCHAR(128), + lease_claim_id UUID, + lease_owner_id VARCHAR(128), + lease_expires_at TIMESTAMP WITH TIME ZONE, + cancellation_key_hash CHAR(64), + cancellation_code VARCHAR(128), + job_cancelled_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT etl_job_submission_scope_unique + UNIQUE (principal_scope_hash, submission_key_hash) + ) + """); + } + + @Test + void bindsTheSameRawCancellationKeyToItsPurposeOwnerAndJob() { + UUID alphaJob = submit(SUBMISSION_KEY_ALPHA, "tenant_alpha"); + UUID betaJob = submit(SUBMISSION_KEY_BETA, "tenant_alpha"); + UUID otherOwnerJob = submit(SUBMISSION_KEY_OTHER_OWNER, "tenant_beta"); + + jobService.cancelOwned(alphaJob, CANCELLATION_KEY, "tenant_alpha"); + jobService.cancelOwned(betaJob, CANCELLATION_KEY, "tenant_alpha"); + jobService.cancelOwned(otherOwnerJob, CANCELLATION_KEY, "tenant_beta"); + + String alphaHash = cancellationHash(alphaJob); + String betaHash = cancellationHash(betaJob); + String otherOwnerHash = cancellationHash(otherOwnerJob); + + assertEquals(expectedHash("tenant_alpha", alphaJob), alphaHash); + assertEquals(expectedHash("tenant_alpha", betaJob), betaHash); + assertEquals(expectedHash("tenant_beta", otherOwnerJob), otherOwnerHash); + assertNotEquals(alphaHash, betaHash); + assertNotEquals(alphaHash, otherOwnerHash); + assertNotEquals(betaHash, otherOwnerHash); + assertNotEquals(CANCELLATION_KEY, alphaHash); + assertEquals(64, alphaHash.length()); + } + + @Test + void normalizesQuotedAndLegacyRawKeysToOneReplayIdentity() { + UUID jobRecordId = submit(SUBMISSION_KEY_ALPHA, "tenant_alpha"); + + EtlJobCancellation first = jobService.cancelOwned( + jobRecordId, + CANCELLATION_KEY, + "tenant_alpha" + ); + String firstHash = cancellationHash(jobRecordId); + EtlJobCancellation replay = jobService.cancelOwned( + jobRecordId, + "\"" + CANCELLATION_KEY + "\"", + "tenant_alpha" + ); + + assertFalse(first.replayed()); + assertTrue(replay.replayed()); + assertEquals(first.snapshot(), replay.snapshot()); + assertEquals(firstHash, cancellationHash(jobRecordId)); + } + + private UUID submit(String submissionKey, String principalScope) { + return jobService.submit(PAYLOAD, submissionKey, principalScope).jobRecordId(); + } + + private String cancellationHash(UUID jobRecordId) { + return jdbcTemplate.queryForObject( + "SELECT cancellation_key_hash FROM etl_job_records WHERE job_record_id = ?", + String.class, + jobRecordId + ); + } + + private static String expectedHash(String principalScope, UUID jobRecordId) { + String principalScopeHash = Sha256Digest.digest(principalScope); + return Sha256Digest.digest( + CANCELLATION_DOMAIN + + principalScopeHash + + ":" + + jobRecordId + + ":" + + CANCELLATION_KEY + ); + } + + /** Minimal transaction-enabled context for cancellation-domain integration tests. */ + @Configuration + @EnableTransactionManagement + static class TestConfiguration { + + @Bean + DataSource dataSource() { + return new EmbeddedDatabaseBuilder() + .generateUniqueName(true) + .setType(EmbeddedDatabaseType.H2) + .build(); + } + + @Bean + JdbcTemplate jdbcTemplate(DataSource dataSource) { + return new JdbcTemplate(dataSource); + } + + @Bean + PlatformTransactionManager transactionManager(DataSource dataSource) { + return new DataSourceTransactionManager(dataSource); + } + + @Bean + ObjectMapper objectMapper() { + return new ObjectMapper(); + } + + @Bean + EtlBatchProperties etlBatchProperties() { + return new EtlBatchProperties(); + } + + @Bean + EtlRequestLock etlRequestLock() { + return lockHash -> true; + } + + @Bean + EtlJobService etlJobService( + JdbcTemplate jdbcTemplate, + ObjectMapper objectMapper, + EtlBatchProperties batchProperties, + EtlRequestLock requestLock + ) { + return new EtlJobService( + jdbcTemplate, + objectMapper, + batchProperties, + requestLock + ); + } + } +} From 7c53c8fd8e8b317a263652149742a56395dd01a0 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:35:50 +0900 Subject: [PATCH 17/18] test(etl): restore cancellation documentation contract --- ...rableJobCancellationDocumentationTest.java | 121 ++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 etl-service/src/test/java/com/xtrmetl/etl/documentation/DurableJobCancellationDocumentationTest.java diff --git a/etl-service/src/test/java/com/xtrmetl/etl/documentation/DurableJobCancellationDocumentationTest.java b/etl-service/src/test/java/com/xtrmetl/etl/documentation/DurableJobCancellationDocumentationTest.java new file mode 100644 index 00000000..6ff9d0f4 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/documentation/DurableJobCancellationDocumentationTest.java @@ -0,0 +1,121 @@ +package com.xtrmetl.etl.documentation; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Keeps the durable cancellation API, race contract, replay identity, rollout, rollback, and + * changelog aligned. + */ +class DurableJobCancellationDocumentationTest { + + @Test + void operationsRunbookDocumentsTheAuthoritativeCancellationContract() throws IOException { + String runbook = read("docs/operations/durable-job-cancellation.md") + .replaceAll("\\s+", " "); + + assertTrue(runbook.contains("POST /api/etl/jobs/{job_record_id}/cancellation")); + assertTrue(runbook.contains("Idempotency-Replayed: false")); + assertTrue(runbook.contains("Idempotency-Replayed: true")); + assertTrue(runbook.contains("etl_job_cancellation_key_reused")); + assertTrue(runbook.contains("etl_job_already_succeeded")); + assertTrue(runbook.contains("etl_job_already_failed")); + assertTrue(runbook.contains("Cancellation commits first")); + assertTrue(runbook.contains("Success commits first")); + assertTrue(runbook.contains("StaleEtlJobLeaseException")); + assertTrue(runbook.contains( + "rolls back its target and `etl_idempotency_records` writes" + )); + assertTrue(runbook.contains("V6__add_etl_job_cancellation.sql")); + assertTrue(runbook.contains("mightyetl.etl.jobs.intake-enabled=false")); + assertTrue(runbook.contains("does not claim arbitrary external side-effect reversal")); + assertTrue(runbook.contains( + "Never silently map `CANCELLED` to `FAILED` or `SUCCEEDED`" + )); + assertTrue(runbook.contains("RFC 9110")); + assertTrue(runbook.contains("RFC 9457")); + assertTrue(runbook.contains("PostgreSQL Global Development Group. (2026)")); + } + + @Test + void designAndPlanKeepDatabaseAuthorityAndVerificationExplicit() throws IOException { + String design = read( + "docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md" + ).replaceAll("\\s+", " "); + String plan = read( + "docs/superpowers/plans/2026-08-06-durable-job-cancellation.md" + ).replaceAll("\\s+", " "); + + assertTrue(design.contains("One conditional `UPDATE` is the cancellation authority")); + assertTrue(design.contains("Exactly one terminal outcome wins")); + assertTrue(design.contains("same-key replay")); + assertTrue(design.contains("transactional target effects")); + assertTrue(plan.contains("Added production statement and branch coverage remains 100%")); + assertTrue(plan.contains("No project test may be skipped")); + assertTrue(plan.contains("Run all verification")); + } + + @Test + void doctoringDocumentsDomainSeparatedReplayIdentityAndCompatibility() throws IOException { + String evidence = read( + "docs/doctoring/durable-job-cancellation-key-domain-separation.md" + ).replaceAll("\\s+", " "); + + assertTrue(evidence.contains("mightyetl:durable-job-cancellation:v1:")); + assertTrue(evidence.contains("principal_scope_hash")); + assertTrue(evidence.contains("job_record_id")); + assertTrue(evidence.contains("normalized_cancellation_key")); + assertTrue(evidence.contains("same normalized raw key")); + assertTrue(evidence.contains("another job in the same principal namespace")); + assertTrue(evidence.contains("another principal namespace")); + assertTrue(evidence.contains("EtlJobCancellationKeyDomainIntegrationTest")); + assertTrue(evidence.contains("Changing it would make every existing cancelled row fail")); + assertTrue(evidence.contains("does not claim to use cSHAKE")); + assertTrue(evidence.contains("NIST Special Publication 800-185")); + } + + @Test + void changelogRecordsTheBuyerVisibleCancellationSlice() throws IOException { + String changelog = read("CHANGELOG.md").replaceAll("\\s+", " "); + + assertTrue(changelog.contains("Owner-scoped durable-job cancellation")); + assertTrue(changelog.contains("CANCELLED")); + assertTrue(changelog.contains("cancellation_key_hash")); + assertTrue(changelog.contains("Cancellation-first")); + assertTrue(changelog.contains("transactional target and response-ledger effects")); + } + + private static String read(String relativePath) throws IOException { + return Files.readString(projectRoot().resolve(relativePath), StandardCharsets.UTF_8); + } + + /** + * Finds the reactor root from repository-root or module-scoped Maven execution. + * + * @return repository root containing the Maven reactor + */ + private static Path projectRoot() { + Path current = Paths.get(System.getProperty("user.dir")).toAbsolutePath(); + Path lastPomParent = null; + while (current != null) { + if (Files.exists(current.resolve(".git"))) { + return current; + } + if (Files.exists(current.resolve("pom.xml"))) { + lastPomParent = current; + } + current = current.getParent(); + } + if (lastPomParent != null) { + return lastPomParent; + } + throw new IllegalStateException("Could not find project root"); + } +} From 1fd2691dde1ebfd235c53703e6018eee11249f04 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 11 Aug 2026 09:39:20 +0900 Subject: [PATCH 18/18] docs(etl): restore cancellation operating evidence --- CHANGELOG.md | 1 + ...-job-cancellation-key-domain-separation.md | 88 +++++++ docs/operations/durable-job-cancellation.md | 244 ++++++++++++++++++ .../2026-08-06-durable-job-cancellation.md | 91 +++++++ ...6-08-06-durable-job-cancellation-design.md | 99 +++++++ 5 files changed, 523 insertions(+) create mode 100644 docs/doctoring/durable-job-cancellation-key-domain-separation.md create mode 100644 docs/operations/durable-job-cancellation.md create mode 100644 docs/superpowers/plans/2026-08-06-durable-job-cancellation.md create mode 100644 docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md diff --git a/CHANGELOG.md b/CHANGELOG.md index c5e12622..68e92d3b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Owner-scoped durable-job cancellation now commits an idempotent `CANCELLED` terminal state through one principal-bound conditional update, stores the domain-separated `cancellation_key_hash` without raw identity, clears payload and lease state, invalidates stale workers, removes cancelled-job polling guidance, and exposes a no-store authenticated cancellation resource. Cancellation-first races roll back transactional target and response-ledger effects; external non-transactional side effects remain outside this guarantee. - Owner-scoped durable-job status responses now emit deterministic weak SHA-256 `ETag` validators; ordinary and wildcard `If-None-Match` requests return an empty RFC 9110 `304 Not Modified` response only after authenticated owner-safe lookup, while `Cache-Control: no-store` remains unchanged. - Active durable-job status responses now emit an RFC 9110 `Retry-After` delay for `PENDING` and `RUNNING` states only when local worker execution is enabled, derived from the bounded worker fixed-delay configuration with upward whole-second rounding; terminal states and intake-only maintenance mode omit the advisory. - 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. diff --git a/docs/doctoring/durable-job-cancellation-key-domain-separation.md b/docs/doctoring/durable-job-cancellation-key-domain-separation.md new file mode 100644 index 00000000..498572f0 --- /dev/null +++ b/docs/doctoring/durable-job-cancellation-key-domain-separation.md @@ -0,0 +1,88 @@ +# Durable-job cancellation replay identity domain separation + +## Decision + +mightyETL never stores a raw cancellation `Idempotency-Key`. It stores one lowercase SHA-256 replay identity computed from an explicit versioned domain, the authenticated-principal hash, the durable job identifier, and the normalized key: + +```text +SHA-256( + "mightyetl:durable-job-cancellation:v1:" + || principal_scope_hash + || ":" + || job_record_id + || ":" + || normalized_cancellation_key +) +``` + +This value is used only to prove that a later request addresses the same principal, the same job, and the same semantic cancellation key. It is not an authentication credential and grants no job authority; every transition and replay read independently binds the owner hash and job identifier in SQL. + +## Threat addressed + +Hashing the raw client key alone would hide its plaintext but preserve equality across every row. A database observer could correlate two jobs or tenants that reused the same cancellation key even though ordinary API representations never reveal the key or hash. + +The versioned contextual prefix and explicit principal/job components partition the replay identity. The same normalized raw key therefore produces: + +- the same stored hash for the same principal and job, preserving deterministic replay; +- a different stored hash for another job in the same principal namespace; +- a different stored hash for the same job-shaped identifier under another principal namespace; +- a different stored hash after a deliberate future domain-version change. + +The implementation does not claim to use cSHAKE, TupleHash, or another NIST SP 800-185 primitive. It uses the existing SHA-256 utility with an unambiguous fixed-layout contextual input. NIST SP 800-185 is cited as primary methodological evidence for customization and tuple/domain separation concepts, not as an implementation-conformance claim. NIST announced in March 2025 that SP 800-185 will be revised; until a replacement is finalized, this document cites the current final publication and the revision decision separately. + +## Compatibility boundary + +The exact domain string is persisted protocol behavior: + +```text +mightyetl:durable-job-cancellation:v1: +``` + +Changing it would make every existing cancelled row fail same-key replay comparison. A future `v2` requires an explicit migration and dual-read compatibility window or a documented replay-breaking release. Silent replacement of the prefix is prohibited. + +The current concatenation is unambiguous because: + +- `principal_scope_hash` is exactly 64 lowercase hexadecimal characters; +- the separator is a literal colon; +- `job_record_id` is the canonical UUID text form; +- the second separator is a literal colon; +- the normalized cancellation key follows the bounded safe-ASCII profile and is the final component. + +If a later version introduces variable-width or independently nested components, use explicit length prefixes or a tuple-hash construction rather than extending this layout informally. + +## Test-first evidence + +`EtlJobCancellationKeyDomainIntegrationTest` uses the same raw cancellation key for: + +1. two different jobs owned by one principal; +2. one job owned by another principal. + +The test requires three distinct 64-character stored hashes. Existing service-integration tests separately prove that a quoted and legacy-raw representation of the same key on the same job replay one cancellation, while a genuinely different key fails with `etl_job_cancellation_key_reused`. + +## Privacy and logging + +The raw cancellation key and resulting hash remain absent from: + +- HTTP response bodies and headers; +- RFC 9457 problem details; +- ordinary logs; +- metric labels; +- status, list, polling, and ETag representations; +- worker lease models. + +The hash is a replay identity stored in `cancellation_key_hash`; it is not safe to publish merely because it is one-way. Database access, backups, exports, and support tooling must treat it as internal pseudonymous security data. + +## Rollback + +Rolling application code back across this change can make same-key replay behavior inconsistent if an older binary derives a raw-key-only hash. Keep the domain-separated implementation deployed while rows created by it are active. A rollback requires either: + +- retaining the new comparison algorithm in the older release line; or +- a reviewed data migration with explicit compatibility evidence. + +Never rewrite hashes from user-supplied guesses and never log candidate keys while diagnosing replay mismatches. + +## References — APA 7th + +Kelsey, J., Chang, S., & Perlner, R. (2016). *SHA-3 derived functions: cSHAKE, KMAC, TupleHash, and ParallelHash* (NIST Special Publication 800-185). National Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-185 + +National Institute of Standards and Technology. (2025, March 12). *Decision to update FIPS 202 and revise SP 800-185*. https://csrc.nist.gov/news/2025/decision-to-update-fips-202-and-revise-sp-800-185 diff --git a/docs/operations/durable-job-cancellation.md b/docs/operations/durable-job-cancellation.md new file mode 100644 index 00000000..ecc33022 --- /dev/null +++ b/docs/operations/durable-job-cancellation.md @@ -0,0 +1,244 @@ +# Durable ETL job cancellation + +## Purpose + +Authenticated operators can stop a durable ETL job that is still `PENDING` or `RUNNING` through: + +```http +POST /api/etl/jobs/{job_record_id}/cancellation HTTP/1.1 +Authorization: Basic +Idempotency-Key: "70dc8b50-e8b2-4e1a-8c5f-d84814708a77" +``` + +A successful response proves that the database transition committed or that the same principal, +job identifier, and normalized cancellation key had already committed. HTTP request acceptance by +itself is never treated as cancellation success. + +The first slice establishes a database-owned terminal state. It does not forcibly terminate a Java +thread, interrupt arbitrary connector computation, or compensate a non-transactional external +warehouse. + +## HTTP contract + +A first cancellation returns: + +```http +HTTP/1.1 200 OK +Cache-Control: no-store +Idempotency-Replayed: false +ETag: W/"" +Content-Type: application/json + +{ + "jobRecordId": "cf4f083f-8c90-4f34-a8b6-b53761de44ef", + "jobStatus": "CANCELLED", + "attemptCount": 1, + "createdAt": "2026-08-05T01:00:00Z", + "updatedAt": "2026-08-06T03:00:00Z" +} +``` + +Repeating the same semantic key returns `Idempotency-Replayed: true` and the same terminal resource. +A different key for the already-cancelled job returns +`422 etl_job_cancellation_key_reused`. Malformed, missing, and foreign-owned identifiers share the +same `404 etl_job_not_found` response. Cancellation after committed success or failure returns +`409 etl_job_already_succeeded` or `409 etl_job_already_failed` under RFC 9110 current-state conflict +semantics. + +Every success and covered failure uses `Cache-Control: no-store`. The response exposes no payload, +principal, cancellation key, key hash, lease identifier, SQL, exception message, or target identity. + +## State machine + +```mermaid +stateDiagram-v2 + [*] --> PENDING + PENDING --> RUNNING: lease-fenced claim + PENDING --> CANCELLED: owner cancellation + RUNNING --> PENDING: exact-lease retry + RUNNING --> SUCCEEDED: target + ledger + exact-lease commit + RUNNING --> FAILED: exact-lease terminal failure + RUNNING --> CANCELLED: owner cancellation wins + CANCELLED --> CANCELLED: same-key replay +``` + +`SUCCEEDED`, `FAILED`, and `CANCELLED` are terminal. Cancelled status never receives `Retry-After`. +Its state and `updatedAt` value also invalidate every earlier weak status `ETag`. + +## Database authority + +`EtlJobService.cancelOwned` first validates the job identifier, cancellation key, and authenticated +principal, then performs one conditional update inside a Spring transaction: + +```sql +UPDATE etl_job_records +SET job_status = 'CANCELLED', + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + cancellation_key_hash = ?, + cancellation_code = 'etl_job_cancelled_by_owner', + job_cancelled_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP +WHERE job_record_id = ? + AND principal_scope_hash = ? + AND job_status IN ('PENDING', 'RUNNING'); +``` + +The update count is the authority. A follow-up owner-scoped read classifies a zero-row result as: + +- identical `CANCELLED` replay; +- conflicting cancellation key; +- already `SUCCEEDED`; +- already `FAILED`; +- an active row whose concurrent transition is still unresolved; or +- owner-safe not found. + +The raw principal and key are never stored. Their lowercase SHA-256 values are used only for owner +selection and replay identity. + +## Cancellation-versus-success race + +### Cancellation commits first + +```mermaid +sequenceDiagram + participant C as Cancellation request + participant D as PostgreSQL + participant W as Worker transaction + C->>D: conditional PENDING/RUNNING → CANCELLED + D-->>C: one row committed + W->>D: target + response ledger writes + W->>D: markSucceeded(exact former lease) + D-->>W: zero rows updated + W-->>D: rollback target + ledger +``` + +The cancelled row has no lease fields. The former worker's exact-live-lease success predicate updates +zero rows and raises `StaleEtlJobLeaseException`; Spring rolls back its target and +`etl_idempotency_records` writes. + +### Success commits first + +```mermaid +sequenceDiagram + participant W as Worker transaction + participant D as PostgreSQL + participant C as Cancellation request + W->>D: target + response ledger + SUCCEEDED commit + C->>D: conditional PENDING/RUNNING → CANCELLED + D-->>C: zero rows updated + C->>D: owner-scoped terminal read + D-->>C: SUCCEEDED + C-->>C: 409 etl_job_already_succeeded +``` + +Exactly one terminal state wins. The endpoint never rewrites `SUCCEEDED` or `FAILED` as cancelled. + +## Migration + +`V6__add_etl_job_cancellation.sql` adds these descriptive multi-word `snake_case` columns: + +- `cancellation_key_hash`; +- `cancellation_code`; +- `job_cancelled_at`. + +It replaces lifecycle checks so that: + +- `PENDING` and `RUNNING` retain a payload; +- `SUCCEEDED`, `FAILED`, and `CANCELLED` have no payload; +- only `RUNNING` has lease fields; +- only `FAILED` has `failure_code`; +- only `CANCELLED` has the three cancellation fields; +- hash and code values satisfy bounded fixed formats. + +The migration is transactional. Before production rollout, rehearse it against a representative +PostgreSQL 18 copy and confirm that no out-of-contract legacy row violates the replacement checks. +Monitor migration duration, lock wait, transaction age, replication lag, and application error rates. + +## Rollout + +1. Verify PR exact-head CI on Ubuntu, macOS, and Windows with no skipped project test. +2. Verify dependency review, CycloneDX SBOM, SAST, security scan, unresolved-thread, and independent + current-head approval gates. +3. Apply Flyway V6 before serving the new route. +4. Keep `mightyetl.etl.jobs.intake-enabled=false` during a conservative schema-only rollout if the + deployment process cannot guarantee application/schema ordering. +5. Enable the new application build and perform an owner-isolation smoke test with a disposable job. +6. Verify a first cancellation, same-key replay, different-key rejection, and a status read. +7. Confirm cancelled rows have null payload and lease fields and a fixed cancellation code. +8. Observe worker `stale` outcomes during deliberate running-job cancellation; this is expected + fencing evidence, not a duplicate-execution success. + +## Monitoring + +The cancellation endpoint uses fixed observation name `etl.jobs.cancel`. Do not attach job IDs, +principals, raw keys, key hashes, lease IDs, payloads, SQL, exception classes, messages, target +identities, or queue depth as metric labels. + +Monitor at least: + +- request rate and HTTP outcome count; +- cancellation latency; +- database update latency and lock waits; +- worker `stale` outcome changes; +- cancellation replay and key-conflict rate; +- cancelled rows retaining payload or lease fields, which must remain zero; +- target or ledger effects associated with cancellation-first tests, which must remain zero. + +## Incident response + +### Cancellation returns `etl_job_cancellation_in_progress` + +Re-read the owner-scoped status. A concurrent claim, retry, success, failure, or cancellation may have +won after the request's conditional update. Do not retry with a new idempotency key until the current +terminal or active state is understood. + +### Worker reports stale after cancellation + +This is the expected safety outcome when cancellation invalidates a running lease. Confirm the target +and response-ledger transaction rolled back. Repeated stale outcomes without operator cancellations +may indicate lease expiry, another worker, or database clock/latency problems. + +### Cancelled row retains payload or lease data + +Treat this as a high-severity lifecycle integrity incident. Stop intake and workers, preserve the row +and transaction evidence, verify the deployed schema constraints and application SHA, and do not +manually rewrite the state until the root cause and rollback effects are understood. + +## Rollback + +Stop serving the cancellation endpoint before application rollback. Older binaries do not understand +`CANCELLED`, so they must not read or process cancelled rows as if only four states existed. + +Do not drop V6 columns or restore the old status constraint while any cancelled row remains. A +controlled database rollback must first archive cancelled resources and their audit evidence under an +approved retention policy. Never silently map `CANCELLED` to `FAILED` or `SUCCEEDED`. + +After cancelled rows are safely removed and every older binary is deployed, an explicit reviewed +migration may drop the V6 constraints and columns and restore the four-state lifecycle. Do not edit or +repair the applied V6 migration file in place. + +## Connector limitation + +The cancellation-first rollback guarantee is valid for target and response-ledger writes that join the +same transaction and database as the job state. A remote warehouse, file upload, external API, or +message broker that cannot participate in that transaction requires connector-native cancellation, +idempotency, or compensation before the same guarantee can be advertised. This release deliberately +does not claim arbitrary external side-effect reversal. + +## References — APA 7th + +Fielding, R., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC 9110). RFC Editor. +https://www.rfc-editor.org/rfc/rfc9110 + +Nottingham, M., Wilde, E., & Dalal, S. (2023). *Problem details for HTTP APIs* (RFC 9457). RFC Editor. +https://www.rfc-editor.org/rfc/rfc9457 + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Data consistency checks at +the application level*. https://www.postgresql.org/docs/18/applevel-consistency.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: UPDATE*. +https://www.postgresql.org/docs/18/sql-update.html diff --git a/docs/superpowers/plans/2026-08-06-durable-job-cancellation.md b/docs/superpowers/plans/2026-08-06-durable-job-cancellation.md new file mode 100644 index 00000000..760dcd0a --- /dev/null +++ b/docs/superpowers/plans/2026-08-06-durable-job-cancellation.md @@ -0,0 +1,91 @@ +# Durable ETL Job Cancellation Implementation Plan + +**Goal:** Deliver owner-safe, idempotent, lease-fenced cancellation for pending and running durable ETL jobs on the repaired stack rooted at the exact conditional-status predecessor. + +**Architecture:** Keep PostgreSQL as terminal-state authority. A single conditional cancellation update clears payload and lease state, records only bounded pseudonymous replay identity, and lets exact-lease worker predicates reject stale commits. The HTTP layer exposes only the existing operator-safe status model. + +## Global constraints + +- Preserve standalone operation and modular MSA integration. +- Do not weaken branch protection, security, independent review, or exact-head evidence. +- Do not use `COPILOT_GITHUB_TOKEN` or invent credentials. +- Database objects use descriptive multi-word `snake_case` names. +- Raw principals, cancellation keys, payloads, hashes, lease identifiers, SQL, and exception messages do not enter client responses, logs, or metric labels. +- Added production statement and branch coverage remains 100%. +- Every public production API has beginner-readable Javadoc. +- No project test may be skipped. +- Update authoritative operations, doctoring, and `CHANGELOG.md` evidence before merge. + +## Task 1 — lifecycle and migration + +1. Write fail-first lifecycle assertions for `CANCELLED`, cancellation identity, payload clearing, and lease clearing. +2. Add `V6__add_etl_job_cancellation.sql` with immutable Flyway history and descriptive constraints. +3. Verify migration documentation and clean-install/upgrade invariants. + +Acceptance: `CANCELLED` is terminal; only active states retain payload; only `RUNNING` owns lease fields; only `CANCELLED` owns cancellation fields. + +## Task 2 — service cancellation authority + +1. Write fail-first service tests for pending/running cancellation, same-key replay, different-key rejection, owner isolation, completed-state conflicts, and invalid input before JDBC access. +2. Implement `EtlJobCancellation`, `EtlJobStatus.CANCELLED`, stable RFC 9457 errors, and `EtlJobService.cancelOwned`. +3. Keep one conditional update as authority; classify a zero-row result through an owner-scoped read. +4. Verify replay identity is domain-separated by principal and job. + +Acceptance: one committed cancellation or one competing terminal transition wins; raw identity never persists. + +## Task 3 — authenticated HTTP action + +1. Add failing controller tests before the route. +2. Expose `POST /api/etl/jobs/{jobRecordId}/cancellation`. +3. Require authentication and a bounded `Idempotency-Key` before service access. +4. Return `200`, `Cache-Control: no-store`, a weak `ETag`, `Idempotency-Replayed`, and only operator-safe status fields. +5. Cover malformed identifiers, typed conflicts, data-access failures, and unexpected failures without leaking internal messages. + +Acceptance: exact controller tests and strict controller coverage pass. + +## Task 4 — worker and representation compatibility + +1. Prove cancellation clears the worker lease and prevents former exact-lease success from committing. +2. Prove transactional target and response-ledger effects roll back when cancellation wins first. +3. Prove `CANCELLED` never retains `Retry-After`. +4. Prove cancellation invalidates the prior conditional status validator. +5. Prove success-first and cancellation-first races each leave exactly one terminal state. + +Acceptance: concurrency tests demonstrate stale-lease rejection rather than duplicate terminal commit. + +## Task 5 — operator evidence and rollback + +1. Add documentation contract tests before missing docs are restored. +2. Document endpoint semantics, replay, database authority, race outcomes, observability, migration rehearsal, rollback, and external connector limitation. +3. Record domain-separated replay identity in doctoring with current NIST publication/revision evidence. +4. Update `CHANGELOG.md` under Unreleased with the buyer-visible cancellation slice. +5. Preserve old PR #133 and its fail-first history until this replacement proves equivalent or stronger exact-head behavior; old checks and approvals do not transfer. + +## Run all verification + +Run all verification after the final source/documentation change: + +```bash +./mvnw -B test +git diff --check +``` + +Then refetch the exact head, live base tip, CI, Dependency Review, SBOM, commit statuses, formal reviews, unresolved threads, and mergeability. SAST or security evidence absent because this PR is stacked on a non-default base remains not passing and must not be inferred from a predecessor or synthetic merge. + +## Merge and rollout acceptance + +- Exact ancestry from the immediate predecessor remains intact. +- Exact-head CI, dependency, SBOM, security, coverage, and commit-status gates pass where required. +- Zero actionable unresolved review thread remains. +- A qualifying independent non-author formal `APPROVED` review is anchored to the exact unchanged head. +- Migration/rollback and compatibility evidence are complete. +- Protected merge uses expected-head semantics without bypass. +- Post-merge operational proof confirms the protected branch serves the cancellation contract before the next dependent slice deepens the stack. + +## References — APA 7th + +Fielding, R., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC 9110). RFC Editor. https://www.rfc-editor.org/rfc/rfc9110 + +Nottingham, M., Wilde, E., & Dalal, S. (2023). *Problem details for HTTP APIs* (RFC 9457). RFC Editor. https://www.rfc-editor.org/rfc/rfc9457 + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: UPDATE*. https://www.postgresql.org/docs/18/sql-update.html diff --git a/docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md b/docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md new file mode 100644 index 00000000..40a5b402 --- /dev/null +++ b/docs/superpowers/specs/2026-08-06-durable-job-cancellation-design.md @@ -0,0 +1,99 @@ +# Durable ETL Job Cancellation Design + +## Purpose + +mightyETL durable jobs need an operator-controlled stop action that preserves owner isolation, idempotent replay, database lease fencing, and truthful transactional guarantees. Cancellation is therefore modeled as a database-owned terminal transition, not as a best-effort thread interrupt. + +## API contract + +```http +POST /api/etl/jobs/{job_record_id}/cancellation +Authorization: +Idempotency-Key: "client-generated-safe-key" +``` + +A successful first cancellation returns the existing operator-safe status representation, `200 OK`, `Cache-Control: no-store`, a weak `ETag`, and `Idempotency-Replayed: false`. The same principal, job, and normalized key replays the terminal resource with `Idempotency-Replayed: true`. + +## State machine + +```mermaid +stateDiagram-v2 + [*] --> PENDING + PENDING --> RUNNING: exact database claim + PENDING --> CANCELLED: owner cancellation + RUNNING --> SUCCEEDED: target + ledger + exact lease commit + RUNNING --> FAILED: exact lease terminal failure + RUNNING --> PENDING: exact lease retry release + RUNNING --> CANCELLED: owner cancellation wins row update + CANCELLED --> CANCELLED: same-key replay +``` + +`SUCCEEDED`, `FAILED`, and `CANCELLED` are terminal. Exactly one terminal outcome wins. Completed success or failure is never rewritten as cancellation. + +## Database authority + +One conditional `UPDATE` is the cancellation authority: + +```sql +UPDATE etl_job_records +SET job_status = 'CANCELLED', + request_payload = NULL, + failure_code = NULL, + lease_claim_id = NULL, + lease_owner_id = NULL, + lease_expires_at = NULL, + cancellation_key_hash = ?, + cancellation_code = 'etl_job_cancelled_by_owner', + job_cancelled_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP +WHERE job_record_id = ? + AND principal_scope_hash = ? + AND job_status IN ('PENDING', 'RUNNING'); +``` + +The affected-row count determines whether cancellation committed. A zero-row update is classified only by an owner-scoped read as same-key replay, conflicting key, already succeeded, already failed, concurrent active transition, or owner-safe not found. Read-then-write logic is not cancellation authority. + +## Replay identity + +Raw principals and raw cancellation keys are never persisted. Replay identity uses an explicit versioned cancellation domain together with the authenticated-principal hash, job identifier, and normalized key. The same-key replay is deterministic for one principal/job while the same client key on another job or principal produces a distinct stored identity. + +## Race and transaction semantics + +When cancellation commits first, it clears the live lease. A former worker can continue computing in memory, but its exact-live-lease success transition must affect zero rows and raise `StaleEtlJobLeaseException`. Transactional target effects and the durable response ledger then roll back with that worker transaction. + +When success commits first, cancellation affects zero rows, the owner-scoped terminal read observes `SUCCEEDED`, and the API returns `409 etl_job_already_succeeded`. The corresponding failed-job race returns `409 etl_job_already_failed`. + +The phrase transactional target effects is deliberate: this guarantee covers effects participating in the same transaction as the job-state and response-ledger commit. Remote warehouses, file uploads, external APIs, and message brokers require connector-native idempotency, cancellation, or compensation before equivalent rollback can be claimed. + +## HTTP representation + +The wire representation exposes `jobStatus=CANCELLED` without exposing cancellation hashes, codes, raw keys, principals, payloads, or lease identifiers. Cancellation changes lifecycle state and update time, so the weak status `ETag` changes. `CANCELLED` is terminal and therefore never retains `Retry-After` polling guidance. + +## Persistence and migration + +`V6__add_etl_job_cancellation.sql` adds descriptive multi-word `snake_case` columns `cancellation_key_hash`, `cancellation_code`, and `job_cancelled_at`. Lifecycle checks require payloads only while work is active, leases only while `RUNNING`, failure metadata only for `FAILED`, and cancellation metadata only for `CANCELLED`. + +The migration is immutable once applied. Rollback must first stop serving cancellation, preserve or archive cancelled-row evidence, and use a new reviewed migration. Never silently map `CANCELLED` to `FAILED` or `SUCCEEDED`. + +## Error taxonomy + +- `400 etl_job_cancellation_key_required`: missing or invalid key. +- `404 etl_job_not_found`: malformed, missing, or foreign-owned identifier. +- `409 etl_job_cancellation_in_progress`: an eligible transition remained unresolved after the authoritative update. +- `409 etl_job_already_succeeded`: success won the race. +- `409 etl_job_already_failed`: failure won the race. +- `422 etl_job_cancellation_key_reused`: a different key addresses an already-cancelled job. + +Problem responses use the existing RFC 9457 handler and never disclose SQL, exception text, principal identity, keys, hashes, payloads, or leases. + +## Verification contract + +Exact-head verification must prove pending and running cancellation, payload and lease clearing, same-key replay, different-key rejection, owner isolation, completed-state conflicts, malformed input, controller failure mapping, stale-lease fencing, cancellation polling terminality, conditional-validator change, migration invariants, and rollback documentation. Added production statements and branches remain fully covered. + +## References — APA 7th + +Fielding, R., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC 9110). RFC Editor. https://www.rfc-editor.org/rfc/rfc9110 + +Nottingham, M., Wilde, E., & Dalal, S. (2023). *Problem details for HTTP APIs* (RFC 9457). RFC Editor. https://www.rfc-editor.org/rfc/rfc9457 + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: UPDATE*. https://www.postgresql.org/docs/18/sql-update.html