Chore/queue disconnect cleanup debug

src/main/java/io/mapsmessaging/engine/destination/DestinationImpl.java

@@ -665,6 +665,7 @@ public void removeSchemaSubscription(@NonNull @NotNull Subscription subscription
    * @param messageId that the delivery is complete
    */
   public void complete(long messageId) {
+    QueueCleanupTrace.record(fullyQualifiedNamespace, "COMPLETE", messageId, "retained", retainManager.current());
     if (retainManager.current() != messageId) {
       completionQueue.add(messageId);
       requestReaper();
@@ -874,7 +875,9 @@ public int handleTask(@NonNull @NotNull Callable<Response> task) throws IOExcept
    */
   public void removeMessage(long messageId) throws IOException {
     long nano = System.nanoTime();
+    QueueCleanupTrace.record(fullyQualifiedNamespace, "REMOVE-BEGIN", messageId);
     resource.remove(messageId);
+    QueueCleanupTrace.record(fullyQualifiedNamespace, "REMOVE-END", messageId);
     if (messageId == retainManager.current()) {
       stats.retainedMessages(-1);
       retainManager.replace(-1);
@@ -893,6 +896,7 @@ public void removeMessage(long messageId) throws IOException {
   public void addMessage(Message message) throws IOException {
     long nano = System.nanoTime();
     resource.add(message);
+    QueueCleanupTrace.record(fullyQualifiedNamespace, "STORED", message.getIdentifier());
     if (message.isRetain()) {
       if (message.getOpaqueData() == null || message.getOpaqueData().length == 0) {
         retainManager.replace(-1);
@@ -917,6 +921,17 @@ public DestinationSubscription getSubscription(String subscriptionName) {
 
   //</editor-fold>
 
+  /** Diagnostic snapshot only; subscription/resource observations are not an atomic cross-thread view. */
+  public void traceCleanupState(String stage) throws IOException {
+    if (QueueCleanupTrace.isEnabled(fullyQualifiedNamespace)) {
+      QueueCleanupTrace.record(fullyQualifiedNamespace, stage, -1, "storedIds", resource.getKeys(), "interest", subscriptionManager.getAll());
+      synchronized (reaperLock) {
+        QueueCleanupTrace.record(fullyQualifiedNamespace, "REAPER-STATE", -1, "queued", reaperQueued, "running", reaperRunning,
+            "pending", reaperPending, "completionEmpty", completionQueue.isEmpty(), "retry", reaperRetries.cleanupTraceState());
+      }
+    }
+  }
+
   /** Guard already queued removals against failures discovered after their interest snapshot. */
   public boolean prepareCompletedRemoval(Queue<Long> messageIds) {
     synchronized (reaperLock) {
@@ -943,6 +958,9 @@ public void deferCompletedMessages(Queue<Long> messageIds, boolean batchFailure)
       if (closed) {
         return;
       }
+      if (QueueCleanupTrace.isEnabled(fullyQualifiedNamespace)) {
+        QueueCleanupTrace.record(fullyQualifiedNamespace, "DEFER", -1, "batchFailure", batchFailure, messageIds.toString());
+      }
       reaperRetries.defer(messageIds, batchFailure);
     }
   }
@@ -1001,7 +1019,14 @@ public Response taskCall() {
           reaperRetries.drainReadyInto(completed);
         }
         if (!completed.isEmpty()) {
-          completed.removeAll(subscriptionManager.getAll());
+          Queue<Long> interested = subscriptionManager.getAll();
+          if (QueueCleanupTrace.isEnabled(fullyQualifiedNamespace)) {
+            QueueCleanupTrace.record(fullyQualifiedNamespace, "REAPER-INTEREST", -1, "candidates", completed.toString(), "interested", interested.toString());
+          }
+          completed.removeAll(interested);
+          if (QueueCleanupTrace.isEnabled(fullyQualifiedNamespace)) {
+            QueueCleanupTrace.record(fullyQualifiedNamespace, "REAPER-REMOVE", -1, completed.toString());
+          }
           if (!completed.isEmpty()) {
             // Same serial task scheduler, so removal cannot overtake reaper processing.
             subscriptionTaskQueue.submit(new BulkRemoveMessageTask(DestinationImpl.this, completed));

src/main/java/io/mapsmessaging/engine/destination/QueueCleanupTrace.java

@@ -0,0 +1,106 @@
+/*
+ *
+ *  Copyright [ 2020 - 2024 ] Matthew Buckton
+ *  Copyright [ 2024 - 2026 ] MapsMessaging B.V.
+ *
+ *  Licensed under the Apache License, Version 2.0 with the Commons Clause
+ *  (the "License"); you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at:
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *      https://commonsclause.com/
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package io.mapsmessaging.engine.destination;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReferenceArray;
+
+/** Temporary, bounded diagnostic recorder. Only the selected test destination is recorded. */
+public final class QueueCleanupTrace implements AutoCloseable {
+
+  private static final int CAPACITY = 32768;
+  private static volatile QueueCleanupTrace active;
+
+  private final String destination;
+  private final long started = System.nanoTime();
+  private final Queue<Runnable> finalSnapshots = new ConcurrentLinkedQueue<>();
+  private final AtomicLong sequence = new AtomicLong();
+  private final AtomicReferenceArray<String> entries = new AtomicReferenceArray<>(CAPACITY);
+
+  private QueueCleanupTrace(String destination) {
+    this.destination = destination;
+  }
+
+  public static QueueCleanupTrace start(String destination) {
+    QueueCleanupTrace trace = new QueueCleanupTrace(destination);
+    active = trace;
+    return trace;
+  }
+
+  public static boolean isEnabled(String destination) {
+    QueueCleanupTrace trace = active;
+    return trace != null && trace.destination.equals(destination);
+  }
+
+  public static void record(String destination, String action, long messageId, Object... details) {
+    QueueCleanupTrace trace = active;
+    if (trace == null || !trace.destination.equals(destination)) {
+      return;
+    }
+    long index = trace.sequence.getAndIncrement();
+    String entry = index + " +" + (System.nanoTime() - trace.started) + "ns [" + Thread.currentThread().getName()
+        + "] " + action + " id=" + messageId + " " + Arrays.toString(details);
+    trace.entries.set((int) (index % CAPACITY), entry);
+  }
+
+  public static void captureOnFinish(String destination, Runnable snapshot) {
+    QueueCleanupTrace trace = active;
+    if (trace != null && trace.destination.equals(destination)) {
+      trace.finalSnapshots.add(snapshot);
+    }
+  }
+
+  public void captureFinalState() {
+    for (Runnable snapshot : finalSnapshots) {
+      try {
+        snapshot.run();
+      } catch (RuntimeException diagnosticFailure) {
+        record(destination, "SNAPSHOT-FAILED", -1, diagnosticFailure.toString());
+      }
+    }
+  }
+
+  public void writeTo(Path path) throws IOException {
+    Files.createDirectories(path.getParent());
+    long end = sequence.get();
+    StringBuilder output = new StringBuilder("destination=" + destination + " events=" + end + " capacity=" + CAPACITY + "\n");
+    for (long index = Math.max(0, end - CAPACITY); index < end; index++) {
+      String entry = entries.get((int) (index % CAPACITY));
+      if (entry != null) {
+        output.append(entry).append('\n');
+      }
+    }
+    Files.writeString(path, output, StandardCharsets.UTF_8);
+  }
+
+  @Override
+  public void close() {
+    if (active == this) {
+      active = null;
+    }
+  }
+}

src/main/java/io/mapsmessaging/engine/destination/ReaperRetryQueue.java

@@ -100,6 +100,10 @@ void drainReadyInto(Queue<Long> completed) {
     batchFailure = false;
   }
 
+  String cleanupTraceState() {
+    return "held=" + held + " pendingDeadlines=" + pending + " batchFailure=" + batchFailure + " retryAfter=" + retryAfter + " now=" + nanoClock.getAsLong();
+  }
+
   void clear() {
     pending.clear();
     held.clear();

src/main/java/io/mapsmessaging/engine/destination/subscription/impl/DestinationSubscription.java

@@ -19,6 +19,7 @@
 
 package io.mapsmessaging.engine.destination.subscription.impl;
 
+import io.mapsmessaging.engine.destination.QueueCleanupTrace;
 import io.mapsmessaging.engine.session.DeliveryStateCoordinator;
 import io.mapsmessaging.admin.SubscriptionJMX;
 import io.mapsmessaging.api.message.Message;
@@ -49,6 +50,7 @@
 
 import java.io.IOException;
 import java.nio.channels.CancelledKeyException;
+import java.util.ArrayList;
 import java.util.Map;
 import java.util.Queue;
 import java.util.concurrent.Future;
@@ -157,6 +159,7 @@ public void expired(long id) {
 
   @Override
   public void close() {
+    traceCleanupState("SUBSCRIPTION-CLOSE");
     releaseProtocolSendSlot();
     if (mbean != null) {
       mbean.close();
@@ -409,7 +412,9 @@ protected Message peekNextMessage() throws IOException {
   }
 
   protected void allocateMessage(Message message) {
+    QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "ALLOCATE-BEGIN", message.getIdentifier(), sessionId);
     messageStateManager.allocate(message);
+    QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "ALLOCATE-END", message.getIdentifier(), sessionId);
     messagesSent++;
   }
 
@@ -506,6 +511,7 @@ public int register(long messageId) {
   public void handleTransaction(boolean isAck, long messageId) {
     ThreadLocalContext.checkDomain(DestinationImpl.SUBSCRIPTION_TASK_KEY);
 
+    QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "TRANSACTION-BEGIN", messageId, "ack", isAck, sessionId);
     if (isAck) {
       logger.log(ServerLogMessages.DESTINATION_SUBSCRIPTION_ACK, messageId, destinationImpl.getFullyQualifiedNamespace(), sessionId);
       clearDeliveryMapping(messageId);
@@ -519,6 +525,7 @@ public void handleTransaction(boolean isAck, long messageId) {
       messageStateManager.rollback(messageId);
       messagesRolledBack++;
     }
+    QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "TRANSACTION-END", messageId, "ack", isAck, sessionId);
     schedule();
   }
 
@@ -536,6 +543,7 @@ public void acknowledgePreviousEvent() {
   }
 
   public void completeMessage(long messageId) {
+    QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "AUTO-COMMIT", messageId, sessionId);
     clearDeliveryMapping(messageId);
     messageStateManager.commit(messageId);
     messagesAcked++;
@@ -624,6 +632,20 @@ public void run() {
     }
   }
 
+  /** Best-effort diagnostic view; does not alter subscription scheduling or message ownership. */
+  public void traceCleanupState(String stage) {
+    if (QueueCleanupTrace.isEnabled(destinationImpl.getFullyQualifiedNamespace())) {
+      try {
+        QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), stage, -1, "subscription", sessionId,
+            "atRest", new ArrayList<>(messageStateManager.getAllAtRest()), "all", new ArrayList<>(messageStateManager.getAll()),
+            "hibernating", hibernating, "paused", isPaused,
+            "sent", messagesSent, "acked", messagesAcked, "rolledBack", messagesRolledBack);
+      } catch (RuntimeException diagnosticFailure) {
+        QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "SNAPSHOT-FAILED", -1, stage, diagnosticFailure.toString());
+      }
+    }
+  }
+
   public boolean schedule() {
     if (!flowControlBlocked && isReady()) {
       destinationImpl.scanForDelivery(this);

src/main/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SessionSharedSubscription.java

@@ -19,6 +19,7 @@
 
 package io.mapsmessaging.engine.destination.subscription.impl.shared;
 
+import io.mapsmessaging.engine.destination.QueueCleanupTrace;
 import io.mapsmessaging.engine.session.DeliveryStateCoordinator;
 import io.mapsmessaging.api.message.Message;
 import io.mapsmessaging.api.SubscribedEventManager;
@@ -70,13 +71,16 @@ public SessionSharedSubscription(
 
   @Override
   public void cancel() throws IOException {
+    traceCleanupState("CANCEL-BEGIN");
     clearDeliveryMappings();
     releaseProtocolSendSlot();
     List<OutstandingEventDetails> outstandingEvents = new ArrayList<>(acknowledgementController.getOutstanding());
     for (OutstandingEventDetails outstandingEvent : outstandingEvents) {
+      QueueCleanupTrace.record(getName(), "CANCEL-ROLLBACK", outstandingEvent.getId(), sessionId);
       sharedSubscription.handleTransaction(false, outstandingEvent.getId());
     }
     sharedSubscription.removeSession(sessionImpl);
+    traceCleanupState("CANCEL-END");
   }
 
   @Override
@@ -91,10 +95,13 @@ public void delete() throws IOException {
 
   @Override
   public void hibernate() {
+    traceCleanupState("HIBERNATE-BEGIN");
     releaseProtocolSendSlot();
     for (OutstandingEventDetails outstanding : acknowledgementController.getOutstanding()) {
+      QueueCleanupTrace.record(getName(), "HIBERNATE-ROLLBACK", outstanding.getId(), sessionId);
       sharedSubscription.getAcknowledgementController().rollback(outstanding.getId());
     }
+    traceCleanupState("HIBERNATE-END");
   }
 
   @Override
@@ -301,6 +308,11 @@ public int getDepth() {
 
   @Override
   public void sendMessage(Message message) {
+    if (QueueCleanupTrace.isEnabled(getName())) {
+      QueueCleanupTrace.record(getName(), "SEND", message.getIdentifier(), "session", sessionId,
+          "closed", sessionImpl.isClosed(), "hibernating", hibernating, "controller", System.identityHashCode(acknowledgementController),
+          "ackType", acknowledgementController.getType());
+    }
     String name = "";
     String endpoint = "";
     String version = "";
@@ -323,6 +335,22 @@ public void sendMessage(Message message) {
     ThreadContext.clearMap();
   }
 
+  /** Keep removed consumers inspectable by the diagnostic test after their expiry callback. */
+  public void traceCleanupState(String stage) {
+    if (QueueCleanupTrace.isEnabled(getName())) {
+      try {
+        SessionImpl session = sessionImpl;
+        QueueCleanupTrace.record(getName(), stage, -1, "session", sessionId, "attached", session != null,
+            "closed", session != null && session.isClosed(), "hibernating", hibernating, "blocked", flowControlBlocked,
+            "controller", System.identityHashCode(acknowledgementController), "ackType", acknowledgementController.getType(),
+            "credit", acknowledgementController.getMaxOutstanding(), "outstanding", acknowledgementController.getOutstanding());
+        sharedSubscription.traceCleanupState(stage + "-SHARED");
+      } catch (RuntimeException diagnosticFailure) {
+        QueueCleanupTrace.record(getName(), "SNAPSHOT-FAILED", -1, stage, diagnosticFailure.toString());
+      }
+    }
+  }
+
   @Override
   public Future<Response> getNext() throws IOException {
     return sharedSubscription.getNext();