From 312fa7980e96a27f38cc3e91ac5885aa21631fd2 Mon Sep 17 00:00:00 2001 From: Devesh Bhardwaj Date: Wed, 19 Aug 2026 13:12:52 +0530 Subject: [PATCH] SK-3061: fix bulkInsert(sync) interceptor-exception leak processBulkInsertSync called insertBatchFutures - which synchronously invokes the caller's RequestInterceptor - before entering its own try block, so a throwing interceptor escaped bulkInsert() as a raw, undeclared exception instead of the documented SkyflowException. processBulkDetokenizeSync/processBulkDeleteTokensSync/processBulkTokenizeSync already call their own *BatchFutures inside their own try/catch(Exception), which is why they were not exploitable the same way. This moves insertBatchFutures's call inside processBulkInsertSync's existing try, matching that same structure, instead of adding a new catch-all to the 4 public sync methods (the broader fix reverted here). Added a regression test asserting a throwing interceptor surfaces as SkyflowException from bulkInsert. Confirmed via TDD: failed before this change (raw IllegalStateException), passes after. mvn -pl common,flowvault -am test -> 709 (flowvault) + 106 (common/ValidationsTests) tests, 0 failures, BUILD SUCCESS. Co-Authored-By: Claude Sonnet 5 --- .../vault/controller/VaultController.java | 2 +- .../controller/VaultControllerTests.java | 26 +++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/flowvault/src/main/java/com/skyflow/vault/controller/VaultController.java b/flowvault/src/main/java/com/skyflow/vault/controller/VaultController.java index 0d6d4618..0a0e962f 100644 --- a/flowvault/src/main/java/com/skyflow/vault/controller/VaultController.java +++ b/flowvault/src/main/java/com/skyflow/vault/controller/VaultController.java @@ -769,9 +769,9 @@ private BulkInsertResponse processBulkInsertSync( ) throws ExecutionException, InterruptedException, SkyflowException { LogUtil.printInfoLog(InfoLogs.PROCESSING_BATCHES.getLog()); List records = new ArrayList<>(); - List> futures = this.insertBatchFutures(insertRequest, interceptor, cfg); try { + List> futures = this.insertBatchFutures(insertRequest, interceptor, cfg); CompletableFuture allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); try { allFutures.join(); diff --git a/flowvault/src/test/java/com/skyflow/vault/controller/VaultControllerTests.java b/flowvault/src/test/java/com/skyflow/vault/controller/VaultControllerTests.java index 6b4aa516..af75e540 100644 --- a/flowvault/src/test/java/com/skyflow/vault/controller/VaultControllerTests.java +++ b/flowvault/src/test/java/com/skyflow/vault/controller/VaultControllerTests.java @@ -1164,4 +1164,30 @@ public void testBulkTokenize_interceptorInvokedOncePerBatchWithDistinctContext() Mockito.verify(mockRaw, Mockito.times(EXPECTED_BATCH_COUNT)).tokenize(any(), captor.capture()); assertInterceptorRanOncePerBatch(interceptor, captor.getAllValues()); } + + @Test + public void testBulkInsert_throwingInterceptorWrappedAsSkyflowException() throws Exception { + ApiClient mockApi = Mockito.mock(ApiClient.class); + RawFlowserviceClient mockRaw = mockRawFlowservice(mockApi); + stubInsertEcho(mockRaw); + VaultController controller = createControllerWithMock(mockApi); + + Map data = new HashMap<>(); + data.put("name", "john"); + ArrayList records = new ArrayList<>(); + records.add(BulkInsertRequestRecord.builder().tableName("table1").data(data).build()); + BulkInsertRequest request = BulkInsertRequest.builder().records(records).build(); + + RequestInterceptor interceptor = ctx -> { + throw new IllegalStateException("sync insert interceptor blew up"); + }; + BulkInsertOptions options = BulkInsertOptions.builder().interceptor(interceptor).build(); + + try { + controller.bulkInsert(request, options); + Assert.fail(EXCEPTION_NOT_THROWN); + } catch (SkyflowException e) { + Assert.assertEquals("sync insert interceptor blew up", e.getMessage()); + } + } }