fix(sync): skip already processed messages

xmtp / libxmtp

LibXMTP is a shared library encapsulating the core functionality of the XMTP messaging protocol, such as cryptography, networking, and language bindings.

MIT License

47 stars 20 forks source link

fix(sync): skip already processed messages #1267

Closed mchenani closed 1 week ago

mchenani commented 1 week ago

Description: Added a check to skip processing messages that have already been handled, returning an AlreadyProcessed error when a duplicate message is detected.

Issue: Previously, a race condition or repeated messages from an integrator feed could cause the message processor to reprocess already-handled messages. This could result in multiple wrong-epoch errors, increasing the risk of missing critical messages due to excessive error handling within the message stream.

Tests:

[x] Added unit tests. Created a test to simulate the scenario where a message has already been processed, ensuring the new check correctly identifies and skips duplicates.

neekolas commented 1 week ago

@mchenani Are we sure this change actually does anything?

process_for_id checks that the call to update_cursor returns true (the cursor is less than the message's ID). If not, it returns the same AlreadyProcessed error.

This PR repeats that check but before we start the transaction.

Was the test that was added failing before the change? If so, maybe there is some other bug we should be looking at.

mchenani commented 1 week ago

@neekolas thanks for pointing this out, I thought the same but in my tests I noticed in some race conditions we hit deadlocks and we still processing the already processed messages! I have some more improvements to come to reduce the deadlock issue, I noticed with this change at least the wrong-epoch was blocked and not reproduced in certain cases.