fix(etl): reject invalid amounts before persistence

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- ETL amount integrity: invalid `AMOUNT` values fail closed before persistence instead of being rewritten to `0.00`, preserving the distinction from a genuine zero.
 - Durable `POST /api/etl/jobs` submissions now return RFC 9110 `202 Accepted`, a stable pending-job representation, `Location` status-monitor metadata, and explicit replay metadata without changing the synchronous `/api/etl/process` contract. The incomplete intake controller is fail-closed and requires explicit `xtrmetl.etl.jobs.intake-enabled=true` operator opt-in until worker execution and terminal payload clearing are implemented.
 - Concurrent requests using the same authenticated-principal-scoped semantic idempotency key now return immediate RFC 9457 `409 etl_idempotency_request_in_progress` responses through PostgreSQL `pg_try_advisory_xact_lock`; retries after completion still replay the committed response.
 - `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.

docs/etl/bounded-atomic-batches.md

@@ -43,11 +43,13 @@ Fields are transformed directly from the Jackson JSON tree; values are not split
 - Field names use locale-independent uppercase normalization and must remain unique after normalization.
 - `NAME` values use locale-independent uppercase conversion.
 - `EMAIL` values use locale-independent lowercase conversion.
-- `AMOUNT` values use `BigDecimal`, `HALF_UP`, scale `2`, and `toPlainString()`.
-- Invalid, excessive-precision, or extreme-scale amounts retain the legacy fallback `0.00` without expanding attacker-controlled exponents into huge strings.
+- Valid `AMOUNT` values use `BigDecimal`, `HALF_UP`, scale `2`, and `toPlainString()`; a genuine numeric zero therefore remains the legitimate transformed value `0.00`.
+- Malformed, blank, excessive-precision, or extreme-scale `AMOUNT` values are rejected as invalid records before the first JDBC call. Because the complete batch is transformed before persistence, one invalid amount causes the request to fail with no database writes instead of manufacturing a valid-looking zero.
 - Nested arrays and objects are retained as compact JSON instead of collapsing to empty text.
 - Response lines remain `Processed: <id>` in input order. Identifier whitespace, ISO control, Unicode format-control, Unicode line-separator characters, and length are bounded so one record cannot inject, visually reorder, conceal, or amplify response lines.
 
+The fail-closed amount policy prevents new malformed monetary input from becoming indistinguishable from genuine zero after persistence. Historical `processed_data` rows created under the prior fallback cannot be reliably classified after the fact solely from a stored `0.00`; reconciliation therefore requires upstream/source evidence rather than an automated destructive rewrite.
+
 ## Configuration
 
 Preferred keys use the mightyETL product namespace; compatibility keys remain available:
@@ -72,7 +74,8 @@ Values outside the supported range fail configuration binding instead of silentl
 
 - 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 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.
+- Monitor request latency, transaction duration, rollback rate, database pool wait time, and deterministic invalid-record rejection rates before raising either limit.
+- Treat an increased invalid-amount rejection rate as upstream data-quality evidence. Do not log the raw amount or payload merely to diagnose the failure; use source-system reconciliation under purpose-bound access.
 - Use descriptive string identifiers. Numeric JSON identifier types are rejected to keep identifier contracts explicit and stable across systems.
 
 ## Remaining boundary

etl-service/src/main/java/com/xtrmetl/etl/service/EtlService.java

@@ -400,18 +400,19 @@ private String transformValue(String key, @Nullable JsonNode valueNode) {
     }
 
     private String formatAmount(String value) {
+        final BigDecimal amount;
         try {
-            BigDecimal amount = new BigDecimal(value.trim());
-            int scale = amount.scale();
-            if (amount.precision() > MAX_AMOUNT_PRECISION
-                    || scale < -MAX_AMOUNT_ABSOLUTE_SCALE
-                    || scale > MAX_AMOUNT_ABSOLUTE_SCALE) {
-                return "0.00";
-            }
-            return amount.setScale(2, RoundingMode.HALF_UP).toPlainString();
+            amount = new BigDecimal(value.trim());
         } catch (NumberFormatException exception) {
-            return "0.00";
+            throw invalidRecord();
+        }
+        int scale = amount.scale();
+        if (amount.precision() > MAX_AMOUNT_PRECISION
+                || scale < -MAX_AMOUNT_ABSOLUTE_SCALE
+                || scale > MAX_AMOUNT_ABSOLUTE_SCALE) {
+            throw invalidRecord();
         }
+        return amount.setScale(2, RoundingMode.HALF_UP).toPlainString();
     }
 
     private record StoredIdempotencyRecord(String requestDigest, String responseBody) {

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

@@ -45,6 +45,17 @@ void runbookStatesAdmissionRollbackAndIngressBoundaries() throws IOException {
         assertTrue(runbook.contains("Duplicate field names are rejected"));
     }
 
+    @Test
+    void runbookRejectsInvalidAmountsInsteadOfManufacturingZero() throws IOException {
+        String runbook = read("docs/etl/bounded-atomic-batches.md");
+
+        assertTrue(runbook.contains(
+                "Malformed, blank, excessive-precision, or extreme-scale `AMOUNT` values are rejected"
+        ));
+        assertTrue(runbook.contains("before the first JDBC call"));
+        assertFalse(runbook.contains("legacy fallback `0.00`"));
+    }
+
     @Test
     void configurationAndChangelogUseTheSameLimits() throws IOException {
         String application = read("etl-service/src/main/resources/application.yml");
@@ -56,6 +67,15 @@ void configurationAndChangelogUseTheSameLimits() throws IOException {
         assertTrue(changelog.contains("retry only transient Spring data-access failures"));
     }
 
+    @Test
+    void changelogRecordsFailClosedAmountIntegrity() throws IOException {
+        String changelog = read("CHANGELOG.md");
+
+        assertTrue(changelog.contains("invalid `AMOUNT` values fail closed"));
+        assertTrue(changelog.contains("genuine zero"));
+        assertFalse(changelog.contains("invalid `AMOUNT` values to `0.00`"));
+    }
+
     private static String read(String relativePath) throws IOException {
         return Files.readString(projectRoot().resolve(relativePath), StandardCharsets.UTF_8);
     }

etl-service/src/test/java/com/xtrmetl/etl/service/EtlServiceAmountIntegrityTest.java

@@ -0,0 +1,71 @@
+package com.xtrmetl.etl.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+/**
+ * Proves that invalid monetary input is rejected rather than converted into a legitimate zero.
+ */
+class EtlServiceAmountIntegrityTest {
+
+    private JdbcTemplate jdbcTemplate;
+    private EtlService etlService;
+
+    @BeforeEach
+    void setUp() {
+        jdbcTemplate = mock(JdbcTemplate.class);
+        EtlBatchProperties properties = new EtlBatchProperties();
+        properties.setMaxPayloadBytes(65_536);
+        properties.setMaxBatchRecords(100);
+        etlService = new EtlService(jdbcTemplate, new ObjectMapper(), properties);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {
+            "",
+            "not-a-number",
+            "123456789012345678901234567890123456789",
+            "0.0000000000000000001",
+            "1E+19"
+    })
+    void rejectsInvalidOrUnsupportedAmountsBeforeJdbc(String amount) {
+        EtlRequestException exception = assertThrows(
+                EtlRequestException.class,
+                () -> etlService.processData(jsonRecord("record_alpha", amount))
+        );
+
+        assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+        verifyNoInteractions(jdbcTemplate);
+    }
+
+    @Test
+    void invalidAmountMakesTheWholeBatchFailBeforeAnyJdbcWrite() {
+        String payload = """
+                [
+                  {"id":"record_alpha","amount":"10.00"},
+                  {"id":"record_beta","amount":"not-a-number"}
+                ]
+                """;
+
+        EtlRequestException exception = assertThrows(
+                EtlRequestException.class,
+                () -> etlService.processData(payload)
+        );
+
+        assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+        verifyNoInteractions(jdbcTemplate);
+    }
+
+    private static String jsonRecord(String id, String amount) {
+        return "[{\"id\":\"" + id + "\",\"amount\":\"" + amount + "\"}]";
+    }
+}

etl-service/src/test/java/com/xtrmetl/etl/service/EtlServiceBatchSafetyTest.java

@@ -274,15 +274,16 @@ void usesLocaleIndependentTextAndDecimalTransformations() {
     }
 
     @Test
-    void boundsExtremeDecimalInputsWithoutHugePlainStringExpansion() {
+    void rejectsExtremeDecimalInputsWithoutHugePlainStringExpansion() {
         EtlService service = service();
 
-        service.processData("[{\"id\":\"record_alpha\",\"amount\":\"1E+1000000\"}]");
-
-        verify(jdbcTemplate).update(
-                "INSERT INTO processed_data (data) VALUES (?)",
-                "ID:record_alpha,AMOUNT:0.00,"
+        EtlRequestException exception = assertThrows(
+                EtlRequestException.class,
+                () -> service.processData("[{\"id\":\"record_alpha\",\"amount\":\"1E+1000000\"}]")
         );
+
+        assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+        verifyNoInteractions(jdbcTemplate);
     }
 
     @Test

etl-service/src/test/java/com/xtrmetl/etl/service/EtlServiceTest.java

@@ -129,12 +129,16 @@ void formatsAmountsDeterministically(String input, String expected) {
         }
 
         @Test
-        void fallsBackToZeroForInvalidAmount() {
-            etlService.processData(
-                    "[{\"id\":\"record_alpha\",\"amount\":\"not-a-number\"}]"
+        void rejectsInvalidAmountInsteadOfManufacturingZero() {
+            EtlRequestException exception = assertThrows(
+                    EtlRequestException.class,
+                    () -> etlService.processData(
+                            "[{\"id\":\"record_alpha\",\"amount\":\"not-a-number\"}]"
+                    )
             );
 
-            verify(jdbcTemplate).update(anyString(), contains("AMOUNT:0.00"));
+            assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+            verifyNoInteractions(jdbcTemplate);
         }
 
         @Test
@@ -154,19 +158,18 @@ void retainsUnknownAndUnicodeFields() {
         }
 
         @Test
-        void preservesEmptyOptionalValues() {
+        void preservesEmptyOptionalNameAndEmailValues() {
             etlService.processData("""
                     [{
                       "id":"record_alpha",
                       "name":"",
-                      "email":"",
-                      "amount":""
+                      "email":""
                     }]
                     """);
 
             verify(jdbcTemplate).update(
                     anyString(),
-                    eq("ID:record_alpha,NAME:,EMAIL:,AMOUNT:0.00,")
+                    eq("ID:record_alpha,NAME:,EMAIL:,")
             );
         }
     }