fix(cluster): Join on specified attempt id

[chakaz]

The Bug

Before this fix, source nodes would send FIN entries to target nodes (in all thread flows), and would then send a DFLYMIGRATE ACK command to verify that all flows received the FIN in time.

If they didn't, the source node would retry this logic in a loop, until successful.

The problem is that, in some rear cases, one or more of the flows would indeed be in a FIN state, but of a previous FIN that is already outdated. If that's indeed the case, all data between that FIN and the next FIN(s) will be lost.

The Fix

We already have an attempt id that we send in the DFLYMIGRATE ACK command, and return it in the response. This fix utilizes the same attempt id to be sent to all flows, and then when joined, we make sure we join on the correct (latest) attempt id.

Unfortunately, we can't use FIN opcode now, because the protocol does not send any additional metadata for this opcode. I chose to use LSN instead because it has exactly the fields that we need, and one could possibly think of Log Sequence Number as an attempt id, but I could change that if it's unclear or too hacky.

Testing

To reproduce this, one needs to lower
--slot_migration_connection_timeout_ms significantly, say to 500ms. This would fail, on my laptop, every ~2 runs.

With this fix, it runs hundreds of times and never reproduces.

Fixes #3257

[adiholden]

In cases that FIN opcode was sent and than source node retries to finalize it will continue sending more data and therefore tx_reader.NextTxData in the next line will return nullopt?
if so why didnt we get it when you reproduced the bug? is it due to timeout on socket?

do we still need this logic of calling tx_reader.NextTxData in the next line with your fix?

[adiholden]

also I am kind of confused with the logic of the blocking counter now as we wait on it in the join but we might decrease it but it could be that we returned in line 63 but its actually not the attempt that we are currently waiting on

[chakaz]

if so why didnt we get it when you reproduced the bug? is it due to timeout on socket?

Yes, it will continue sending data. But there is a short period of time after Dec() and before Add() that it is able to "catch" those earlier FINs.
This is why you must set a pretty aggressive timeout, for this to happen a lot.

do we still need this logic of calling tx_reader.NextTxData in the next line with your fix?

Yes, we still need it, otherwise we won't be able to continue after a timeout. The overall protocol/design is ok, we just need to confirm that we're waiting on the right attempt on both source and target.

also I am kind of confused with the logic of the blocking counter now as we wait on it in the join but we might decrease it but it could be that we returned in line 63 but its actually not the attempt that we are currently waiting on

If we return in line 63 it means that the socket was closed, so either the source node decided to finalize, or there's something wrong, but anyway there is no way to continue with the loop because there's nothing to read from

[adiholden]

why do we have a while loop here? why not just return if
std::all_of(shard_flows_.begin(), shard_flows_.end(),
[&](const auto& flow) { return flow->GetLastAttempt() == attempt; }))
is false?

We will call this join function on each DFLYMIGRATE ACK command sent, how does this works?

[chakaz]

We are given a timeout here, let's say 2s (default).
So we should be returning if (a) we reached the correct join state, or (b) if 2s have passed.
This loop is here to "protect" us from returning earlier than 2s but without (a), in case we were able to join all on an earlier FIN (now LSN).
So for example if attempt == 4, but one of more of the flows are currently stopped on attempt 3, we will try again, with a timeout equals to the remainder of the time.
I hope this makes sense. If you could phrase some comment to explain that, I'm more than happy to add it!