From db593141b9c2e5e1134061c73628161cb205b30a Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Tue, 10 Nov 2026 19:43:05 -0800 Subject: [PATCH 1/2] test(events): build the shared processor at the class's persistence The direct-construction helper always built a persisting store committing off the caller's thread, so the scheduler, close and diagnostic tests ran at deferred persistence in both classes and never without a disk. It now follows persistence(): immediate in DirectEventProcessorTest, disabled in its WithoutPersistence subclass. Co-authored-by: Cursor --- .../launchdarkly/sdk/android/DirectEventProcessorTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java index 5f21ef2b..87e8dce0 100644 --- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java @@ -4794,8 +4794,8 @@ private DirectEventProcessor makeEventProcessor(EventSender diagnosticSender, return makeEventProcessor(diagnosticSender, eventsUri, diagnosticStore, flushIntervalMillis, diagnosticIntervalMillis, closeBudgetMillis, scheduler, diagnosticExecutor, EventStore.create(eventsDirectory.getRoot(), MOBILE_KEY, - DEFAULT_CAPACITY, true, logging.logger), - DEFAULT_CAPACITY); + DEFAULT_CAPACITY, persistence() != EventPersistence.DISABLED, logging.logger), + DEFAULT_CAPACITY, persistence() == EventPersistence.IMMEDIATE); } private DirectEventProcessor makeEventProcessor(EventStore store, int capacity, From 07e50b5456498f29920fd7539a8b95df5be7bfd1 Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Tue, 10 Nov 2026 19:49:24 -0800 Subject: [PATCH 2/2] test(events): cover what the Apple processor tests cover and these did not Ports the Apple examples with no counterpart here: a delivery never closing a batch partway through another commit's run (deferred), going offline and back online delivering nothing on its own, the periodic flush posting nothing with nothing to send and only what is new each time, the batch size recorded for the diagnostic event, and a flush bypassing the backoff posting nothing while offline. Co-authored-by: Cursor --- .../sdk/android/DirectEventProcessorTest.java | 177 ++++++++++++++++++ 1 file changed, 177 insertions(+) diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java index 87e8dce0..071f79ef 100644 --- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java @@ -1227,6 +1227,68 @@ public ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialD } } + @Test + public void neitherGoingOfflineNorComingBackOnlineDeliversOnItsOwn() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + eventProcessor.setOffline(true); + eventProcessor.setOffline(true); + eventProcessor.setOffline(false); + eventProcessor.setOffline(false); + server.getRecorder().requireNoRequests(300, TimeUnit.MILLISECONDS); + + assertTrue(awaitFlush(eventProcessor, 10, TimeUnit.SECONDS)); + assertEquals(1, countEventsOfKind(collectDelivered(server), "custom")); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void thePeriodicFlushPostsNothingWhenThereIsNothingToSend() throws Exception { + try (HttpServer server = startEventsServer()) { + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + server.getUri(), null, 50, 60_000, DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, + scheduler); + try { + eventProcessor.setOffline(false); + + server.getRecorder().requireNoRequests(300, TimeUnit.MILLISECONDS); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + } + + @Test + public void eachPeriodicFlushDeliversWhatWasRecordedSinceTheLastOne() throws Exception { + try (HttpServer server = startEventsServer()) { + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + server.getUri(), null, 50, 60_000, DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, + scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.recordCustomEvent(CONTEXT, "first", LDValue.ofNull(), null); + assertTrue(server.getRecorder().requireRequest(5, TimeUnit.SECONDS).getBody() + .contains("\"key\":\"first\"")); + + eventProcessor.recordCustomEvent(CONTEXT, "second", LDValue.ofNull(), null); + String later = server.getRecorder().requireRequest(5, TimeUnit.SECONDS).getBody(); + assertTrue(later.contains("\"key\":\"second\"")); + assertFalse(later.contains("\"key\":\"first\"")); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + } + @Test public void eventsOnTheirWayIntoTheStoreStillCountAgainstCapacity() throws Exception { // A commit takes the pending events and encodes them before staging, and for that long they @@ -1302,6 +1364,74 @@ boolean stageReserved(byte[] serializedEvent) { } } + @Test + public void aDeliveryNeverClosesABatchPartwayThroughAnotherCommitsRun() throws Exception { + // Events committed together go out together. A batch closed while a commit has staged part of + // its run would leave the rest for a later delivery. + CountDownLatch firstStaged = new CountDownLatch(1); + CountDownLatch releaseCommit = new CountDownLatch(1); + AtomicReference beforeClosing = new AtomicReference<>(); + EventStore store = new EventStore(eventsDirectory.newFolder(), DEFAULT_CAPACITY, true, + logging.logger) { + @Override + boolean stageReserved(byte[] serializedEvent) { + boolean result = super.stageReserved(serializedEvent); + if (new String(serializedEvent, StandardCharsets.UTF_8).contains("\"key\":\"first\"")) { + firstStaged.countDown(); + awaitQuietly(releaseCommit, 5, TimeUnit.SECONDS); + } + return result; + } + + @Override + List closeBatch() { + Runnable before = beforeClosing.getAndSet(null); + if (before != null) { + before.run(); + } + try { + return super.closeBatch(); + } finally { + if (before != null) { + releaseCommit.countDown(); + } + } + } + }; + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + try (HttpServer server = startEventsServer()) { + DirectEventProcessor eventProcessor = makeEventProcessor(server, store, scheduler); + Thread[] competing = new Thread[1]; + beforeClosing.set(() -> { + eventProcessor.recordCustomEvent(CONTEXT, "first", LDValue.ofNull(), null); + eventProcessor.recordCustomEvent(CONTEXT, "second", LDValue.ofNull(), null); + competing[0] = new Thread(eventProcessor::flush); + competing[0].start(); + // Long enough for an unguarded commit to stage "first" and stop short of "second". + awaitQuietly(firstStaged, 500, TimeUnit.MILLISECONDS); + }); + try { + eventProcessor.setOffline(false); + awaitFlush(eventProcessor, 10, TimeUnit.SECONDS); + competing[0].join(5_000); + assertTrue(awaitFlush(eventProcessor, 10, TimeUnit.SECONDS)); + + boolean sawFirst = false; + while (server.getRecorder().count() > 0) { + String body = server.getRecorder().requireRequest(1, TimeUnit.SECONDS).getBody(); + assertEquals("a batch held one of two events committed together: " + body, + body.contains("\"key\":\"first\""), body.contains("\"key\":\"second\"")); + sawFirst |= body.contains("\"key\":\"first\""); + } + assertTrue("the events committed together were never delivered", sawFirst); + } finally { + releaseCommit.countDown(); + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + } + @Test public void storedEventsPastTheMaximumAgeAreDiscardedAsDroppedWithoutFailingAFlush() throws Exception { assumePersisting(); @@ -2085,6 +2215,36 @@ public void flushWithTimeoutReportsDeliveredEvents() throws Exception { } } + @Test + public void aDeliveryRecordsHowManyEventsItsBatchCarriedForTheDiagnosticEvent() throws Exception { + DiagnosticStore diagnosticStore = makeDiagnosticStore(); + try (HttpServer server = startEventsServer()) { + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + server.getUri(), diagnosticStore, NO_PERIODIC_FLUSH_MILLIS, 60_000, + DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.recordIdentifyEvent(CONTEXT); + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + recordEvaluation(eventProcessor, false, null); + assertTrue(awaitFlush(eventProcessor, 10, TimeUnit.SECONDS)); + + // The identify, the custom event and the summary. + assertEquals(3, diagnosticStore.createEventAndReset(0, 0).getJsonValue() + .get("eventsInLastBatch").intValue()); + + assertTrue(awaitFlush(eventProcessor, 10, TimeUnit.SECONDS)); + assertEquals("a flush with nothing to send recorded a batch", 0, + diagnosticStore.createEventAndReset(0, 0).getJsonValue() + .get("eventsInLastBatch").intValue()); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + } + @Test public void aCommitNeverSplitsAnEvaluationAcrossTwoCommits() throws Exception { // The other half of the atomicity invariant. close() only ever commits once, so it can show @@ -3411,6 +3571,23 @@ public void aFlushBypassingTheBackoffAskedForBehindAnOrdinaryOneStillPosts() thr } } + @Test + public void aFlushBypassingTheBackoffPostsNothingWhileOffline() throws Exception { + try (HttpServer server = startEventsServer()) { + DirectEventProcessor eventProcessor = + (DirectEventProcessor) makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + eventProcessor.setOffline(true); + + assertFalse(eventProcessor.flushBypassingBackoff().get(10, TimeUnit.SECONDS)); + server.getRecorder().requireNoRequests(300, TimeUnit.MILLISECONDS); + } finally { + eventProcessor.close(); + } + } + } + @Test public void aBatchIsDroppedOnlyForTwo400sInARow() throws Exception { try (HttpServer server = HttpServer.start(Handlers.sequential(Handlers.status(400),