test(qwp): make error dispatcher overflow test deterministic

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderConnectionDispatcherTest.java

@@ -100,10 +100,12 @@ public void testFullInboxDropsOldestAndCounts() throws Exception {
         // entry and admit the new one. Later connection events carry the
         // freshest state, so dropping the head loses the least information.
         CountDownLatch unblock = new CountDownLatch(1);
+        CountDownLatch listenerEntered = new CountDownLatch(1);
         List<SenderConnectionEvent> received = new ArrayList<>();
         Object lock = new Object();
         CountDownLatch allDelivered = new CountDownLatch(5);
         try (SenderConnectionDispatcher d = new SenderConnectionDispatcher(ev -> {
+            listenerEntered.countDown();
             try {
                 unblock.await();
             } catch (InterruptedException ignored) {
@@ -114,11 +116,12 @@ public void testFullInboxDropsOldestAndCounts() throws Exception {
             }
             allDelivered.countDown();
         }, /*capacity=*/ 4)) {
-            // First offer starts the dispatcher and lands in the listener
-            // immediately (and blocks there), freeing one slot. Now we can
-            // fill the bounded inbox to capacity (4), then overflow.
+            // First offer starts the dispatcher. Wait until it lands in the
+            // listener and blocks there before filling the bounded inbox and
+            // overflowing it.
             Assert.assertTrue(d.offer(buildEvent(0)));
-            TimeUnit.MILLISECONDS.sleep(50);
+            Assert.assertTrue("dispatcher should take the head into the listener within 5s",
+                    listenerEntered.await(5, TimeUnit.SECONDS));
             for (int i = 1; i <= 4; i++) {
                 Assert.assertTrue("inbox should accept offer " + i,
                         d.offer(buildEvent(i)));

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SenderErrorDispatcherTest.java

@@ -102,10 +102,12 @@ public void testFullInboxDropsOldestAndCounts() throws Exception {
         // and admit the new one. The latest entry is always the most
         // informative, so the FIFO head loses, not the new arrival.
         CountDownLatch unblock = new CountDownLatch(1);
+        CountDownLatch handlerEntered = new CountDownLatch(1);
         List<SenderError> received = new ArrayList<>();
         Object lock = new Object();
         CountDownLatch allDelivered = new CountDownLatch(5);
         try (SenderErrorDispatcher d = new SenderErrorDispatcher(err -> {
+            handlerEntered.countDown();
             try {
                 unblock.await();
             } catch (InterruptedException ignored) {
@@ -120,9 +122,11 @@ public void testFullInboxDropsOldestAndCounts() throws Exception {
             // immediately (and blocks there). Now we can fill the bounded
             // inbox to capacity, then overflow.
             Assert.assertTrue(d.offer(buildError(0)));
-            // Give the dispatcher a moment to take the head into the
-            // handler so subsequent offers don't get an extra slot.
-            TimeUnit.MILLISECONDS.sleep(50);
+            // The dispatcher polls the head off the inbox before invoking
+            // the handler, so once the handler is entered the head's slot
+            // is free and subsequent offers cannot get an extra slot.
+            Assert.assertTrue("dispatcher should take the head into the handler within 5s",
+                    handlerEntered.await(5, TimeUnit.SECONDS));
             for (int i = 1; i <= 4; i++) {
                 Assert.assertTrue("inbox should accept offer " + i,
                         d.offer(buildError(i)));