ReactiveStreamsConsumer over PollableChannel can silently lose a message on stop()

[artembilan]

Bug description

ReactiveStreamsConsumer over a PollableChannel (e.g. QueueChannel) can silently lose a message when stop() is called. There is a narrow race window in which the Flux built by IntegrationReactiveUtils.messageSourceToFlux(MessageSource) performs one more receive(0) against the channel after doStop() has disposed the subscription. The message is successfully drained from the channel, but because the downstream subscriber is already cancelled by the time the value is ready to be delivered, it is silently discarded (via Mono.create's post-terminate no-op — it doesn't even go through Reactor's Operators.onNextDropped hook). The message is neither delivered nor returned to the channel; it's just gone.

This surfaced as an intermittent CI failure of ReactiveStreamsConsumerTests.testReactiveStreamsConsumerPollableChannel(), e.g.:

java.lang.AssertionError:
Expecting actual:
  GenericMessage [payload=test2, headers={id=..., timestamp=...}]
and:
  GenericMessage [payload=test, headers={id=..., timestamp=...}]
to refer to the same object
    at ReactiveStreamsConsumerTests.testReactiveStreamsConsumerPollableChannel(ReactiveStreamsConsumerTests.java:238)

CI run: https://github.com/spring-projects/spring-integration/actions/runs/31523888182/job/93887348532

Root cause investigation

The existing test's stop→resend→restart pattern made this failure visible, but restart is not the trigger. Instrumenting QueueChannel.doReceive() and the mock Subscriber's invocation count showed:

1. It's genuine loss, not reordering — the "missing" message never reaches onNext at all (invocation count proves it).
2. The message is physically drained from the channel (doReceive(0) returns it) — it just never surfaces anywhere afterward.
3. It reproduces with stop() alone, with no restart whatsoever — confirmed against unmodified production code (Schedulers.single(), unmodified IntegrationReactiveUtils), at roughly 2-4 occurrences per 1500 iterations (~0.2%).

Steps to reproduce

Minimal loop that reproduces the loss without ever calling start() a second time:

@Test
@SuppressWarnings("unchecked")
public void reproMessageLossOnStopAlone() throws InterruptedException {
    int iterations = 1500;
    int stealCount = 0;
    for (int iteration = 0; iteration < iterations; iteration++) {
        QueueChannel testChannel = new QueueChannel();
        Subscriber<Message<?>> testSubscriber = (Subscriber<Message<?>>) Mockito.mock(Subscriber.class);
        BlockingQueue<Message<?>> messages = new LinkedBlockingQueue<>();

        willAnswer(i -> {
            messages.put((Message<?>) i.getArgument(0));
            return null;
        }).given(testSubscriber).onNext(any(Message.class));

        ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber);
        reactiveConsumer.setBeanFactory(TEST_INTEGRATION_CONTEXT);
        reactiveConsumer.afterPropertiesSet();
        reactiveConsumer.start();

        Message<?> testMessage = new GenericMessage<>("test");
        testChannel.send(testMessage);

        ArgumentCaptor<Subscription> captor = ArgumentCaptor.forClass(Subscription.class);
        verify(testSubscriber).onSubscribe(captor.capture());
        captor.getValue().request(1);

        assertThat(messages.poll(10, TimeUnit.SECONDS)).isSameAs(testMessage);

        // No restart - just stop(), then immediately send another message.
        reactiveConsumer.stop();
        testChannel.send(new GenericMessage<>("bait-" + iteration));

        // Tight busy-poll (a sleep gives the race too much time to resolve safely)
        for (int spin = 0; spin < 20000 && testChannel.getQueueSize() > 0; spin++) {
            // spin
        }

        if (testChannel.getQueueSize() == 0) {
            stealCount++; // "bait" vanished: drained from the channel, never delivered
        }
    }
    System.out.println("stealCount=" + stealCount + "/" + iterations);
    assertThat(stealCount).isZero();
}

On unmodified main, this fails with stealCount around 3/1500 locally (varies by machine/load — CI is more likely to hit it given shared runners and JIT warm-up variance).

Suspected mechanism

IntegrationReactiveUtils.messageSourceToFlux:

public static <T> Flux<Message<T>> messageSourceToFlux(MessageSource<T> messageSource) {
    return Mono.<Message<T>>create(monoSink ->
            monoSink.onRequest(value -> monoSink.success(messageSource.receive())))
            .doOnSuccess(...)
            .doOnError(...)
            .subscribeOn(Schedulers.single())
            .repeatWhenEmpty(...)
            .repeat()
            .retryWhen(...);
}

ReactiveStreamsConsumer.doStop() calls this.subscription.dispose(), which cancels asynchronously (via the subscribeOn boundary). There appears to be a window in which the repeat()/repeatWhenEmpty() composition performs (or has already dispatched) one more receive(0) concurrently with cancellation taking effect, and the retrieved value is dropped when monoSink.success(...) is called on an already-terminated sink.

I have not attempted a fix — messageSourceToFlux is public API consumed by every MessageSource-backed reactive adapter (file, ftp, jdbc, mongodb, etc.), so the right fix deserves discussion rather than a quick patch. A superficial monoSink.isCancelled() guard before calling receive() would narrow the window but is a check-then-act race, not a real fix.

Environment

• Reproduces on main (Java 17, Windows and macOS CI runner)

[arimu1]

Proposal for in-flight message rescue semantics on messageSourceToFlux cancel:

PollableChannel: prefer best-effort re-queue via non-blocking send(message, 0) so the message is not silently lost. This re-enters interceptors (wire taps, metrics, message-store persistence may run again) and appends at the tail, so strict FIFO is not guaranteed when the queue is not empty. When re-queue fails (bounded/full channel), fall back to AcknowledgmentCallback nack when the message carries one.

Other MessageSource types: nack via AcknowledgmentCallback when present; otherwise there is no generic re-queue API and loss remains possible (same as today for sources without ack headers).

Ack placement: autoAck still runs on successful Mono completion today; discard hooks fire before downstream delivery on the CANCELLED/TERMINATED emit paths, so those rescues happen without ack. Downstream async cancellation after ack is a separate concern worth addressing later.

Implementation in the PR follows this: doOnDiscard(Message.class, …) plus a sequence-local reactor.onNextDropped.local hook on the existing Mono.create chain.