Do the failover immediately if the replica is the best ranked replica

[enjoy-binbin]

In #2023 (#2209, etc.), we are exploring ways to make failover faster, that is,
to minimize the delay.

When a node is marked as FAIL and before the failover starts, there is a delay of
500-1000ms. The original purpose of this delay:

1. Allow FAIL to propagate to at least a majority of the primaries. This makes sure
   they will vote when a replica sends failover auth request.
2. Allow replicas to exchange their offsets, so they will have a correct view of
   their own rank.

We want to minimize this delay while ensuring safety. It is useful for example in
these cases:

1. If there is only one replica, then we don't need any delay, or
2. If there are more replicas, with a fast network, the replicas can exchange the
   offsets very quickly and start the failover within a few milliseconds instead
   of 500-1000ms.

In this PR, when we can be sure that the replica is the best ranked replica, we let
it initiate a failover immediately and completely remove the delay.

How to ensure safety?

1. To make sure this replica has the best rank, it only skips the delay if it is
   sure that it have the best rank and that all replicas in the same shard agree that
   the primary is failing. A new flag CLUSTER_NODE_MY_PRIMARY_FAIL is introduced to
   indicate that each replica has marked its primary as FAIL. If all replicas say that
   the primary is failing, we also know that the offset is not updated, because the
   offset is not incrementing when the primary is failing. We can skip the delay only
   if we have received a message from all replicas and they all have set this flag.

2. To make sure the primaries will vote even if they didn't receive the FAIL yet,
   we use the CLUSTERMSG_FLAG0_FORCEACK to make sure they will vote. This is equivalent
   to broadcasting a FAIL message to all primaries before we broadcast the failover
   auth request (but cheaper).

The race between FAIL (broadcast by A) and AUTH REQUEST (broadcast by R) is illustrated
in the following sequence diagram:

A      R        B      C
|      |        |      |
| FAIL |        |      |
|----->| AUTH R.|      |
|      |------->|      |
| FAIL |        |      |
|-------------->|      |
|      | AUTH R.|      |
|      |-------------->|
| FAIL |        |      |
|--------------------->|

Details

This is the how the failover is initiated, with new steps marked with (new):

1. A majority of primaries have marked another primary as PFAIL.
2. Some nodes counts failure reports and marks the failing primary as FAIL. The node
   that detects FAIL broadcasts it to all nodes in the cluster.
3. When a replica receives FAIL (or detects FAIL itself by counting PFAIL reports) it
   schedules a failover:
   a. It sets a timeout (500ms + random 0-500ms).
   b. It broadcasts pong to the other replicas in the same shard.
   c. (new) The pong (actually the clusterMsg header) has a new flag CLUSTER_NODE_MY_PRIMARY_FAIL.
   When the replicas broadcast pong to each other here, this flag is set.
4. (new) When the following conditions are met, skip the remaining delay and start
   the failover using AUTH REQUEST with the FORCE ACK flag set, that is if
   a. a PONG is received from every other replica in the same shard (broadcast within
   the shard) and
   b. all replicas have marked that its primary is FAIL in their last message (the new
   CLUSTER_NODE_MY_PRIMARY_FAIL flag is set) and
   c. this is the best replica (rank = 0) and
   d. my replication offset != 0.
5. When the delay has passed and no other replica has initiated failover, then initiate failover.

Notes:

• With 3(c), we don't need to wait for FAIL to propagate to all voting primaries.
  At this point, a FAIL has already been broadcast by some node, but there is a race
  so our auth request may arrive to some node before the FAIL. Using the FORCE ACK flag
  ensures the primaries will vote for us. (It is equivalent to broacasting another FAIL
  just before broadcasting auth request.)
• 4(b) ensures that we have received the replication offset from all other replicas and
  that it's up to date. If a replica says that it's primary is failing, it also means that
  the replication from the primary to that replica has stopped.
