fix(etl): bound durable job intake before MVC materializes the body

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `POST /api/etl/process` now supports optional authenticated-principal-scoped `Idempotency-Key` retries with atomic target writes, durable response replay, payload-conflict rejection, and explicit replay response metadata.
 - `Idempotency-Key` now prefers the quoted RFC 9651 Structured Field String representation while retaining and normalizing the legacy raw representation to the same durable ledger key.
 - ETL request errors now use RFC 9457 `application/problem+json` responses with a stable `errorCode`, fixed type URI, explicit 400/401/404/409/413/422/503/500 taxonomy, and no internal exception text in client responses.
+- Durable `POST /api/etl/jobs` submissions now share the same transport-level payload byte admission as `POST /api/etl/process`: known oversized `Content-Length` values are rejected without reading the entity, and unknown-length bodies are bounded before MVC converts the request to a String.
 - ETL requests now enforce bounded UTF-8 payload and record-count limits, prevalidate and transform the complete batch before the first JDBC call, and commit accepted records inside one Spring transaction.
 - ETL transformations now preserve comma/colon-bearing values, use locale-independent text conversion and deterministic `BigDecimal` amount formatting, and retry only transient Spring data-access failures.
 - Product branding: user-facing docs and suggested image tags use **mightyETL** (formerly xtrmETL).

docs/etl/bounded-atomic-batches.md

@@ -70,7 +70,7 @@ Values outside the supported range fail configuration binding instead of silentl
 
 ## Operational guidance
 
-- Keep the payload limit aligned with gateway and ingress body-size limits. The service-level check occurs after the MVC stack has materialized the request string and is not a substitute for edge enforcement.
+- Keep the payload limit aligned with gateway and ingress body-size limits. The MVC transport guard rejects known oversized `Content-Length` values without reading the entity and bounds unknown-length reads before String conversion for both `POST /api/etl/process` and `POST /api/etl/jobs`. The service-level UTF-8 check remains defense in depth and is not a substitute for edge enforcement.
 - Keep the record limit below the transaction size that the target database can commit within the request timeout and lock budget.
 - Monitor request latency, transaction duration, rollback rate, database pool wait time, and rejected payload/record-limit errors before raising either limit.
 - Use descriptive string identifiers. Numeric JSON identifier types are rejected to keep identifier contracts explicit and stable across systems.

docs/etl/durable-job-intake.md

@@ -14,7 +14,11 @@ are supplied, `mightyetl.*` wins. Enabling intake accepts the temporary boundary
 payloads remain retained in `PENDING` jobs until the worker and terminal payload-clearing slice is
 implemented. Deployments that cannot accept that retention boundary must leave the setting false.
 
-The existing synchronous `POST /api/etl/process` endpoint remains unchanged.
+The existing synchronous `POST /api/etl/process` endpoint remains unchanged. Enabled job intake
+shares that endpoint's `max-payload-bytes` ceiling at the HTTP transport: known oversized
+`Content-Length` values are rejected without reading the entity, and unknown-length or understated
+bodies are bounded before MVC converts the request to a String. Service-level UTF-8 admission stays
+in place as defense in depth.
 
 ## Submit a job
 

etl-service/src/main/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdvice.java

@@ -16,14 +16,15 @@
 import java.util.Objects;
 
 /**
- * Enforces the synchronous ETL payload byte limit before Spring MVC materializes a request body.
+ * Enforces the ETL payload byte limit before Spring MVC materializes a request body.
  *
- * <p>Known oversized bodies are rejected from their {@code Content-Length} metadata without reading
+ * <p>The guard covers the synchronous process endpoint and the durable job-intake endpoint.
+ * Known oversized bodies are rejected from their {@code Content-Length} metadata without reading
  * the entity. Unknown-length or understated bodies are wrapped in a byte-counting stream that reads
  * at most one byte beyond the configured limit before raising the existing typed payload error. The
  * service-level admission check remains in place as defense in depth.</p>
  */
