fix(api): a failure below the body cache is not a 200

datahub-api/src/main/java/ai/intellistream/datahub/api/filters/CachingBodyFilter.java

@@ -4,19 +4,26 @@
 import jakarta.servlet.*;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
-import lombok.extern.slf4j.Slf4j;
 import org.springframework.web.util.ContentCachingRequestWrapper;
 import org.springframework.web.util.ContentCachingResponseWrapper;
 
 import java.io.IOException;
 
-@Slf4j
 public class CachingBodyFilter implements Filter {
 
     // https://stackoverflow.com/questions/39935190/contentcachingresponsewrapper-produces-empty-response
 
+    /**
+     * Nothing is caught here, deliberately. This filter exists to make bodies loggable, so a
+     * failure below it is never its business to handle: swallowing one would leave the response
+     * at Tomcat's default 200 with an empty body, reporting a request that was dropped on the
+     * floor as a success. DispatcherServlet wraps everything a handler throws — {@code Error}s
+     * included — into {@code ServletException: Handler dispatch failed}, so a catch here would
+     * mask every unhandled failure on every endpoint, not just the odd I/O fault.
+     */
     @Override
-    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) {
+    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
+            throws IOException, ServletException {
         HttpServletRequest httpRequest = (HttpServletRequest) request;
 
         // Streaming endpoints must NOT be wrapped: a file download or graph export response can be
@@ -26,21 +33,20 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha
         // wrappers are present, so passing the raw request/response through simply skips body
         // logging here.
         if (StreamingEndpoints.matches(httpRequest)) {
-            try {
-                chain.doFilter(request, response);
-            } catch (IOException | ServletException e) {
-                log.error("Error in streaming request", e);
-            }
+            chain.doFilter(request, response);
             return;
         }
 
         ContentCachingRequestWrapper reqWrapper = new ContentCachingRequestWrapper(httpRequest, 1024 * 1024 * 20);
         ContentCachingResponseWrapper resWrapper = new ContentCachingResponseWrapper((HttpServletResponse) response);
         try {
             chain.doFilter(reqWrapper, resWrapper);
+        } finally {
+            // In a finally, not after the call: whatever was buffered before a failure has to reach
+            // the real response either way, or an error body written further down is discarded and
+            // the caller gets an empty one. Copying an empty buffer is a no-op, so the container is
+            // still free to write its own error page over an uncommitted response.
             resWrapper.copyBodyToResponse();
-        } catch (IOException | ServletException e) {
-            log.error("Error extracting body", e);
         }
     }
 

datahub-api/src/main/java/ai/intellistream/datahub/config/StrictRequestBodyConfig.java

@@ -5,6 +5,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.converter.ByteArrayHttpMessageConverter;
 import org.springframework.http.converter.HttpMessageConverter;
+import org.springframework.http.converter.HttpMessageConverters;
 import org.springframework.http.converter.StringHttpMessageConverter;
 import org.springframework.http.converter.json.JacksonJsonHttpMessageConverter;
 import org.jspecify.annotations.NonNull;
@@ -51,11 +52,19 @@ public StrictRequestBodyConfig(JsonMapper configuredMapper) {
     }
 
     /**
-     * Extend rather than configure: the default converter list is already built, so replacing the
-     * Jackson one in place leaves ordering and every other converter untouched.
+     * Swaps the converter in the finished list rather than naming a JSON converter on the builder.
+     *
+     * <p>{@code withJsonConverter} would race Spring Boot, whose own configurer sets the JSON converter
+     * on the same builder, and whichever of the two runs last would win. A list configurer runs inside
+     * {@code build()}, after every configurer has had its turn, so it sees the converter Boot settled
+     * on and replaces it in place, leaving ordering and every other converter untouched.
      */
     @Override
-    public void extendMessageConverters(@NonNull List<HttpMessageConverter<?>> converters) {
+    public void configureMessageConverters(HttpMessageConverters.@NonNull ServerBuilder builder) {
+        builder.configureMessageConvertersList(this::replaceJacksonConverter);
+    }
+
+    void replaceJacksonConverter(List<HttpMessageConverter<?>> converters) {
         int replaced = 0;
         for (int i = 0; i < converters.size(); i++) {
             if (converters.get(i) instanceof JacksonJsonHttpMessageConverter) {

datahub-api/src/test/java/ai/intellistream/datahub/api/controllers/DatapointInsertHttpTest.java

@@ -0,0 +1,151 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+package ai.intellistream.datahub.api.controllers;
+
+import ai.intellistream.datahub.api.ApiDatahubApplication;
+import ai.intellistream.datahub.api.datasecurity.DataSecurity;
+import ai.intellistream.datahub.api.init.pulsar.SubscriptionTopicProvisioner;
+import ai.intellistream.datahub.api.services.IngestQuotaService;
+import ai.intellistream.datahub.api.services.TimeseriesService;
+import ai.intellistream.datahub.clickhouse.ClickHouseClientPool;
+import ai.intellistream.datahub.config.InstanceLock;
+import ai.intellistream.datahub.tenant.Tenant;
+import ai.intellistream.datahub.tenant.TenantConfigService;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.security.oauth2.jwt.Jwt;
+import org.springframework.security.oauth2.jwt.JwtDecoder;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.bean.override.mockito.MockitoBean;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * {@code POST /timeseries/data} over real HTTP, in the booted application: the production filter
+ * chain and the message converters Spring Boot actually assembled, which together decide the status
+ * a caller sees. The mocks are the start-up network clients, as in {@code SecurityFilterChainTest}.
+ */
+@SpringBootTest(classes = ApiDatahubApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@ActiveProfiles("ctxtest")
+class DatapointInsertHttpTest {
+
+    /** Any string does: the JwtDecoder is a mock, and this is the value it is told to accept. */
+    private static final String FAKE_BEARER = "dummy-bearer-insert-http";
+    private static final String TENANT = "tenant-insert-http";
+
+    @Value("${local.server.port}")
+    private int port;
+
+    @Autowired
+    private TenantConfigService tenantConfigService;
+
+    @MockitoBean
+    private JwtDecoder jwtDecoder;
+
+    @MockitoBean
+    private PulsarClient pulsarClient;
+
+    @MockitoBean
+    private PulsarAdmin pulsarAdmin;
+
+    @MockitoBean
+    private ClickHouseClientPool clickHouseClientPool;
+
+    @MockitoBean
+    private SubscriptionTopicProvisioner subscriptionTopicProvisioner;
+
+    @MockitoBean
+    private InstanceLock instanceLock;
+
+    @MockitoBean(name = "eventMessageProducer")
+    private Producer<?> eventMessageProducer;
+
+    @MockitoBean(name = "subscriptionNotifyProducer")
+    private Producer<?> subscriptionNotifyProducer;
+
+    @MockitoBean(name = "allDatapointProducer")
+    private Producer<?> allDatapointProducer;
+
+    @MockitoBean(name = "httpMessageProducer")
+    private Producer<?> httpMessageProducer;
+
+    @MockitoBean
+    private DataSecurity dataSecurity;
+
+    @MockitoBean
+    private IngestQuotaService ingestQuota;
+
+    @MockitoBean
+    private TimeseriesService timeseriesService;
+
+    @BeforeEach
+    void setUp() {
+        when(jwtDecoder.decode(FAKE_BEARER)).thenReturn(Jwt.withTokenValue(FAKE_BEARER)
+                .header("alg", "none")
+                .claim("organization", Map.of("org", Map.of("id", TENANT)))
+                .claim("realm_access", Map.of("roles", List.of("DATAHUB_ACCESS")))
+                .subject("insert-http-test")
+                .build());
+        tenantConfigService.cachedTenants.put(TENANT, new Tenant());
+    }
+
+    /**
+     * The controller answers every {@link Exception} itself, so what escapes it is an {@link Error},
+     * and the likeliest one under a large concurrent ingest is running out of heap. The body-cache
+     * filter used to swallow it: nothing had set a status, so the caller was told {@code 200} and
+     * every SDK counted the datapoints as stored.
+     */
+    @Test
+    void anInsertThatRunsOutOfMemoryIsAServerErrorNotA200() throws Exception {
+        when(timeseriesService.insertDatapoints(any())).thenThrow(new OutOfMemoryError("Java heap space"));
+
+        HttpResponse<String> response = post("""
+                {"items":[{"externalId":"sensor_a",
+                           "datapoints":[{"timestamp":1745328000000,"value":"22.4"}]}]}""");
+
+        // A status the SDKs retry. A 200 here tells the caller the datapoints were stored.
+        assertThat(response.statusCode()).as(response.body()).isEqualTo(500);
+    }
+
+    /**
+     * Strictness is attached by {@code StrictRequestBodyConfig} to the list Spring Boot builds, so
+     * only a booted application shows it survived: a unit test of the config cannot tell whether
+     * Boot's converter assembly still calls it.
+     */
+    @Test
+    void anUnknownFieldInTheBodyIsA400NamingIt() throws Exception {
+        HttpResponse<String> response = post("""
+                {"items":[{"externalId":"sensor_a", "tableEngine":"MERGETREE",
+                           "datapoints":[{"timestamp":1745328000000,"value":"22.4"}]}]}""");
+
+        assertThat(response.statusCode()).as(response.body()).isEqualTo(400);
+        assertThat(response.body()).contains("tableEngine");
+        verify(timeseriesService, never()).insertDatapoints(any());
+    }
+
+    private HttpResponse<String> post(String json) throws Exception {
+        return HttpClient.newHttpClient().send(HttpRequest.newBuilder()
+                        .uri(URI.create("http://localhost:" + port + "/timeseries/data"))
+                        .header("Authorization", "Bearer " + FAKE_BEARER)
+                        .header("Content-Type", "application/json")
+                        .POST(HttpRequest.BodyPublishers.ofString(json))
+                        .build(),
+                HttpResponse.BodyHandlers.ofString());
+    }
+}

datahub-api/src/test/java/ai/intellistream/datahub/api/filters/CachingBodyFilterTest.java

@@ -0,0 +1,103 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+package ai.intellistream.datahub.api.filters;
+
+import jakarta.servlet.FilterChain;
+import jakarta.servlet.ServletException;
+import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.mock.web.MockFilterChain;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.mock.web.MockHttpServletResponse;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Body caching must not change what the caller is told happened.
+ *
+ * <p>The filter used to log a failure below it and return, which left the response at Tomcat's
+ * default 200 with an empty body — a request the server dropped, reported to the client as a
+ * success. DispatcherServlet wraps everything a handler throws into {@code ServletException}, so
+ * that applied to every unhandled failure on every non-streaming endpoint.
+ */
+class CachingBodyFilterTest {
+
+    private final CachingBodyFilter filter = new CachingBodyFilter();
+    private final MockHttpServletResponse response = new MockHttpServletResponse();
+
+    private static MockHttpServletRequest post(String uri) {
+        MockHttpServletRequest request = new MockHttpServletRequest("POST", uri);
+        request.setContent("{\"items\":[]}".getBytes(StandardCharsets.UTF_8));
+        request.setContentType(MediaType.APPLICATION_JSON_VALUE);
+        return request;
+    }
+
+    @Test
+    void aFailureBelowTheFilterIsNotReportedAsSuccess() {
+        FilterChain exploding = (req, res) -> {
+            throw new ServletException("Handler dispatch failed", new OutOfMemoryError("Java heap space"));
+        };
+
+        assertThatThrownBy(() -> filter.doFilter(post("/timeseries/data"), response, exploding))
+                .isInstanceOf(ServletException.class)
+                .hasMessage("Handler dispatch failed");
+
+        // Left uncommitted, so the container is still free to turn this into a 500.
+        assertThat(response.isCommitted()).isFalse();
+        assertThat(response.getContentAsByteArray()).isEmpty();
+    }
+
+    @Test
+    void anIoFailureBelowTheFilterAlsoPropagates() {
+        FilterChain exploding = (req, res) -> {
+            throw new IOException("broken pipe");
+        };
+
+        assertThatThrownBy(() -> filter.doFilter(post("/events/create"), response, exploding))
+                .isInstanceOf(IOException.class);
+    }
+
+    @Test
+    void aFailureOnAStreamingEndpointAlsoPropagates() {
+        FilterChain exploding = (req, res) -> {
+            throw new ServletException("Handler dispatch failed");
+        };
+
+        assertThatThrownBy(() -> filter.doFilter(post("/resources/import"), response, exploding))
+                .isInstanceOf(ServletException.class);
+    }
+
+    @Test
+    void aBodyWrittenBeforeAFailureStillReachesTheCaller() throws Exception {
+        FilterChain writesThenFails = (req, res) -> {
+            res.getWriter().write("{\"error\":\"boom\"}");
+            throw new ServletException("Handler dispatch failed");
+        };
+
+        assertThatThrownBy(() -> filter.doFilter(post("/events/create"), response, writesThenFails))
+                .isInstanceOf(ServletException.class);
+
+        assertThat(response.getContentAsString()).isEqualTo("{\"error\":\"boom\"}");
+    }
+
+    @Test
+    void aSuccessfulResponseIsCopiedThroughExactlyOnce() throws Exception {
+        MockFilterChain chain = new MockFilterChain() {
+            @Override
+            public void doFilter(jakarta.servlet.ServletRequest req, jakarta.servlet.ServletResponse res)
+                    throws IOException, ServletException {
+                res.getWriter().write("{\"items\":[]}");
+                super.doFilter(req, res);
+            }
+        };
+
+        filter.doFilter(post("/events/create"), response, chain);
+
+        assertThat(response.getStatus()).isEqualTo(HttpStatus.OK.value());
+        assertThat(response.getContentAsString()).isEqualTo("{\"items\":[]}");
+    }
+}

datahub-api/src/test/java/ai/intellistream/datahub/config/StrictRequestBodyConfigTest.java

@@ -5,6 +5,7 @@
 import org.junit.jupiter.api.Test;
 import org.springframework.http.converter.ByteArrayHttpMessageConverter;
 import org.springframework.http.converter.HttpMessageConverter;
+import org.springframework.http.converter.HttpMessageConverters;
 import org.springframework.http.converter.StringHttpMessageConverter;
 import org.springframework.http.converter.json.JacksonJsonHttpMessageConverter;
 import tools.jackson.core.type.TypeReference;
@@ -55,17 +56,34 @@ void theRequestBodyConverterIsTheStrictOne() {
         JsonMapper shared = JsonMapper.builder().build();
         List<HttpMessageConverter<?>> converters = defaultishConverters();
 
-        new StrictRequestBodyConfig(shared).extendMessageConverters(converters);
+        new StrictRequestBodyConfig(shared).replaceJacksonConverter(converters);
 
         assertThat(converters).anyMatch(StrictJacksonJsonHttpMessageConverter.class::isInstance);
     }
 
+    /**
+     * Through the real builder, with the JSON converter named after this config has run, as Spring
+     * Boot's own configurer may: the swap must still land, since it happens when the list is built.
+     */
+    @Test
+    void theBuiltServerConvertersCarryTheStrictOneWhoeverNamesTheJsonConverterLast() {
+        HttpMessageConverters.ServerBuilder builder = HttpMessageConverters.forServer().registerDefaults();
+
+        new StrictRequestBodyConfig(JsonMapper.builder().build()).configureMessageConverters(builder);
+        builder.withJsonConverter(new JacksonJsonHttpMessageConverter());
+
+        List<HttpMessageConverter<?>> built = new ArrayList<>();
+        builder.build().forEach(built::add);
+        assertThat(built).anyMatch(StrictJacksonJsonHttpMessageConverter.class::isInstance);
+        assertThat(built).noneMatch(c -> c.getClass() == JacksonJsonHttpMessageConverter.class);
+    }
+
     /** Replacing the Jackson converter must not disturb the others or their order. */
     @Test
     void leavesTheOtherConvertersAlone() {
         List<HttpMessageConverter<?>> converters = defaultishConverters();
 
-        new StrictRequestBodyConfig(JsonMapper.builder().build()).extendMessageConverters(converters);
+        new StrictRequestBodyConfig(JsonMapper.builder().build()).replaceJacksonConverter(converters);
 
         assertThat(converters).hasSize(3);
         assertThat(converters.get(0)).isInstanceOf(ByteArrayHttpMessageConverter.class);
@@ -79,7 +97,7 @@ void toleratesAConverterListWithoutJackson() {
         List<HttpMessageConverter<?>> converters = new ArrayList<>(List.of(new StringHttpMessageConverter()));
 
         assertThatCode(() -> new StrictRequestBodyConfig(JsonMapper.builder().build())
-                .extendMessageConverters(converters)).doesNotThrowAnyException();
+                .replaceJacksonConverter(converters)).doesNotThrowAnyException();
         assertThat(converters).hasSize(1);
     }
 }