Inbox - duplication in processing messages during restart of consumer (kafka) #1725

Describe the bug

Removal of processed messages is performed by DurabilityAgent - cyclical batch processing of database operations. In these batch there is DeleteExpiredEnvelopesOperation and CheckRecoverableIncomingMessagesOperation. Options used to control it are:

• DurabilitySettings.KeepAfterMessageHandling (default 5m)
• DurabilitySettings.ScheduledJobFirstExecution (default 500-5000 ms)
• DurabilitySettings.ScheduledJobPolling (default 5s)

Scenario:

1. Messages are processing and stored in Inbox with status Handled
2. Application is disabled
3. After a few hours, the application resumes operation.
4. If there are a lot of services connected to the topic rebalancing can occur about 1 min after start (kafka says that message duplication is possible during rebalancing)
5. The job that deletes processed messages runs, for example, 10 seconds after the application is launched and deletes all messages whose keep_until field <= utcNow (a few hours have passed since processing, so it will delete all of them at once)
6. During rebalancing, a message that has already been processed is inserted and processed again.

If we set the JobFirstExecution variable to 2 hours from the application start - this will also delay the call to the first CheckRecoverableIncomingMessagesOperation - which means that the unprocessed messages in the inbox will be moved by at least 2 hours from the application start - we lose the processing order (strict order scenario).

Still, if we control that the message can be deleted (KeepAfterMessageHandling) after 4 hours of processing, if the unavailability lasts longer, duplication is possible during rebalancing.

Maybe when deleting, instead of comparing the time from the keep_until field to the current time, it should be compared to the time of publication to the transport layer or addition to the inbox for the last received message, and this could be configured using options in DurabilitySettings.

To Reproduce
The problem is non-deterministic and occurs when the transport layer is Kafka. Scenario is described in section above.

Expected behavior
CheckRecoverableIncomingMessagesOperation and DeleteExpiredEnvelopesOperation could be processed independently when we want to process in sequential order. If there are messages with the status 'handled' in the inbox, they should not be deleted immediately after the application is started (if it was unavailable for some time, for example) in order to wait some controllable period of time for rebalancing after reconnecting to the transport layer

Additional context
Problem was discovered in cooperation with @dominikjeske.
Wolverine version: 4.10.1