diff --git a/CHANGELOG.md b/CHANGELOG.md index 57a0b362..1cedd20f 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 +- Active durable-job status responses now emit an RFC 9110 `Retry-After` delay for `PENDING` and `RUNNING` states only when local worker execution is enabled, derived from the bounded worker fixed-delay configuration with upward whole-second rounding; terminal states and intake-only maintenance mode omit the advisory. - The durable job pagination index now uses PostgreSQL `CREATE INDEX CONCURRENTLY` with a migration-local Flyway `executeInTransaction=false` companion configuration, preserving production writers and documenting invalid-index recovery and concurrent rollback. - Durable job operators can now list only their own jobs through bounded newest-first keyset pagination with canonical opaque cursors, deterministic timestamp-plus-UUID ordering, `Cache-Control: no-store`, and RFC 8288 next-page links without offset drift or cross-tenant existence leakage. - The durable-job claim eligibility index now builds in a separate PostgreSQL `CREATE INDEX CONCURRENTLY` migration with Flyway non-transactional script configuration and session-level PostgreSQL migration locking, preserving normal job writes during rollout while keeping lease columns and constraints transactional. @@ -33,6 +34,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Controller-scoped polling advice, deterministic active/terminal lifecycle tests, disabled-worker fail-closed behavior, sub-second rounding coverage, rollback guidance, and APA 7th standards evidence in `docs/etl/durable-job-polling.md`. - Owner-scoped durable job list models and HTTP contract, strict cursor and page-limit validation, one-extra-row next-page detection, the descriptive `etl_job_owner_pagination_index`, deterministic tenant-isolation and equal-timestamp tests, migration rollback guidance, and APA 7th standards evidence in `docs/etl/durable-job-intake.md`. - A production rollout and invalid-index recovery runbook for the nonblocking durable-job claim index: `docs/operations/durable-job-claim-index-rollout.md`. - PostgreSQL `FOR UPDATE SKIP LOCKED` durable-job claiming, per-process and per-claim lease fencing, expiry reclaim, bounded attempts, exact-live-lease transitions, terminal payload clearing, stable failure codes, and finite-cardinality worker metrics. @@ -70,6 +72,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- Polling advice exposes only a bounded delay integer and is omitted when local execution is disabled or terminal; it never contains job, lease, principal, key, hash, payload, SQL, exception, target, or queue-depth data. - Job-list cursors contain only ordering keys, never authority or sensitive values; every page query independently binds the hashed authenticated principal, malformed and non-canonical cursors fail before database access, and list responses exclude payloads, principals, keys, hashes, lease identifiers, SQL, and exception text. - Durable-worker metrics and ordinary logs exclude payloads, raw principals, raw idempotency keys, hashes, job and lease identifiers, SQL, exception messages, and unbounded exception labels. - Retained payload or response-ledger identity conflicts fail closed with `etl_job_integrity_failure`; an expired or superseded lease rolls back target, ledger, and terminal-state effects. @@ -112,7 +115,7 @@ existing xtrmETL platform. - Complete data model specifications - API specifications with examples - Deployment architecture - - Use cases and scenarios + - Use cases - Future enhancements roadmap - Success metrics and KPIs - Risk assessment and mitigation strategies diff --git a/docs/etl/durable-job-polling.md b/docs/etl/durable-job-polling.md new file mode 100644 index 00000000..d2e72c6b --- /dev/null +++ b/docs/etl/durable-job-polling.md @@ -0,0 +1,78 @@ +# Durable job polling advisory + +## Purpose + +Authenticated clients poll `GET /api/etl/jobs/{job_record_id}` while a durable ETL job is active. Without a machine-readable cadence, independently implemented clients tend to use arbitrary tight loops that waste control-plane capacity and produce inconsistent operator behavior. + +mightyETL emits the RFC 9110 `Retry-After` response field only when both conditions hold: + +1. the owner-scoped status representation is `PENDING` or `RUNNING`; and +2. durable-job execution is explicitly enabled. + +The field is advisory: it does not grant authority, alter the job state machine, extend a worker lease, or replace client-side exponential backoff and jitter. Intake-only maintenance mode does not emit a processing cadence because no local worker is available to advance accepted jobs. + +## Wire contract + +An active job with the worker enabled returns the existing operator-safe status representation, `Cache-Control: no-store`, and a whole-second polling delay: + +```http +HTTP/1.1 200 OK +Cache-Control: no-store +Retry-After: 5 +Content-Type: application/json + +{ + "jobRecordId": "cf4f083f-8c90-4f34-a8b6-b53761de44ef", + "jobStatus": "RUNNING", + "attemptCount": 1, + "createdAt": "2026-08-05T01:00:00Z", + "updatedAt": "2026-08-05T01:00:05Z" +} +``` + +`SUCCEEDED` and `FAILED` are terminal and omit `Retry-After`. An active job also omits the field when `mightyetl.etl.jobs.worker.enabled=false`, preventing a maintenance-mode intake deployment from advertising a cadence that cannot advance work. Submission responses, list responses, problem responses, and unrelated controllers are unchanged. + +## Cadence derivation + +The response advice reads the validated `xtrmetl.etl.jobs.worker.enabled` and `xtrmetl.etl.jobs.worker.fixed-delay-milliseconds` values. The preferred `mightyetl.*` configuration namespace continues to map through the existing compatibility alias processor. The delay is already bounded from one millisecond through one day. + +When the worker is enabled, the wire value is calculated as: + +```text +retry_after_seconds = ceil(fixed_delay_milliseconds / 1000) +``` + +Fractional seconds are rounded upward, so one millisecond through one second advertises `Retry-After: 1`, 1,001 milliseconds advertises `Retry-After: 2`, and the one-day maximum advertises `Retry-After: 86400`. This prevents a valid positive scheduler delay from becoming a zero-second tight-polling instruction. + +The worker schedule remains a local execution setting rather than a completion promise. Queue depth, target latency, retries, process restarts, and lease recovery can make a job remain active for multiple polling intervals. When execution is disabled, the absent field signals that the service has no active local cadence to advertise; clients must use their maintenance-window or operator-escalation policy instead of tight polling. + +## Security and privacy boundary + +The polling field contains only a bounded integer. It exposes no request payload, authenticated principal, idempotency key, internal hash, job or lease identifier, SQL, exception message, target identity, or queue depth. + +The response still requires authenticated owner scope. A malformed, absent, or foreign-owned identifier returns the same owner-safe not-found problem and does not receive an active-job polling advisory. The advice is scoped to `EtlJobController` and modifies only `EtlJobStatusResponse` bodies. + +## Client guidance + +Clients should: + +1. treat `Retry-After` as the minimum delay before the next status request; +2. add bounded jitter when many jobs are polled concurrently; +3. apply a larger local backoff after transport failures or `429`/`503` responses; +4. stop polling when the returned state is `SUCCEEDED` or `FAILED`; +5. treat an absent field on an active job as a maintenance or externally managed execution condition and use an operator-approved fallback interval; +6. retain their own overall timeout and operator escalation policy. + +Clients must not interpret the field as a guarantee that the job will finish within one interval. + +## Rollback + +The slice adds no database object, migration, persisted state, or API body field. Rolling back the application removes the advisory header while preserving submission, list, status, worker, and lease behavior. Older clients already operate without the header; newer clients must tolerate its absence because HTTP response fields are optional unless a separate client contract makes them mandatory. + +## Standards basis + +RFC 9110 Section 10.2.3 defines `Retry-After` as either an HTTP-date or a non-negative decimal number of seconds indicating how long a user agent ought to wait before a follow-up request. mightyETL uses the bounded `delay-seconds` form because it is deterministic, timezone-independent, and directly derived from an enabled worker's configured cadence. + +### Reference + +Fielding, R., Nottingham, M., & Reschke, J. (2022). *HTTP semantics* (RFC 9110). RFC Editor. https://www.rfc-editor.org/rfc/rfc9110 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 new file mode 100644 index 00000000..3973f581 --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlJobPollingAdvice.java @@ -0,0 +1,132 @@ +package com.xtrmetl.etl.controller; + +import com.xtrmetl.etl.job.EtlJobStatusResponse; +import com.xtrmetl.etl.job.EtlJobWorkerProperties; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.core.MethodParameter; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.http.server.ServerHttpRequest; +import org.springframework.http.server.ServerHttpResponse; +import org.springframework.lang.Nullable; +import org.springframework.web.bind.annotation.ControllerAdvice; +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyAdvice; + +import java.util.Objects; + +/** + * Adds a bounded RFC 9110 polling advisory to active durable-job status responses. + * + *

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.