• 4(c) is to avoid a special bad case. It can happen that not all replicas know about each
  other. In this case, two replicas can think they are both the best replica and start the
  failover at the same time. This can already happen without this PR. When it happens, it
  usually means that a new replica has just joined and it has no data (offset = 0) and if
  it wins the election, there is a problem of dataloss (discussed and partially mitigated
  for the replica migration case in Fix data loss when replica do a failover with a old history repl offset #885). To avoid this case, skip this fast failover path
  if the replica has offset = 0.

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 71.48%. Comparing base (e9edba4) to head (2c0b049).

Additional details and impacted files

@@             Coverage Diff              @@
##           unstable    #2227      +/-   ##
============================================
+ Coverage     71.36%   71.48%   +0.11%     
============================================
  Files           123      123              
  Lines         67094    67121      +27     
============================================
+ Hits          47884    47980      +96     
+ Misses        19210    19141      -69     

Files with missing lines	Coverage Δ
src/cluster_legacy.c	86.93% <100.00%> (+0.21%)	⬆️

... and 13 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[zuiderkwast]

Good job!

I have some ideas. See comments.

Some test case would be nice to see, to check that we can do a failover faster than the delay that we skip. For example with node timeout 2000, check that we can do failover in 2500 ms? I guess a test like this can be flaky but at least try it to see if the feature works, would be good.

I'm a little confused about CLUSTER_TODO_HANDLE_FASTER_FAILOVER. Why do we need it? Why is it different to CLUSTER_TODO_HANDLE_FAILOVER, when can we use it, when can we not use it, etc. It is not clear to me.

IIUC, it's not really needed for the the feature skip delay if clusterAllReplicasThinkPrimaryIsFail() returns true and my rank is 0, is it? If it's a separate optimization, then I guess we skip the FASTER_FAILOVER idea in this PR, to keep it simpler. WDYT?

[enjoy-binbin]

Some test case would be nice to see, to check that we can do a failover faster than the delay that we skip. For example with node timeout 2000, check that we can do failover in 2500 ms? I guess a test like this can be flaky but at least try it to see if the feature works, would be good.

Yes, i will add the test later, i test it locally yesterday and it work.

I'm a little confused about CLUSTER_TODO_HANDLE_FASTER_FAILOVER. Why do we need it? Why is it different to CLUSTER_TODO_HANDLE_FAILOVER, when can we use it, when can we not use it, etc. It is not clear to me.

Yes, it is indeed a bit confusing, sorry i did not mention it. I want to put CLUSTER_TODO_HANDLE_FASTER_FAILOVER after other flags, like after CLUSTER_TODO_UPDATE_STATE, so that we will first print CLUSTER IS DOWN and then try to trigger the failover. That is now we will trigger failover faster, in the past, we will update the cluster state and then try to failover in ClusterCron.

IIUC, it's not really needed for the the feature skip delay if clusterAllReplicasThinkPrimaryIsFail() returns true and my rank is 0, is it? If it's a separate optimization, then I guess we skip the FASTER_FAILOVER idea in this PR, to keep it simpler. WDYT?

Sure, i can remove FASTER_FAILOVER and then only put it in #2209

[zuiderkwast]

@hpatro I have discussed with @enjoy-binbin and updated the top comment with a more thorough reasoning about the possible scenarios. PTAL.

The special case that a new replica joins and doesn't know about the other replicas is a serious problem, but it's not new. It's not solved here but we can avoid making it worse. (I hope we can solve it in another PR, to make sure a replica is made aware of the other replicas in the shard ASAP, for example it starts replicating.)

[enjoy-binbin]

Let's also include a check that myself's replication offset != 0 here?

It makes sure that if myself is a new replica just connected that don't yet know about other replicas, we should not try to do this fast failover.

The special case that a new replica joins and doesn't know about the other replicas is a serious problem, but it's not new. It's not solved here but we can avoid making it worse. (I hope we can solve it in another PR, to make sure a replica is made aware of the other replicas in the shard ASAP, for example it starts replicating.)

Ok, i pushed the new commit, sorry for the late response, i somehow forgot it and thought we are waitting response from @hpatro

[zuiderkwast]

@madolson This is based on your idea to keep track of other replicas' view of the primay's fail status. Great if you can take a look.

[hpatro]

Haven't taken a complete look.

@enjoy-binbin Could you please share if we still continue the primary rank logic and not bypass that with fast failover logic?

[enjoy-binbin]

Could you please share if we still continue the primary rank logic and not bypass that with fast failover logic?

What do you means? sorry i don't quite get the question.

[zuiderkwast]

I would like to have this in 9.0. I hope others will agree. Maybe we can merge it after RC1.

[madolson]

I would like to pause this for 9.0. We have seen a lot of cluster instability in AWS after merging 8.1 (much worse than in any past release), and I don't think we have the testing in place for all of these optimizations to make sure we aren't ending up in odd states.