-@ControllerAdvice(assignableTypes = EtlController.class)
+@ControllerAdvice(assignableTypes = {EtlController.class, EtlJobController.class})
 public final class EtlPayloadAdmissionAdvice extends RequestBodyAdviceAdapter {
 
     private final EtlBatchProperties batchProperties;
@@ -41,12 +42,12 @@ public EtlPayloadAdmissionAdvice(EtlBatchProperties batchProperties) {
     }
 
     /**
-     * Applies admission control to string request bodies handled by {@link EtlController}.
+     * Applies admission control to string request bodies handled by the ETL HTTP adapters.
      *
      * @param methodParameter controller method parameter receiving the request body
      * @param targetType declared request-body target type
      * @param converterType selected HTTP message converter type
-     * @return {@code true} only for the synchronous ETL string body
+     * @return {@code true} only for ETL string request bodies
      */
     @Override
     public boolean supports(

etl-service/src/test/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdviceTest.java

@@ -39,6 +39,11 @@ void supportsOnlyStringRequestParameters() throws Exception {
                 String.class,
                 StringHttpMessageConverter.class
         ));
+        assertTrue(advice.supports(
+                jobRequestBodyParameter(),
+                String.class,
+                StringHttpMessageConverter.class
+        ));
         assertFalse(advice.supports(
                 integerParameter(),
                 Integer.class,
@@ -200,6 +205,16 @@ private static MethodParameter requestBodyParameter() throws NoSuchMethodExcepti
         return new MethodParameter(method, 0);
     }
 
+    private static MethodParameter jobRequestBodyParameter() throws NoSuchMethodException {
+        Method method = EtlJobController.class.getMethod(
+                "submit",
+                String.class,
+                String.class,
+                Principal.class
+        );
+        return new MethodParameter(method, 0);
+    }
+
     private static MethodParameter integerParameter() throws NoSuchMethodException {
         Method method = EtlPayloadAdmissionAdviceTest.class.getDeclaredMethod("integerBody", Integer.class);
         return new MethodParameter(method, 0);

etl-service/src/test/java/com/xtrmetl/etl/documentation/EtlBatchDocsAlignmentTest.java

@@ -36,6 +36,8 @@ void runbookStatesAdmissionRollbackAndIngressBoundaries() throws IOException {
 
         assertTrue(runbook.contains("A rejected request performs no database writes"));
         assertTrue(runbook.contains("rolls back earlier writes"));
+        assertTrue(runbook.contains("without reading the entity"));
+        assertTrue(runbook.contains("POST /api/etl/jobs"));
         assertTrue(runbook.contains("not a substitute for edge enforcement"));
         assertTrue(normalizedRunbook.contains("numeric json identifier types are rejected"));
         assertTrue(runbook.contains("no more than 256 Unicode code points"));

etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobHttpPayloadAdmissionTest.java

@@ -0,0 +1,161 @@
+package com.xtrmetl.etl.job;
+
+import com.xtrmetl.etl.controller.EtlJobController;
+import com.xtrmetl.etl.service.EtlBatchProperties;
+import jakarta.servlet.Filter;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletRequestWrapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.security.test.context.support.WithMockUser;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.util.UUID;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.not;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.csrf;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+/**
+ * Proves durable job-intake request bytes are bounded before MVC invokes the job service.
+ */
+@WebMvcTest(EtlJobController.class)
+@EnableConfigurationProperties(EtlBatchProperties.class)
+@TestPropertySource(properties = "xtrmetl.etl.jobs.intake-enabled=true")
+@Import(EtlJobHttpPayloadAdmissionTest.UnknownLengthRequestConfig.class)
+class EtlJobHttpPayloadAdmissionTest {
+
+    private static final String JOBS_PATH = "/api/etl/jobs";
+    private static final String IDEMPOTENCY_KEY = "\"550e8400-e29b-41d4-a716-446655440000\"";
+    private static final String OVERSIZED_MARKER = "oversized-private-marker";
+    private static final String UNKNOWN_LENGTH_HEADER = "X-Test-Unknown-Content-Length";
+    private static final UUID JOB_RECORD_ID = UUID.fromString("cf4f083f-8c90-4f34-a8b6-b53761de44ef");
+
+    @Autowired
+    private MockMvc mockMvc;
+
+    @MockBean
+    private EtlJobService etlJobService;
+
+    @Test
+    @WithMockUser
+    void rejectsKnownOversizedBodyBeforeControllerInvocation() throws Exception {
+        String request = oversizedJsonRequest();
+        when(etlJobService.submit(request, IDEMPOTENCY_KEY, "user"))
+                .thenReturn(new EtlJobSubmission(JOB_RECORD_ID, EtlJobStatus.PENDING, false));
+
+        mockMvc.perform(post(JOBS_PATH)
+                        .with(csrf())
+                        .header("Idempotency-Key", IDEMPOTENCY_KEY)
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(request))
+                .andExpect(status().isPayloadTooLarge())
+                .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_PROBLEM_JSON))
+                .andExpect(header().string(HttpHeaders.CACHE_CONTROL, "no-store"))
+                .andExpect(jsonPath("$.errorCode").value("etl_payload_too_large"))
+                .andExpect(jsonPath("$.instance").value(JOBS_PATH))
+                .andExpect(content().string(not(containsString(OVERSIZED_MARKER))));
+
+        verifyNoInteractions(etlJobService);
+    }
+
+    @Test
+    @WithMockUser
+    void rejectsUnknownLengthOversizedBodyBeforeControllerInvocation() throws Exception {
+        String request = oversizedJsonRequest();
+        when(etlJobService.submit(request, IDEMPOTENCY_KEY, "user"))
+                .thenReturn(new EtlJobSubmission(JOB_RECORD_ID, EtlJobStatus.PENDING, false));
+
+        mockMvc.perform(post(JOBS_PATH)
+                        .with(csrf())
+                        .header("Idempotency-Key", IDEMPOTENCY_KEY)
+                        .header(UNKNOWN_LENGTH_HEADER, "true")
+                        .header(HttpHeaders.TRANSFER_ENCODING, "chunked")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(request))
+                .andExpect(status().isPayloadTooLarge())
+                .andExpect(jsonPath("$.errorCode").value("etl_payload_too_large"));
+
+        verifyNoInteractions(etlJobService);
+    }
+
+    @Test
+    @WithMockUser
+    void acceptsKnownLengthBodyAtExactByteLimit() throws Exception {
+        String request = sizedJsonRequest(EtlBatchProperties.DEFAULT_MAX_PAYLOAD_BYTES);
+        when(etlJobService.submit(request, IDEMPOTENCY_KEY, "user"))
+                .thenReturn(new EtlJobSubmission(JOB_RECORD_ID, EtlJobStatus.PENDING, false));
+
+        mockMvc.perform(post(JOBS_PATH)
+                        .with(csrf())
+                        .header("Idempotency-Key", IDEMPOTENCY_KEY)
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(request))
+                .andExpect(status().isAccepted())
+                .andExpect(jsonPath("$.jobRecordId").value(JOB_RECORD_ID.toString()));
+
+        verify(etlJobService).submit(request, IDEMPOTENCY_KEY, "user");
+    }
+
+    private static String oversizedJsonRequest() {
+        return "[{\"id\":\"" + OVERSIZED_MARKER + ""
+                + "x".repeat(EtlBatchProperties.DEFAULT_MAX_PAYLOAD_BYTES)
+                + "\"}]";
+    }
+
+    private static String sizedJsonRequest(int totalBytes) {
+        String prefix = "[{\"id\":\"";
+        String suffix = "\"}]";
+        int fillerLength = totalBytes - prefix.length() - suffix.length();
+        return prefix + "x".repeat(fillerLength) + suffix;
+    }
+
+    /**
+     * Test-only transport shim that models chunked input whose byte length is not known up front.
+     */
+    @TestConfiguration
+    static class UnknownLengthRequestConfig {
+
+        @Bean
+        @Order(Ordered.HIGHEST_PRECEDENCE)
+        Filter unknownLengthRequestFilter() {
+            return (request, response, chain) -> {
+                if (request instanceof HttpServletRequest httpRequest
+                        && httpRequest.getHeader(UNKNOWN_LENGTH_HEADER) != null) {
+                    chain.doFilter(new HttpServletRequestWrapper(httpRequest) {
+                        @Override
+                        public int getContentLength() {
+                            return -1;
+                        }
+
+                        @Override
+                        public long getContentLengthLong() {
+                            return -1L;
+                        }
+                    }, response);
+                    return;
+                }
+                chain.doFilter(request, response);
+            };
+        }
+    }
+}

etl-service/src/test/java/com/xtrmetl/etl/job/EtlJobMigrationDocumentationTest.java

@@ -66,6 +66,8 @@ void runbookDocumentsAcceptedSemanticsOwnershipAndTheWorkerBoundary() throws IOE
         assertTrue(runbook.contains("disabled by default"));
         assertTrue(runbook.contains("mightyetl.etl.jobs.intake-enabled=true"));
         assertTrue(runbook.contains("xtrmetl.etl.jobs.intake-enabled=true"));
+        assertTrue(runbook.contains("max-payload-bytes"));
+        assertTrue(runbook.contains("without reading the entity"));
     }
 
     private static String read(String relativePath) throws IOException {