+ * + *

The header is only client guidance. PostgreSQL lease fencing, principal-scoped authorization, + * lifecycle validation, rate limiting, and client-side backoff remain independent correctness and + * operational boundaries.

+ */ +@ConditionalOnBooleanProperty( + prefix = "xtrmetl.etl.jobs", + name = "intake-enabled", + havingValue = true, + matchIfMissing = false +) +@ControllerAdvice(assignableTypes = EtlJobController.class) +public class EtlJobPollingAdvice implements ResponseBodyAdvice { + + private static final long MILLISECONDS_PER_SECOND = 1_000L; + + private final boolean workerEnabled; + private final String retryAfterSeconds; + + /** + * Creates polling advice aligned to the configured durable-worker state and cadence. + * + *

{@link EtlJobWorkerProperties} validates the fixed delay as one millisecond through one + * day. Converting with upward rounding therefore always yields a positive RFC 9110 + * {@code delay-seconds} value from one through 86,400 when the worker is enabled. The computed + * value remains dormant while the worker is disabled.

+ * + * @param workerProperties validated durable-worker scheduling configuration + * @throws NullPointerException when the configuration is {@code null} + */ + public EtlJobPollingAdvice(EtlJobWorkerProperties workerProperties) { + EtlJobWorkerProperties requiredProperties = Objects.requireNonNull( + workerProperties, + "workerProperties must not be null" + ); + this.workerEnabled = requiredProperties.isEnabled(); + long fixedDelayMilliseconds = requiredProperties.getFixedDelayMilliseconds(); + long roundedSeconds = (fixedDelayMilliseconds + MILLISECONDS_PER_SECOND - 1L) + / MILLISECONDS_PER_SECOND; + this.retryAfterSeconds = Long.toString(roundedSeconds); + } + + /** + * Participates in response handling for the scoped durable-job controller. + * + *

The controller advice annotation already limits invocation to {@link EtlJobController}. + * The body-type decision remains in {@link #beforeBodyWrite(Object, MethodParameter, MediaType, + * Class, ServerHttpRequest, ServerHttpResponse)} so problem, submission, list, and status + * responses follow one deterministic path.

+ * + * @param returnType controller method return metadata + * @param converterType selected HTTP message converter type + * @return always {@code true} within the controller-scoped advice chain + */ + @Override + public boolean supports( + MethodParameter returnType, + Class> converterType + ) { + return true; + } + + /** + * Adds or removes {@code Retry-After} according to worker availability and job lifecycle state. + * + *

Only {@link EtlJobStatusResponse} bodies are modified. Active states set the configured + * rounded delay only when a worker is enabled. Disabled-worker and terminal responses + * explicitly remove the field, protecting maintenance-mode intake, embedded adapters, and + * future response builders from retaining a misleading polling suggestion. All other bodies and + * headers pass through unchanged.

+ * + * @param body response body selected by the controller or exception handler + * @param returnType controller method return metadata + * @param selectedContentType selected response media type + * @param selectedConverterType selected HTTP message converter type + * @param request current server request + * @param response mutable server response + * @return the original response body without replacement + */ + @Override + @Nullable + public Object beforeBodyWrite( + @Nullable Object body, + MethodParameter returnType, + MediaType selectedContentType, + Class> selectedConverterType, + ServerHttpRequest request, + ServerHttpResponse response + ) { + if (body instanceof EtlJobStatusResponse statusResponse) { + switch (statusResponse.jobStatus()) { + case PENDING, RUNNING -> { + if (workerEnabled) { + response.getHeaders().set( + HttpHeaders.RETRY_AFTER, + retryAfterSeconds + ); + } else { + response.getHeaders().remove(HttpHeaders.RETRY_AFTER); + } + } + case SUCCEEDED, FAILED -> response.getHeaders().remove(HttpHeaders.RETRY_AFTER); + } + } + return body; + } +} 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 new file mode 100644 index 00000000..e21dd479 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlJobPollingAdviceTest.java @@ -0,0 +1,171 @@ +package com.xtrmetl.etl.controller; + +import com.xtrmetl.etl.job.EtlJobStatus; +import com.xtrmetl.etl.job.EtlJobStatusResponse; +import com.xtrmetl.etl.job.EtlJobWorkerProperties; +import org.junit.jupiter.api.Test; +import org.springframework.core.MethodParameter; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; +import org.springframework.http.server.ServerHttpRequest; +import org.springframework.http.server.ServerHttpResponse; + +import java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Defines the bounded RFC 9110 polling-advisory contract for durable job status responses. + */ +class EtlJobPollingAdviceTest { + + private static final UUID JOB_RECORD_ID = UUID.fromString( + "cf4f083f-8c90-4f34-a8b6-b53761de44ef" + ); + private static final Instant CREATED_AT = Instant.parse("2026-08-05T01:00:00Z"); + private static final Instant UPDATED_AT = Instant.parse("2026-08-05T01:00:05Z"); + + @Test + void activeJobsAdvertiseTheRoundedWorkerPollingCadence() { + EtlJobWorkerProperties workerProperties = enabledWorkerProperties(); + workerProperties.setFixedDelayMilliseconds(1_001L); + EtlJobPollingAdvice advice = new EtlJobPollingAdvice(workerProperties); + + assertRetryAfter(advice, EtlJobStatus.PENDING, "2"); + assertRetryAfter(advice, EtlJobStatus.RUNNING, "2"); + } + + @Test + void subSecondWorkerCadenceAdvertisesAtLeastOneSecond() { + EtlJobWorkerProperties workerProperties = enabledWorkerProperties(); + workerProperties.setFixedDelayMilliseconds(1L); + EtlJobPollingAdvice advice = new EtlJobPollingAdvice(workerProperties); + + assertRetryAfter(advice, EtlJobStatus.PENDING, "1"); + } + + @Test + void disabledWorkerDoesNotAdvertiseAProcessingCadence() { + EtlJobPollingAdvice advice = new EtlJobPollingAdvice(new EtlJobWorkerProperties()); + HttpHeaders headers = new HttpHeaders(); + headers.set(HttpHeaders.RETRY_AFTER, "999"); + EtlJobStatusResponse responseBody = response(EtlJobStatus.PENDING); + + Object returnedBody = apply(advice, responseBody, headers); + + assertSame(responseBody, returnedBody); + assertTrue(headers.getOrEmpty(HttpHeaders.RETRY_AFTER).isEmpty()); + } + + @Test + void terminalJobsRemoveAnyRetryAfterSuggestion() { + EtlJobPollingAdvice advice = new EtlJobPollingAdvice(enabledWorkerProperties()); + + assertTerminalHeaderRemoved(advice, EtlJobStatus.SUCCEEDED); + assertTerminalHeaderRemoved(advice, EtlJobStatus.FAILED); + } + + @Test + void unrelatedBodiesRemainUnchanged() { + EtlJobPollingAdvice advice = new EtlJobPollingAdvice(enabledWorkerProperties()); + HttpHeaders headers = new HttpHeaders(); + headers.set(HttpHeaders.RETRY_AFTER, "7"); + String body = "unrelated-body"; + + Object returnedBody = apply(advice, body, headers); + + assertSame(body, returnedBody); + assertEquals("7", headers.getFirst(HttpHeaders.RETRY_AFTER)); + } + + @Test + void participatesInTheControllerResponseAdviceChain() { + EtlJobPollingAdvice advice = new EtlJobPollingAdvice(enabledWorkerProperties()); + + assertTrue(advice.supports( + mock(MethodParameter.class), + MappingJackson2HttpMessageConverter.class + )); + } + + @Test + void rejectsMissingWorkerConfiguration() { + assertThrows(NullPointerException.class, () -> new EtlJobPollingAdvice(null)); + } + + private static EtlJobWorkerProperties enabledWorkerProperties() { + EtlJobWorkerProperties workerProperties = new EtlJobWorkerProperties(); + workerProperties.setEnabled(true); + return workerProperties; + } + + private static void assertRetryAfter( + EtlJobPollingAdvice advice, + EtlJobStatus jobStatus, + String expectedSeconds + ) { + HttpHeaders headers = new HttpHeaders(); + EtlJobStatusResponse responseBody = response(jobStatus); + + Object returnedBody = apply(advice, responseBody, headers); + + assertSame(responseBody, returnedBody); + assertEquals(expectedSeconds, headers.getFirst(HttpHeaders.RETRY_AFTER)); + } + + private static void assertTerminalHeaderRemoved( + EtlJobPollingAdvice advice, + EtlJobStatus jobStatus + ) { + HttpHeaders headers = new HttpHeaders(); + headers.set(HttpHeaders.RETRY_AFTER, "999"); + EtlJobStatusResponse responseBody = response(jobStatus); + + Object returnedBody = apply(advice, responseBody, headers); + + assertSame(responseBody, returnedBody); + assertTrue(headers.getOrEmpty(HttpHeaders.RETRY_AFTER).isEmpty()); + } + + private static Object apply( + EtlJobPollingAdvice advice, + Object body, + HttpHeaders headers + ) { + ServerHttpResponse response = mock(ServerHttpResponse.class); + when(response.getHeaders()).thenReturn(headers); + return advice.beforeBodyWrite( + body, + mock(MethodParameter.class), + MediaType.APPLICATION_JSON, + converterType(), + mock(ServerHttpRequest.class), + response + ); + } + + @SuppressWarnings("unchecked") + private static Class> converterType() { + return (Class>) + (Class) MappingJackson2HttpMessageConverter.class; + } + + private static EtlJobStatusResponse response(EtlJobStatus jobStatus) { + return new EtlJobStatusResponse( + JOB_RECORD_ID, + jobStatus, + jobStatus == EtlJobStatus.PENDING ? 0 : 1, + jobStatus == EtlJobStatus.FAILED ? "etl_target_failure" : null, + CREATED_AT, + UPDATED_AT + ); + } +}