diff --git a/src/main/java/io/mapsmessaging/engine/destination/DestinationImpl.java b/src/main/java/io/mapsmessaging/engine/destination/DestinationImpl.java index 8785122241..84a1fdc0a2 100644 --- a/src/main/java/io/mapsmessaging/engine/destination/DestinationImpl.java +++ b/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 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) { // + /** 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 messageIds) { synchronized (reaperLock) { @@ -943,6 +958,9 @@ public void deferCompletedMessages(Queue 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 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)); diff --git a/src/main/java/io/mapsmessaging/engine/destination/QueueCleanupTrace.java b/src/main/java/io/mapsmessaging/engine/destination/QueueCleanupTrace.java new file mode 100644 index 0000000000..4f81059d0a --- /dev/null +++ b/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 finalSnapshots = new ConcurrentLinkedQueue<>(); + private final AtomicLong sequence = new AtomicLong(); + private final AtomicReferenceArray 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; + } + } +} diff --git a/src/main/java/io/mapsmessaging/engine/destination/ReaperRetryQueue.java b/src/main/java/io/mapsmessaging/engine/destination/ReaperRetryQueue.java index 4a55120b97..d050edd434 100644 --- a/src/main/java/io/mapsmessaging/engine/destination/ReaperRetryQueue.java +++ b/src/main/java/io/mapsmessaging/engine/destination/ReaperRetryQueue.java @@ -100,6 +100,10 @@ void drainReadyInto(Queue completed) { batchFailure = false; } + String cleanupTraceState() { + return "held=" + held + " pendingDeadlines=" + pending + " batchFailure=" + batchFailure + " retryAfter=" + retryAfter + " now=" + nanoClock.getAsLong(); + } + void clear() { pending.clear(); held.clear(); diff --git a/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/DestinationSubscription.java b/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/DestinationSubscription.java index 5baeb8ad50..fddb57ec4c 100644 --- a/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/DestinationSubscription.java +++ b/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); diff --git a/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SessionSharedSubscription.java b/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SessionSharedSubscription.java index abdf5a737f..756dd9c579 100644 --- a/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SessionSharedSubscription.java +++ b/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 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 getNext() throws IOException { return sharedSubscription.getNext(); diff --git a/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SharedSubscription.java b/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SharedSubscription.java index c1fe4839ce..ae3ef4eb45 100644 --- a/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SharedSubscription.java +++ b/src/main/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SharedSubscription.java @@ -19,6 +19,7 @@ package io.mapsmessaging.engine.destination.subscription.impl.shared; +import io.mapsmessaging.engine.destination.QueueCleanupTrace; import io.mapsmessaging.api.message.Message; import io.mapsmessaging.engine.destination.DestinationImpl; import io.mapsmessaging.engine.destination.subscription.Subscription; @@ -33,6 +34,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; public class SharedSubscription extends DestinationSubscription { @@ -69,6 +71,10 @@ public void close() { public Subscription addSession(SessionImpl sessionImpl, String sessionId, SubscriptionContext context, AcknowledgementController acknowledgementController) { SessionSharedSubscription subscription = new SessionSharedSubscription(this, sessionImpl, sessionId, context, acknowledgementController); subscriptions.add(subscription); + subscription.traceCleanupState("MEMBER-ADD"); + if (QueueCleanupTrace.isEnabled(destinationImpl.getFullyQualifiedNamespace())) { + QueueCleanupTrace.captureOnFinish(destinationImpl.getFullyQualifiedNamespace(), () -> subscription.traceCleanupState("TEST-FINAL")); + } if (hibernating && sessionImpl != null) { wakeUp(sessionImpl); @@ -80,7 +86,10 @@ public Subscription addSession(SessionImpl sessionImpl, String sessionId, Subscr } public void removeSession(SessionImpl remove) { + QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "MEMBER-REMOVE-BEGIN", -1, + remove == null ? "null" : remove.getName(), "members", subscriptions.size()); subscriptions.remove(remove); + QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "MEMBER-REMOVE-END", -1, "members", subscriptions.size()); if (subscriptions.isEmpty()) { destinationImpl.removeSubscription(getSessionId()); close(); @@ -96,6 +105,7 @@ protected Message retrieveNextMessageWithFlowControl() throws IOException { reservedSubscription = subscriptions.reserveNext(message); if (reservedSubscription == null) { + QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "NO-ELIGIBLE-CONSUMER", message.getIdentifier(), "members", subscriptions.size()); return null; } @@ -130,6 +140,7 @@ public void sendMessage(Message message) { if (subscription != null) { subscription.releaseProtocolSendSlot(); } + QueueCleanupTrace.record(destinationImpl.getFullyQualifiedNamespace(), "SEND-ROLLBACK", message.getIdentifier()); rollbackReceived(message.getIdentifier()); // Unable to send for any number of reasons } @@ -154,7 +165,7 @@ private static class SessionSubscriptionMap { public SessionSubscriptionMap() { lookupMap = new LinkedHashMap<>(); - flatMap = new ArrayList<>(); + flatMap = new CopyOnWriteArrayList<>(); idx = -1; } diff --git a/src/main/java/io/mapsmessaging/engine/destination/subscription/tasks/MessageDeliveredListener.java b/src/main/java/io/mapsmessaging/engine/destination/subscription/tasks/MessageDeliveredListener.java index 5587c50c16..0f3fb4f95d 100644 --- a/src/main/java/io/mapsmessaging/engine/destination/subscription/tasks/MessageDeliveredListener.java +++ b/src/main/java/io/mapsmessaging/engine/destination/subscription/tasks/MessageDeliveredListener.java @@ -19,6 +19,7 @@ package io.mapsmessaging.engine.destination.subscription.tasks; +import io.mapsmessaging.engine.destination.QueueCleanupTrace; import io.mapsmessaging.engine.destination.subscription.impl.DestinationSubscription; import io.mapsmessaging.engine.destination.subscription.transaction.AcknowledgementController; import io.mapsmessaging.engine.tasks.EngineTask; @@ -39,6 +40,8 @@ public MessageDeliveredListener(DestinationSubscription subscription, Acknowledg @Override public Response taskCall() { long messageId = acknowledgementController.messageSent(); + QueueCleanupTrace.record(subscription.getDestinationImpl().getFullyQualifiedNamespace(), "DELIVERED-CALLBACK", messageId, + "controller", System.identityHashCode(acknowledgementController), "ackType", acknowledgementController.getType()); if (messageId > -1) { subscription.completeMessage(messageId); } diff --git a/src/main/java/io/mapsmessaging/engine/destination/tasks/BulkRemoveMessageTask.java b/src/main/java/io/mapsmessaging/engine/destination/tasks/BulkRemoveMessageTask.java index d7fb6ae37e..9e5fe86b6c 100644 --- a/src/main/java/io/mapsmessaging/engine/destination/tasks/BulkRemoveMessageTask.java +++ b/src/main/java/io/mapsmessaging/engine/destination/tasks/BulkRemoveMessageTask.java @@ -20,6 +20,7 @@ package io.mapsmessaging.engine.destination.tasks; import io.mapsmessaging.engine.destination.DestinationImpl; +import io.mapsmessaging.engine.destination.QueueCleanupTrace; import io.mapsmessaging.engine.tasks.EngineTask; import io.mapsmessaging.engine.tasks.FutureResponse; import io.mapsmessaging.engine.tasks.LongResponse; @@ -88,6 +89,7 @@ private Response removeBatch() throws Exception { destination.removeMessage(messageId); removed++; } catch (IOException | RuntimeException failure) { + QueueCleanupTrace.record(destination.getFullyQualifiedNamespace(), "REMOVE-FAILED", messageId, failure.toString()); failedMessages.add(messageId); destination.holdCompletedMessage(messageId); logger.log(ServerLogMessages.DESTINATION_REAPER_REMOVE_FAILED, failure, destination.getFullyQualifiedNamespace(), messageId); diff --git a/src/test/java/io/mapsmessaging/api/subscriptions/UnsubscribeValidationTest.java b/src/test/java/io/mapsmessaging/api/subscriptions/UnsubscribeValidationTest.java index dc4436df94..d7ad60be96 100644 --- a/src/test/java/io/mapsmessaging/api/subscriptions/UnsubscribeValidationTest.java +++ b/src/test/java/io/mapsmessaging/api/subscriptions/UnsubscribeValidationTest.java @@ -19,23 +19,34 @@ package io.mapsmessaging.api.subscriptions; -import io.mapsmessaging.api.*; +import io.mapsmessaging.api.Destination; +import io.mapsmessaging.api.MessageAPITest; +import io.mapsmessaging.api.MessageBuilder; +import io.mapsmessaging.api.MessageEvent; +import io.mapsmessaging.api.MessageListener; +import io.mapsmessaging.api.Session; +import io.mapsmessaging.api.SessionContextBuilder; +import io.mapsmessaging.api.SubscribedEventManager; +import io.mapsmessaging.api.SubscriptionContextBuilder; import io.mapsmessaging.api.features.ClientAcknowledgement; import io.mapsmessaging.api.features.DestinationType; import io.mapsmessaging.api.features.QualityOfService; +import io.mapsmessaging.engine.destination.QueueCleanupTrace; import io.mapsmessaging.engine.destination.subscription.SubscriptionContext; import io.mapsmessaging.engine.session.FakeProtocol; import io.mapsmessaging.network.ProtocolClientConnection; import io.mapsmessaging.test.WaitForState; -import java.io.IOException; -import java.util.concurrent.TimeUnit; -import javax.security.auth.login.LoginException; -import lombok.NonNull; -import lombok.SneakyThrows; import org.jetbrains.annotations.NotNull; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInfo; +import lombok.NonNull; +import lombok.SneakyThrows; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.concurrent.TimeUnit; +import javax.security.auth.login.LoginException; class UnsubscribeValidationTest extends MessageAPITest implements MessageListener { @@ -43,7 +54,6 @@ class UnsubscribeValidationTest extends MessageAPITest implements MessageListene void topicUnsubscribeEventCleanUp(TestInfo testInfo) throws LoginException, IOException { eventCleanUpTest(testInfo.getTestMethod().get().getName(), "topic/topic1", false, true, false); - } @Test @@ -53,15 +63,36 @@ void queueUnsubscribeEventCleanUp(TestInfo testInfo) throws LoginException, IOEx void topicDisconnectEventCleanUp(TestInfo testInfo) throws LoginException, IOException { eventCleanUpTest(testInfo.getTestMethod().get().getName(), "topic/topic1", false, false, false); - } @Test void queueDisconnectEventCleanUp(TestInfo testInfo) throws LoginException, IOException { - eventCleanUpTest(testInfo.getTestMethod().get().getName(), "queue/queue1", true, false, false); + String name = testInfo.getTestMethod().get().getName(); + try (QueueCleanupTrace trace = QueueCleanupTrace.start("queue/queue1")) { + try { + eventCleanUpTest(name, "queue/queue1", true, false, false); + } finally { + // Preserve the original assertion/exception even if a concurrent snapshot or file write fails. + try { + trace.captureFinalState(); + md.getDestinationManager().find("queue/queue1").get(2, TimeUnit.SECONDS).traceCleanupState("TEST-FINAL-DESTINATION"); + } catch (Exception diagnosticFailure) { + if (diagnosticFailure instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + QueueCleanupTrace.record("queue/queue1", "SNAPSHOT-FAILED", -1, diagnosticFailure.toString()); + } + Path tracePath = Path.of("target", "queue-cleanup-debug", name + "-" + System.currentTimeMillis() + ".log"); + try { + trace.writeTo(tracePath); + System.err.println("Queue cleanup trace: " + tracePath.toAbsolutePath()); + } catch (IOException diagnosticFailure) { + diagnosticFailure.printStackTrace(System.err); + } + } + } } - void topicResumeEventCleanUp(TestInfo testInfo) throws LoginException, IOException { eventCleanUpTest(testInfo.getTestMethod().get().getName(), "topic/topic1", false, false, true); } @@ -71,17 +102,16 @@ void queueResumeEventCleanUp(TestInfo testInfo) throws LoginException, IOExcepti eventCleanUpTest(testInfo.getTestMethod().get().getName(), "queue/queue1", true, false, true); } - @SneakyThrows public void eventCleanUpTest(String name, String destinationName, boolean isQueue, boolean unsubscribe, boolean resumeSession) throws LoginException, IOException { - SessionContextBuilder scb1 = new SessionContextBuilder(name+"_1", new ProtocolClientConnection(new FakeProtocol(this))); + SessionContextBuilder scb1 = new SessionContextBuilder(name + "_1", new ProtocolClientConnection(new FakeProtocol(this))); scb1.setReceiveMaximum(1); // ensure it is low scb1.setSessionExpiry(2); // 2 seconds, more than enough time scb1.setPersistentSession(true); // store the details Session delayedSession1 = createSession(scb1, this); - SessionContextBuilder scb2 = new SessionContextBuilder(name+"_2", new ProtocolClientConnection(new FakeProtocol(this))); + SessionContextBuilder scb2 = new SessionContextBuilder(name + "_2", new ProtocolClientConnection(new FakeProtocol(this))); scb2.setReceiveMaximum(1); // ensure it is low scb2.setSessionExpiry(2); // 2 seconds, more than enough time scb2.setPersistentSession(true); // store the details @@ -89,7 +119,7 @@ public void eventCleanUpTest(String name, String destinationName, boolean isQueu Session publisher = createSession(name, 60, 60, false, this); DestinationType destinationType = DestinationType.TOPIC; - if(isQueue){ + if (isQueue) { destinationType = DestinationType.QUEUE; } Destination destination = publisher.findDestination(destinationName, destinationType).get(); @@ -100,21 +130,21 @@ public void eventCleanUpTest(String name, String destinationName, boolean isQueu SubscribedEventManager subscription = delayedSession1.addSubscription(context); Assertions.assertNotNull(subscription); - for(int x=0;x destination.getStoredMessages() == 0); } WaitForState.waitFor(7, TimeUnit.SECONDS, () -> destination.getStoredMessages() == 0); + QueueCleanupTrace.record(destinationName, "TEST-DRAIN-END", -1, "stored", destination.getStoredMessages()); Assertions.assertEquals(0, destination.getStoredMessages()); close(publisher); } - private SubscriptionContext createContext(String destinationName,ClientAcknowledgement clientAcknowledgement, QualityOfService qos, int receiveMax){ + private SubscriptionContext createContext(String destinationName, ClientAcknowledgement clientAcknowledgement, QualityOfService qos, int receiveMax) { SubscriptionContextBuilder subContextBuilder2 = new SubscriptionContextBuilder(destinationName, clientAcknowledgement); subContextBuilder2.setReceiveMaximum(receiveMax); subContextBuilder2.setQos(qos); return subContextBuilder2.build(); - - } @Override public void sendMessage(@NotNull @NonNull MessageEvent messageEvent) { + QueueCleanupTrace.record(messageEvent.getDestinationName(), "TEST-CALLBACK", messageEvent.getMessage().getIdentifier()); messageEvent.getCompletionTask().run(); } } diff --git a/src/test/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SharedSubscriptionCleanupInterleavingTest.java b/src/test/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SharedSubscriptionCleanupInterleavingTest.java new file mode 100644 index 0000000000..6bdb7af0b5 --- /dev/null +++ b/src/test/java/io/mapsmessaging/engine/destination/subscription/impl/shared/SharedSubscriptionCleanupInterleavingTest.java @@ -0,0 +1,128 @@ +/* + * + * 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.subscription.impl.shared; + +import io.mapsmessaging.api.MessageBuilder; +import io.mapsmessaging.api.SubscriptionContextBuilder; +import io.mapsmessaging.api.features.ClientAcknowledgement; +import io.mapsmessaging.api.features.QualityOfService; +import io.mapsmessaging.api.message.Message; +import io.mapsmessaging.engine.Constants; +import io.mapsmessaging.engine.destination.DestinationImpl; +import io.mapsmessaging.engine.destination.subscription.SubscriptionContext; +import io.mapsmessaging.engine.destination.subscription.state.MessageStateManagerImpl; +import io.mapsmessaging.engine.destination.subscription.tasks.MessageDeliveredListener; +import io.mapsmessaging.engine.destination.subscription.transaction.AutoAcknowledgementController; +import io.mapsmessaging.engine.destination.subscription.transaction.FixedCreditManager; +import io.mapsmessaging.engine.session.SessionImpl; +import io.mapsmessaging.utilities.collections.bitset.BitSetFactoryImpl; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +class SharedSubscriptionCleanupInterleavingTest { + + private static final String DESTINATION = "queue/cleanup-interleaving"; + private static final long MESSAGE_ID = 42; + + @Test + void expiry_during_mapping_cleanup_must_not_strand_an_auto_acknowledged_message() throws Exception { + DestinationImpl destination = Mockito.mock(DestinationImpl.class); + Mockito.when(destination.getFullyQualifiedNamespace()).thenReturn(DESTINATION); + Mockito.when(destination.getTypePath()).thenReturn(List.of("type=QueueCleanupInterleavingTest")); + SharedSubscriptionManager registry = new SharedSubscriptionManager(DESTINATION); + Mockito.when(destination.findShareRegister(DESTINATION)).thenReturn(registry); + + SubscriptionContext context = new SubscriptionContextBuilder(DESTINATION, ClientAcknowledgement.AUTO).setQos(QualityOfService.AT_LEAST_ONCE).build(); + MessageStateManagerImpl state = new MessageStateManagerImpl("cleanup-interleaving", 42, new BitSetFactoryImpl(Constants.BITSET_BLOCK_SIZE)); + AutoAcknowledgementController sharedAcknowledgements = new AutoAcknowledgementController(new FixedCreditManager(10)); + SharedSubscription shared = new SharedSubscription(destination, context, "cleanup-interleaving", state, sharedAcknowledgements, "group"); + registry.add("group", shared); + + CountDownLatch cleanupEntered = new CountDownLatch(1); + CountDownLatch resumeCleanup = new CountDownLatch(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + SessionImpl firstSession = session("first"); + SessionImpl expiringSession = session("expiring"); + SessionImpl receivingSession = session("receiving"); + AutoAcknowledgementController receivingAcknowledgements = new AutoAcknowledgementController(new FixedCreditManager(10)); + + shared.addSession(firstSession, "first", context, new AutoAcknowledgementController(new FixedCreditManager(10))); + SessionSharedSubscription expiring = (SessionSharedSubscription) shared.addSession(expiringSession, "expiring", context, new AutoAcknowledgementController(new FixedCreditManager(10))); + shared.addSession(receivingSession, "receiving", context, receivingAcknowledgements); + + // Block at the session's storage boundary after the shared consumer iterator has started. + Mockito.doAnswer(invocation -> { + cleanupEntered.countDown(); + if (!resumeCleanup.await(5, TimeUnit.SECONDS)) { + throw new AssertionError("Expiry did not release delivery-mapping cleanup"); + } + return null; + }).when(firstSession).clearPacketIdMapping(Mockito.anyString(), Mockito.eq(MESSAGE_ID)); + + // Start at the completion boundary: the real message is in flight and its network write has completed. + Message message = new MessageBuilder().build(); + message.setIdentifier(MESSAGE_ID); + state.register(message); + state.allocate(message); + receivingAcknowledgements.sent(message); + Assertions.assertTrue(state.hasMessagesInFlight()); + Assertions.assertEquals(1, receivingAcknowledgements.size()); + + Future completion = executor.submit(() -> new MessageDeliveredListener(shared, receivingAcknowledgements).taskCall()); + Assertions.assertTrue(cleanupEntered.await(5, TimeUnit.SECONDS), "Completion did not reach delivery-mapping cleanup"); + Assertions.assertEquals(0, receivingAcknowledgements.size(), "The completion has already consumed its acknowledgement"); + + // This is the delete/cancel path invoked by session expiry. Keep a third member so the iterator must advance again. + expiring.delete(); + resumeCleanup.countDown(); + + Assertions.assertDoesNotThrow(() -> { + completion.get(5, TimeUnit.SECONDS); + }, () -> "Completion failed after concurrent expiry; outstanding=" + receivingAcknowledgements.size() + + ", inFlight=" + state.hasMessagesInFlight() + ", atRest=" + state.pending()); + Assertions.assertFalse(state.hasMessagesInFlight(), "The completed message must leave the in-flight set"); + Assertions.assertFalse(state.hasMessage(MESSAGE_ID), "The shared subscription must release the completed message"); + Mockito.verify(destination).complete(MESSAGE_ID); + } finally { + resumeCleanup.countDown(); + executor.shutdownNow(); + try { + Assertions.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS), "Completion worker did not terminate"); + } finally { + shared.close(); + } + } + } + + private SessionImpl session(String name) { + SessionImpl session = Mockito.mock(SessionImpl.class); + Mockito.when(session.getName()).thenReturn(name); + return session; + } +}