IllegalStateException in ProducerController upon restart with an empty unconfirmed buffer (Durable Queue)

[diegovallone]

When using Reliable Delivery with a standard ProducerController backed by a durable queue (e.g., EventSourcedProducerQueue), restarting the producer causes an immediate crash if all previous messages were confirmed before the restart.

The ProducerController successfully reloads its state but crashes with the following error before a ConsumerController can even establish demand:
java.lang.IllegalStateException: Unexpected Msg when no demand, requested true, requestedSeqNr 1, currentSeqNr X (where X is the actual restored sequence number).

Note: The WorkPullingProducerController handles state initialization differently and is unaffected by this issue.

Steps to Reproduce:
I have verified this behavior using an isolated test with Pekko's in-memory journal.

1. Start a ProducerController with a durable queue and a ConsumerController.
2. Send a message, and allow the consumer to receive and confirm it (clearing the unconfirmed buffer).
3. Wait for the ProducerController to receive the next RequestNext, ensuring the confirmed state is fully written to the durable queue.
4. Stop and restart the ProducerController to simulate a restart, crash, or deployment.
5. The framework emits a RequestNext with seqNr = 1 instead of the restored sequence number. When the producer supplies the next message, it fails the internal demand check and crashes with an IllegalStateException.

(See the attached ProducerControllerBugTest.scala snippet below for the fully reproducible test case. Note that the test is inverted: it succeeds if it can reproduce the bug).
ProducerControllerBugTest.txt

Root Cause:
In ProducerControllerImpl.scala, the state recovery logic ignores the loaded sequence number when initializing the demand window and requesting the next message from the local producer:

In createState, requestedSeqNr is hardcoded to 1L instead of adopting loadedState.currentSeqNr.

In becomeActive, if state.unconfirmed.isEmpty is true, it hardcodes 1L and 0L into the RequestNext message and the flight recorder, rather than using state.currentSeqNr and state.confirmedSeqNr.

(See the attached bugfix.txt which contains a git diff of the fix that apparently solves this issue)
bugfix.txt

[pjfanning]

@diegovallone would you be able to produce a PR with the code change and the test case added as a scalatest - with the test case passing?

[pjfanning]

This issue does look like a probable bug. Seems strange that it hasn't come up before.

[pjfanning]

I got Copilot to create a PR with the suggested fix. #2762

I'll see how the CI build goes and if it is feasible to add extra test coverage. Any suggestions are appreciated.

[pjfanning]

@diegovallone what do you think of Copilot's suggestion for a test based on your test code? pjfanning#21 (comment)

I think there is an argument for just sticking with testing actor-typed directly like what #2762 does already. I'm not convinced adding a persistence-typed test adds too much more - but I'm open to other opinions.

[diegovallone]

@pjfanning thanks for your fast response on this, I really appreciate it.

I like Copilot's suggestion. The unit test proves the logic is correct in the mocked scenario. The persistence-typed test proves the logic actually solves the real-world problem, and confirms that the mocked scenario is realistic and even surfaces a corner case (the write-behind) that I wasn't aware of.

You have a point that it’s not strictly needed, but I can't complain about having an integration test confirming the fix with non-mocked code.

Especially in this situation... I'm really surprised this bug made it this far (all the way from akka, so many years ago). I feel like I'm the first person to actually use this feature, which is honestly a bit scary.

[diegovallone]

@pjfanning I saw that the PR with the fix has been merged to the main branch, thank you very much.

But I'm currently using 1.2.1.
How is the usual process for these type of bugs that break functionality in previous versions?
What's the usual timeline for having it back-ported?

That will help me decide if I should implement a workaround or just wait for the new version, because this soon will become a blocker for me.
Thanks!

[pjfanning]

@diegovallone how is this a regression? That is, when did Pekko ever support this properly?

The fix is in 2.0 snapshot if you want it. You can always build your own jar too - one that cherry picks the changes that you need.

The aim will be to get out a 1.6.0 release eventually but an 1.5.0 (which does not contain this fix) is still in review and we don't have a lot of volunteers to review releases. There are other modules that we need to release so getting back to another Pekko 1.6.0 release could take a while.