#318: release() no longer strands a delivery that lands while it is running
AmqpReplyInbox.release() used held.remove(target) then iterated the old map. A delivery landing on the consumer work-pool thread after the remove (basicCancel does not flush one already handed to that pool) hit deliverCallback's computeIfAbsent, found the key gone, and created a brand-new map release() never looks at again — delivered-but-unacked forever, never requeued, never redelivered (#298 only closed the "already in held when release runs" case). Fix: release() swaps in a RELEASED tombstone via held.compute(...) instead of held.remove(...). ConcurrentHashMap serializes compute/ computeIfAbsent calls for the same key against each other, so whichever of release() and a concurrent deliverCallback runs first is fully visible to the other — no gap. deliverCallback checks for the tombstone and nacks-with-requeue instead of recreating a map; peek/ack treat it as empty; own() clears a stale tombstone so a target is never poisoned if its id is ever reused (the issue's own text says id reuse doesn't happen, but the tombstone would otherwise sit in `held` forever either way). New test AmqpReplyInboxReleaseRaceTest forces the actual interleaving with a latch (blocks release() inside its nack loop, which is only reachable after the tombstone swap, then fires a concurrent delivery) rather than a sequential call — a sequential test would not have caught this, since #298's own contract test forces settlement before release() runs. Mutation-tested: reverting the fix makes this test fail with "expected: <2> but was: <1>" (m1 never nacked); restored after confirming that failure.
This commit is contained in:
Dai Ha
@@ -22,6 +22,7 @@ import java.util.concurrent.ConcurrentSkipListMap;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
/**
|
||||
* AMQP-backed {@link ReplyInbox} (CB-307 Stage 2): genuine cross-restart durability behind the same
|
||||
@@ -93,8 +94,23 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
|
||||
private final Channel channel;
|
||||
/** All channel operations (publish/declare/ack/cancel) serialize on this — a Channel is not thread-safe. */
|
||||
private final Object channelLock = new Object();
|
||||
/** target → (msgId → held delivery). Per-target map is guarded by synchronizing on itself. */
|
||||
/**
|
||||
* target → (msgId → held delivery). Per-target map is guarded by synchronizing on itself.
|
||||
*
|
||||
* <p><strong>CB-318 tombstone.</strong> The value {@link #RELEASED} is a reserved sentinel: it
|
||||
* marks a target whose {@link #release} has already run, so {@link #deliverCallback} can tell a
|
||||
* delivery landing after release() apart from a fresh target it has never seen. See both methods'
|
||||
* javadoc for why a plain {@code held.remove(target)} is not enough.
|
||||
*/
|
||||
private final ConcurrentHashMap<String, LinkedHashMap<String, Held>> held = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* CB-318 sentinel stored in {@link #held} for a target whose {@link #release} has already run.
|
||||
* Never mutated — every read site compares it by reference ({@code ==}) before touching it as a
|
||||
* map, because it is a single object shared across every released target and calling a mutator on
|
||||
* it would corrupt state for all of them.
|
||||
*/
|
||||
private static final LinkedHashMap<String, Held> RELEASED = new LinkedHashMap<>();
|
||||
/** Targets whose queue is declared and consumer is running, mapped to their broker consumer tag. */
|
||||
private final ConcurrentHashMap<String, String> consumerTags = new ConcurrentHashMap<>();
|
||||
|
||||
@@ -201,6 +217,12 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
|
||||
channel.queueDeclare(queue, true, false, false, null); // durable, non-exclusive, keep on idle
|
||||
String tag = channel.basicConsume(queue, false, deliverCallback(target), _ -> { });
|
||||
consumerTags.put(target, tag);
|
||||
// CB-318: drop a stale RELEASED tombstone from a prior ownership of this same target
|
||||
// string, so a delivery under this fresh consumer is held normally instead of being
|
||||
// nacked forever by deliverCallback's RELEASED check. Safe to do here, still under
|
||||
// channelLock: no delivery for the consumer tag just registered above can reach
|
||||
// deliverCallback before this basicConsume call returns.
|
||||
held.remove(target, RELEASED);
|
||||
log.debug("AMQP inbox owns queue {} for target {}", queue, target);
|
||||
} catch (IOException e) {
|
||||
throw new IllegalStateException("cannot own queue " + queue, e);
|
||||
@@ -243,6 +265,32 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
|
||||
* later connection drop even though this release did not manage to requeue it immediately. A
|
||||
* failed {@code basicCancel} still throws, unchanged from before this fix — that failure means
|
||||
* the consumer may still be attached, so best-effort requeue is not attempted underneath it.
|
||||
*
|
||||
* <p><strong>CB-318: {@code held.remove(target)} alone leaves a second window open.</strong> The
|
||||
* bullet above already explains why cancelling first does not save a tag from going stale — but
|
||||
* that only accounts for a delivery landing before this method starts touching {@link #held}.
|
||||
* {@code basicCancel} stops <em>new</em> dispatches; it does not flush one already handed to the
|
||||
* consumer work pool. So a delivery can still land on that pool's thread and reach
|
||||
* {@link #deliverCallback} at any point during, or after, this method's body — and a plain
|
||||
* {@code held.remove(target)} does nothing to stop it: {@code deliverCallback}'s
|
||||
* {@code computeIfAbsent} finds the key gone and happily creates a brand-new map under it, which
|
||||
* this method — already past its {@code remove} — never looks at again. That entry then sits
|
||||
* delivered-but-unacked on {@link #channel} until the whole inbox closes: never requeued, never
|
||||
* redelivered, and {@link #peek} is never called again for a target nothing owns any more.
|
||||
*
|
||||
* <p>The fix is {@link #held}{@code .compute(target, ...)} instead of {@code remove}: it takes
|
||||
* whatever was held (to nack, same as before) and, in the same atomic step, leaves the
|
||||
* {@link #RELEASED} tombstone behind instead of an absent key. {@code computeIfAbsent} and
|
||||
* {@code compute} calls for the same key are mutually exclusive in {@link ConcurrentHashMap} —
|
||||
* whichever of this call and a concurrent {@code deliverCallback} runs first is fully visible to
|
||||
* the other, with no gap between them. So a delivery that loses the race sees a real map here and
|
||||
* gets nacked by the loop below, same as always; a delivery that wins the race (runs first) is
|
||||
* itself nacked by that same loop, once it settles into {@code held}. A delivery that arrives once
|
||||
* this method has stored {@link #RELEASED} finds it via {@code computeIfAbsent} and refuses itself
|
||||
* — see {@link #deliverCallback}. Either way nothing is silently retained forever, satisfying the
|
||||
* ticket's invariant against dropping a message. This closes the window rather than merely
|
||||
* narrowing it — correctness does not depend on how much time elapses between the swap and this
|
||||
* method returning.
|
||||
*/
|
||||
@Override
|
||||
public void release(String target) {
|
||||
@@ -255,8 +303,13 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
|
||||
throw new IllegalStateException("cannot cancel consumer for " + target, e);
|
||||
}
|
||||
}
|
||||
var perTarget = held.remove(target);
|
||||
if (perTarget != null) {
|
||||
AtomicReference<LinkedHashMap<String, Held>> previouslyHeld = new AtomicReference<>();
|
||||
held.compute(target, (_, v) -> {
|
||||
previouslyHeld.set(v);
|
||||
return RELEASED;
|
||||
});
|
||||
var perTarget = previouslyHeld.get();
|
||||
if (perTarget != null && perTarget != RELEASED) {
|
||||
synchronized (perTarget) {
|
||||
for (Held h : perTarget.values()) {
|
||||
try {
|
||||
@@ -326,7 +379,7 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
|
||||
@Override
|
||||
public List<InboxMessage> peek(String target) {
|
||||
var perTarget = held.get(target);
|
||||
if (perTarget == null) {
|
||||
if (perTarget == null || perTarget == RELEASED) {
|
||||
return List.of();
|
||||
}
|
||||
synchronized (perTarget) {
|
||||
@@ -337,7 +390,7 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
|
||||
@Override
|
||||
public void ack(String target, String msgId) {
|
||||
var perTarget = held.get(target);
|
||||
if (perTarget == null) {
|
||||
if (perTarget == null || perTarget == RELEASED) {
|
||||
return;
|
||||
}
|
||||
Held h;
|
||||
@@ -370,6 +423,20 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
|
||||
}
|
||||
String content = new String(delivery.getBody(), StandardCharsets.UTF_8);
|
||||
var perTarget = held.computeIfAbsent(target, _ -> new LinkedHashMap<>());
|
||||
if (perTarget == RELEASED) {
|
||||
// CB-318: release() already ran for this target and left the RELEASED tombstone in
|
||||
// held (see release()'s javadoc) — computeIfAbsent() is guaranteed to see it rather
|
||||
// than recreate a fresh map, because ConcurrentHashMap serializes compute/
|
||||
// computeIfAbsent calls for the same key against each other. Refuse the delivery
|
||||
// instead of holding it somewhere release() will never look at again: requeue it, the
|
||||
// same way release() nacks its own held entries, so a later owner (or a connection
|
||||
// drop) can still recover it. This does not need channelLock across a broker round
|
||||
// trip — basicNack, like the duplicate-ack case just below, does not wait for one.
|
||||
synchronized (channelLock) {
|
||||
channel.basicNack(tag, false, true);
|
||||
}
|
||||
return;
|
||||
}
|
||||
boolean duplicate;
|
||||
synchronized (perTarget) {
|
||||
if (perTarget.containsKey(msgId)) {
|
||||
|
||||
@@ -0,0 +1,264 @@
|
||||
package dev.ltms.fleet.msg;
|
||||
|
||||
import com.rabbitmq.client.AMQP;
|
||||
import com.rabbitmq.client.Channel;
|
||||
import com.rabbitmq.client.Connection;
|
||||
import com.rabbitmq.client.DeliverCallback;
|
||||
import com.rabbitmq.client.Delivery;
|
||||
import com.rabbitmq.client.Envelope;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.Timeout;
|
||||
|
||||
import java.lang.reflect.InvocationHandler;
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* CB-318: a delivery landing on the consumer work-pool thread <em>after</em>
|
||||
* {@link AmqpReplyInbox#release} has already swapped the target's {@code held} entry for its
|
||||
* tombstone, but <em>before</em> {@code release()} itself returns, must be nacked-with-requeue —
|
||||
* never silently retained in a fresh map {@code release()} has already stopped looking at.
|
||||
*
|
||||
* <p><strong>This forces the actual interleaving, not a sequence of calls.</strong> {@code release()}
|
||||
* runs on its own thread and is made to block <em>inside</em> its nack loop, via a fake
|
||||
* {@link Channel} whose {@code basicNack} blocks on its first invocation. That block is only
|
||||
* reachable after {@code release()}'s {@code held.compute(...)} has already swapped in the
|
||||
* {@code RELEASED} tombstone (the compute call happens strictly before the loop that calls
|
||||
* {@code basicNack}), so observing it is direct, ordering-guaranteed proof that the tombstone is in
|
||||
* place and {@code release()} has not yet returned — still holding {@code channelLock} — when a
|
||||
* second thread fires {@code own()}'s captured {@link DeliverCallback} for a brand-new message on the
|
||||
* same target. No mocking library is on the classpath, so the fake broker is a {@link Proxy}, the
|
||||
* same pattern {@code AmqpReplyInboxRecoveryRaceTest} already uses.
|
||||
*
|
||||
* <p><strong>What this does and does not prove.</strong> It proves that a delivery whose
|
||||
* {@code computeIfAbsent} call is ordered strictly after {@code release()}'s tombstone swap — while
|
||||
* {@code release()} is still running — is nacked-with-requeue rather than silently parked forever.
|
||||
* It does not drive a real broker: {@code basicNack} here is a recorded call on a fake channel, not a
|
||||
* verified requeue-and-redeliver. That half of the contract (a nacked-with-requeue delivery really
|
||||
* does come back to a later owner) is already covered against a real broker by
|
||||
* {@code AmqpReplyInboxContractTest.releaseCancelsConsumerAndRequeuesHeldDeliveryForRecovery}, which
|
||||
* this test does not duplicate.
|
||||
*/
|
||||
class AmqpReplyInboxReleaseRaceTest {
|
||||
|
||||
@Test
|
||||
@Timeout(15)
|
||||
void deliveryArrivingWhileReleaseIsStillRunningIsNackedNotStranded() throws Exception {
|
||||
String target = "worker-release-race";
|
||||
|
||||
List<long[]> nacks = new CopyOnWriteArrayList<>(); // {deliveryTag, requeue(1/0)}
|
||||
List<Long> acks = new CopyOnWriteArrayList<>();
|
||||
AtomicReference<DeliverCallback> deliverCallback = new AtomicReference<>();
|
||||
CountDownLatch nackStarted = new CountDownLatch(1);
|
||||
CountDownLatch releaseMayFinishNack = new CountDownLatch(1);
|
||||
AtomicInteger nackCallCount = new AtomicInteger();
|
||||
|
||||
Channel consumeChannel = fakeConsumeChannel(deliverCallback, nacks, acks, nackCallCount,
|
||||
nackStarted, releaseMayFinishNack);
|
||||
Channel publishChannel = fakeInertChannel();
|
||||
Connection connection = fakeConnection(consumeChannel, publishChannel);
|
||||
|
||||
AmqpReplyInbox inbox = new AmqpReplyInbox(connection, AmqpReplyInbox.DEFAULT_PREFETCH);
|
||||
inbox.own(target);
|
||||
assertTrue(deliverCallback.get() != null, "own() must have registered a DeliverCallback");
|
||||
|
||||
// Seed one already-held delivery (m0) so release()'s nack loop has something to iterate, and
|
||||
// therefore somewhere to block, before it can return.
|
||||
deliverCallback.get().handle("ctag", delivery(1L, "m0", "first"));
|
||||
|
||||
AtomicReference<Throwable> releaseError = new AtomicReference<>();
|
||||
Thread releaseThread = new Thread(() -> {
|
||||
try {
|
||||
inbox.release(target);
|
||||
} catch (Throwable t) {
|
||||
releaseError.set(t);
|
||||
}
|
||||
}, "release-under-test");
|
||||
releaseThread.start();
|
||||
|
||||
// This latch only fires from inside the fake channel's basicNack — i.e. from inside
|
||||
// release()'s nack loop, which release()'s code only reaches AFTER held.compute(...) has
|
||||
// already swapped in RELEASED. Waiting for it is direct proof the swap has happened and
|
||||
// release() has not yet returned (it is stuck mid-loop, still holding channelLock).
|
||||
assertTrue(nackStarted.await(10, TimeUnit.SECONDS),
|
||||
"release() never reached its nack loop — it may not have started");
|
||||
|
||||
// The exact interleaving CB-318 describes: a delivery for a NEW message on the same target
|
||||
// lands on the "consumer work-pool thread" (this second thread) while release() is still
|
||||
// running. With the pre-fix code (a bare held.remove(target)) this created a brand-new map
|
||||
// under computeIfAbsent that release() — already past its remove — never looks at again.
|
||||
AtomicReference<Throwable> deliveryError = new AtomicReference<>();
|
||||
Thread deliveryThread = new Thread(() -> {
|
||||
try {
|
||||
deliverCallback.get().handle("ctag", delivery(2L, "m1", "second"));
|
||||
} catch (Throwable t) {
|
||||
deliveryError.set(t);
|
||||
}
|
||||
}, "concurrent-delivery");
|
||||
deliveryThread.start();
|
||||
|
||||
// Head start for the delivery thread to reach (and, on the fixed code, block on)
|
||||
// channelLock — release() still holds it at this point, so a correct fix cannot have
|
||||
// resolved m1's nack yet. Purely in-memory work (computeIfAbsent, a reference compare)
|
||||
// separates deliveryThread.start() from that block point, so 300ms is a large margin, not a
|
||||
// tight timing assumption.
|
||||
Thread.sleep(300);
|
||||
assertEquals(1, nackCallCount.get(),
|
||||
"the concurrent delivery must not resolve its nack before release() gives up "
|
||||
+ "channelLock — if this is 2 already, the interleaving below is not being "
|
||||
+ "tested, only a sequential call");
|
||||
|
||||
releaseMayFinishNack.countDown(); // let release() finish nacking m0 and return
|
||||
assertTrue(releaseThread.join(Duration.ofSeconds(10)), "release() did not finish");
|
||||
assertTrue(deliveryThread.join(Duration.ofSeconds(10)), "the concurrent delivery did not finish");
|
||||
|
||||
assertNull(releaseError.get(), "release() threw: " + releaseError.get());
|
||||
assertNull(deliveryError.get(), "the concurrent delivery threw: " + deliveryError.get());
|
||||
|
||||
assertEquals(2, nacks.size(),
|
||||
"both the pre-held m0 and the concurrently-arriving m1 must be nacked, got: "
|
||||
+ nacks.stream().map(n -> "[tag=" + n[0] + " requeue=" + n[1] + "]").toList());
|
||||
assertTrue(nacks.stream().allMatch(n -> n[1] == 1L),
|
||||
"invariant 1 (never drop): every nack must set requeue=true");
|
||||
assertTrue(nacks.stream().anyMatch(n -> n[0] == 1L), "m0's delivery tag must be nacked");
|
||||
assertTrue(nacks.stream().anyMatch(n -> n[0] == 2L),
|
||||
"m1 — delivered while release() was still running, after the tombstone swap — must be "
|
||||
+ "nacked, not silently retained in a map release() will never look at again");
|
||||
assertTrue(acks.isEmpty(), "invariant 1 (never drop): a held reply must never be basicAck'd");
|
||||
}
|
||||
|
||||
private static Delivery delivery(long tag, String msgId, String body) {
|
||||
Envelope envelope = new Envelope(tag, false, "", "irrelevant");
|
||||
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();
|
||||
return new Delivery(envelope, props, body.getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
|
||||
/** A {@link Proxy}-backed consume {@link Channel}: blocks the FIRST {@code basicNack} call on
|
||||
* {@code releaseMayFinishNack}, after signalling {@code nackStarted} — everything else records
|
||||
* the call and returns a harmless default, matching the style already used by
|
||||
* {@code AmqpReplyInboxRecoveryRaceTest}. */
|
||||
private static Channel fakeConsumeChannel(AtomicReference<DeliverCallback> deliverCallback,
|
||||
List<long[]> nacks, List<Long> acks,
|
||||
AtomicInteger nackCallCount,
|
||||
CountDownLatch nackStarted,
|
||||
CountDownLatch releaseMayFinishNack) {
|
||||
InvocationHandler handler = (proxy, method, args) -> {
|
||||
String name = method.getName();
|
||||
if (name.equals("basicConsume")) {
|
||||
deliverCallback.set((DeliverCallback) args[2]);
|
||||
return "ctag";
|
||||
}
|
||||
if (name.equals("basicNack")) {
|
||||
long tag = (long) args[0];
|
||||
boolean requeue = (boolean) args[2];
|
||||
if (nackCallCount.incrementAndGet() == 1) {
|
||||
nackStarted.countDown();
|
||||
if (!releaseMayFinishNack.await(10, TimeUnit.SECONDS)) {
|
||||
throw new IllegalStateException("test never released the nack latch");
|
||||
}
|
||||
}
|
||||
nacks.add(new long[] {tag, requeue ? 1L : 0L});
|
||||
return null;
|
||||
}
|
||||
if (name.equals("basicAck")) {
|
||||
acks.add((long) args[0]);
|
||||
return null;
|
||||
}
|
||||
if (name.equals("equals")) {
|
||||
return proxy == args[0];
|
||||
}
|
||||
if (name.equals("hashCode")) {
|
||||
return System.identityHashCode(proxy);
|
||||
}
|
||||
if (name.equals("toString")) {
|
||||
return "FakeConsumeChannel";
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
};
|
||||
return (Channel) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
|
||||
new Class<?>[] {Channel.class}, handler);
|
||||
}
|
||||
|
||||
/** A {@link Proxy}-backed {@link Channel} that answers every call with a harmless default — used
|
||||
* as the publish channel, which this test never actually publishes on. */
|
||||
private static Channel fakeInertChannel() {
|
||||
InvocationHandler handler = (proxy, method, args) -> {
|
||||
String name = method.getName();
|
||||
if (name.equals("equals")) {
|
||||
return proxy == args[0];
|
||||
}
|
||||
if (name.equals("hashCode")) {
|
||||
return System.identityHashCode(proxy);
|
||||
}
|
||||
if (name.equals("toString")) {
|
||||
return "FakeInertChannel";
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
};
|
||||
return (Channel) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
|
||||
new Class<?>[] {Channel.class}, handler);
|
||||
}
|
||||
|
||||
/** A {@link Proxy}-backed {@link Connection} handing out {@code first} then {@code second} from
|
||||
* successive {@code createChannel()} calls, matching {@link AmqpReplyInbox}'s constructor. */
|
||||
private static Connection fakeConnection(Channel first, Channel second) {
|
||||
AtomicInteger calls = new AtomicInteger();
|
||||
InvocationHandler handler = (proxy, method, args) -> {
|
||||
String name = method.getName();
|
||||
if (name.equals("createChannel") && (args == null || args.length == 0)) {
|
||||
return calls.getAndIncrement() == 0 ? first : second;
|
||||
}
|
||||
if (name.equals("equals")) {
|
||||
return proxy == args[0];
|
||||
}
|
||||
if (name.equals("hashCode")) {
|
||||
return System.identityHashCode(proxy);
|
||||
}
|
||||
if (name.equals("toString")) {
|
||||
return "FakeConnection";
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
};
|
||||
return (Connection) Proxy.newProxyInstance(AmqpReplyInboxReleaseRaceTest.class.getClassLoader(),
|
||||
new Class<?>[] {Connection.class}, handler);
|
||||
}
|
||||
|
||||
private static Object defaultValue(Class<?> type) {
|
||||
if (!type.isPrimitive() || type == void.class) {
|
||||
return null;
|
||||
}
|
||||
if (type == boolean.class) {
|
||||
return Boolean.FALSE;
|
||||
}
|
||||
if (type == long.class) {
|
||||
return 0L;
|
||||
}
|
||||
if (type == short.class) {
|
||||
return (short) 0;
|
||||
}
|
||||
if (type == byte.class) {
|
||||
return (byte) 0;
|
||||
}
|
||||
if (type == char.class) {
|
||||
return (char) 0;
|
||||
}
|
||||
if (type == double.class) {
|
||||
return 0.0d;
|
||||
}
|
||||
if (type == float.class) {
|
||||
return 0.0f;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user