From 23054f54f1783e122cc75bca1d5154cc805dcd7f Mon Sep 17 00:00:00 2001
From: James Maes <james@kof22.com>
Date: Mon, 12 Oct 2026 17:48:46 -0500
Subject: [PATCH] fix(sink): stop() waits for in-flight signals before
 returning

The drain thread could take a signal off the queue and still be writing
it when stop() flushed (an empty queue) and returned; the interrupt could
also land mid-write. A resolve submitted just before shutdown could be
lost. That is what failed SinkAndScheduleTest on the release/0.1 build.

stop() now ends the drain loop with a flag, joins the thread (up to 10 s,
interrupting only if it does not finish), then flushes what is left. The
drain poll is 200 ms so stop() returns promptly. New regression test fails
on the old code (3 of 4 runs) and passes on the fix.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../qbits/alerts/engine/QbitAlertsSink.java   | 31 ++++++++++++++++---
 .../alerts/engine/SinkAndScheduleTest.java    | 26 ++++++++++++++++
 2 files changed, 52 insertions(+), 5 deletions(-)

diff --git a/src/main/java/com/kingsrook/qbits/alerts/engine/QbitAlertsSink.java b/src/main/java/com/kingsrook/qbits/alerts/engine/QbitAlertsSink.java
index 594eda9..de08eb5 100644
--- a/src/main/java/com/kingsrook/qbits/alerts/engine/QbitAlertsSink.java
+++ b/src/main/java/com/kingsrook/qbits/alerts/engine/QbitAlertsSink.java
@@ -56,6 +56,8 @@ public class QbitAlertsSink implements AlertSinkInterface
 
    static final int QUEUE_CAPACITY      = 10_000;
    static final int FLOOD_PER_MINUTE    = 1_000;
+   static final long POLL_MILLIS        = 200;
+   static final long STOP_WAIT_MILLIS   = 10_000;
 
    private final AlertsRuntime          runtime;
    private final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);
@@ -65,6 +67,7 @@ public class QbitAlertsSink implements AlertSinkInterface
    private final Map<String, AtomicInteger> perMinute = new ConcurrentHashMap<>();
    private volatile long                minute = -1;
    private volatile Thread              drainThread;
+   private volatile boolean             stopping = false;
 
 
 
@@ -144,16 +147,33 @@ public class QbitAlertsSink implements AlertSinkInterface
 
 
    /*******************************************************************************
-    ** Stop the drain thread (queued signals are processed first).
+    ** Stop the drain thread. Work it already took off the queue finishes first
+    ** (it is not interrupted), then everything still queued is processed here,
+    ** so no signal submitted before stop() is lost.
     *******************************************************************************/
    public void stop()
    {
-      flush();
+      stopping = true;
       Thread thread = drainThread;
       if(thread != null)
       {
-         thread.interrupt();
+         try
+         {
+            thread.join(STOP_WAIT_MILLIS);
+         }
+         catch(InterruptedException e)
+         {
+            Thread.currentThread().interrupt();
+         }
+
+         if(thread.isAlive())
+         {
+            LOG.warn("The alerts drain thread did not stop in time", logPair("waitMillis", STOP_WAIT_MILLIS));
+            thread.interrupt();
+         }
       }
+      flush();
+      drainThread = null;
       started.set(false);
    }
 
@@ -243,13 +263,14 @@ public class QbitAlertsSink implements AlertSinkInterface
    {
       if(started.compareAndSet(false, true))
       {
+         stopping = false;
          drainThread = Thread.ofVirtual().name("alerts-sink").start(() ->
          {
-            while(!Thread.currentThread().isInterrupted())
+            while(!stopping && !Thread.currentThread().isInterrupted())
             {
                try
                {
-                  Runnable work = queue.poll(1, TimeUnit.SECONDS);
+                  Runnable work = queue.poll(POLL_MILLIS, TimeUnit.MILLISECONDS);
                   if(work != null)
                   {
                      work.run();
diff --git a/src/test/java/com/kingsrook/qbits/alerts/engine/SinkAndScheduleTest.java b/src/test/java/com/kingsrook/qbits/alerts/engine/SinkAndScheduleTest.java
index 06208fc..fe4ff63 100644
--- a/src/test/java/com/kingsrook/qbits/alerts/engine/SinkAndScheduleTest.java
+++ b/src/test/java/com/kingsrook/qbits/alerts/engine/SinkAndScheduleTest.java
@@ -52,6 +52,32 @@ class SinkAndScheduleTest extends BaseAlertsTest
 
 
 
+   /*******************************************************************************
+    ** stop() returns only after every signal submitted before it is recorded,
+    ** including one the drain thread already took off the queue (regression:
+    ** stop() used to flush an empty queue and return while the drain thread
+    ** was still writing, so a resolve could be lost).
+    *******************************************************************************/
+   @Test
+   void testStopWaitsForInFlightSignals() throws Exception
+   {
+      AlertsSinkAdapter adapter = new AlertsSinkAdapter();
+      for(int i = 0; i < 25; i++)
+      {
+         String type = "test.stop.thing" + i;
+         adapter.raise(new AlertSignal(type).withTitle("Stop " + i));
+         runtime.sink().stop();
+         adapter.resolve(new AlertResolution(type, null, "done"));
+         runtime.sink().stop();
+      }
+
+      List<QRecord> alerts = all(AlertsAlert.TABLE_NAME);
+      assertThat(alerts).hasSize(25);
+      assertThat(alerts).allSatisfy(alert -> assertThat(alert.getValueString("status")).isEqualTo("RESOLVED"));
+   }
+
+
+
    /*******************************************************************************
     ** Signals queue and are drained (by the thread, or by flush).
     *******************************************************************************/
-- 
2.54.0