No point in time replication bug - on moved flow

[mkaruza]

https://github.com/dragonflydb/dragonfly/actions/runs/15901480960

Caused by #5103

Before - we used version based, point in time isolation for full sync similarly to snapshotting.
After the change above we switched to more relaxed full-sync phase, where the full sync convergence is guaranteed due to the fact that we constantly send journal changes during the full sync, and our iterate loop in snapshot.cc guarantees the full coverage.

[adiholden]

The issue arises with the new algorithm for point-in-time replication, specifically during key bump-up. Currently, when a key is bumped up, it can be serialized after the onMoved handler is invoked but before the corresponding journal change is serialized.
This creates a race condition: the key's serialization may already include the effect of the command (e.g., the new list value), and then the journal change is serialized again — causing the change to be applied twice on the replica.

Example:
LPUSH mylist value
If mylist is bumped up and onMoved triggers serialization, it will be serialized with "value" already in the list. Then the journal records the same LPUSH mylist value operation. On the replica, this results in "value" being added twice — once from the deserialized state and once from replaying the journal.

[adiholden]

We need to find a way to do the bump up and the serialization only after we serialize the journal change.
Calling OnCbFinishBlocking after LogAutoJournalOnShard is not good enough I believe as OnCbFinishBlocking can bumpup items that were fetched by other callbacks as it uses fetched_items_ of dbslice.

[romange]

I did not understand why calling OnCbFinishBlocking after LogAutoJournalOnShard is not good enough.

[adiholden]

For example, consider two transactions running concurrently—one modifying key x and the other modifying key y.

• The callback handler for the transaction modifying x runs first. It adds x to fetched_items_ and performs the mutation on x.
• Next, the callback handler for the transaction modifying y runs. It adds y to fetched_items_ and performs the mutation on y.
• Then, LogAutoJournalOnShard is invoked for the transaction on x, which serializes the journal entry for that change.
• Afterward, OnCbFinishBlocking is triggered. Since both x and y are in fetched_items_, it bumps up both keys and serializes the buckets for x and y.
• Finally, LogAutoJournalOnShard for the transaction on y is executed.

This sequence leads to y being serialized before its journal change is recorded, resulting in a potential inconsistency on the replica.

[romange]

You mean the following?

1. X callback runs from fiber 1 and it preempts for some reason (LogAutoJournalOnShard (x) preempts for example)
2. Y callback runs from fiber 2 and it also preempts for some reason in LogAutoJournalOnShard (y).
3. Fiber 1 resumes and calls OnCbFinishBlocking for transaction X but it also picks up fetched_items_ items of transaction Y and fully serializes both x and y.

Question: can Fiber 2 preempt before it finalizes and flushes its journal changes? Don't we guarantee the order inside
SliceSnapshot::FlushSerialized by ordering the items we want to flush?

So in that case SliceSnapshot will order journal changes for X, Y before flushing the serialization blobs from OnCbFinishBlocking?

[adiholden]

We do keep order in serializing so if LogAutoJournalOnShard (y) was called before the call to serialize a bucket the journal change will be serialized before.

But I believe that the problem is that we can not know the order of the calls to serialization.
f.e a callback mutating x can run add x to fetched_items_ and will preempt on running another hop for executing the x transaction (if it s a multi hop transaction), than a callback y and we will call LogAutoJournalOnShard(y) without preemtion than OnCbFinishBlocking will run serializing bucket of x and y before LogAutoJournalOnShard(x) is called

[romange]

Ok to summarize:

• OnCbFinishBlocking is called at the end of each callback run
• LogAutoJournalOnShard is called on each shard at the end of the command run, after a concluding callback is called.
• Currently OnCbFinishBlocking is always called before LogAutoJournalOnShard but @adiholden says that swapping the order is not enough because the fix won't cover the case of another transaction Y flushing the bumpup changes of X, causing serialization of entries of X before LogAutoJournalOnShard(X) is called. LogAutoJournalOnShard is not idempotent so serialization with already applied changes before LogAutoJournalOnShard breaks correctness of the replication.