WaitForCommitAsync in write operations

[potrusil-osi]

I'm trying to implement my own "reliable queue" on top of your implementation of the Raft protocol. But I don't fully understand how PersistentState.WaitForCommitAsync should be used when implementing write operations. My main confusion is the timeout parameter. If I set it to something like 2 seconds (which is the value in the RaftNode example), what is supposed to happen if the change cannot be committed during those 2 seconds, but it is committed right after WaitForCommitAsync returns false? (Actually, the same can happen with the token parameter.) I'd think that the inability to commit within a reasonable time should cancel the whole transaction, but in reality there is no connection between the operation side of things and the heartbeat (which commits entries). That means the write operation can fail, but it is eventually committed. How should I reason about this?

My second question is about how PersistentState.WaitForCommitAsync is used in the RaftNode project. There it waits on an index that is always one bigger than the last committed index:

            var commitIndex = GetLastIndex(true);
            await AppendAsync(new Int64LogEntry { Content = value, Term = Term }, token);
            await WaitForCommitAsync(commitIndex + 1L, timeout, token);

Is this only for the sake of simplicity? My understanding is that it would be better to wait on the index that is returned by AppendAsync the line above - that way the write operation, UpdateValueAsync, can be executed multiple time in parallel and this code would still work. Am I correct?

            var entryIndex = await AppendAsync(new Int64LogEntry { Content = value, Term = Term }, token);
            await WaitForCommitAsync(entryIndex, timeout, token);

[potrusil-osi]

After thinking this through a bit more I guess the described behavior of WaitForCommitAsync on timeout is OK and the solution is to implement the duplicates elimination (and expect clients to retry). Am I thinking about it right?

[sakno]

WaitForCommitAsync doesn't provide transactional behavior. It's low-level API. Transactional behavior is out of scope of this Raft implementation. It's responsibility of the developer. The method allows you to ensure that the log entry at the specified index is marked as committed by the majority of nodes. It means that the cluster accepts the log record in a consistent way and it cannot be removed from the log. Before that, the log entry has uncommitted state which means that the only local node has this record in the log but all other nodes not. Committing is a process of replication (distribution) of uncommitted log entries by the leader across followers.

If WaitForCommitAsync is timed out then there is no guarantees to the caller about committed state of the log entry. The entry can be rewritten if leader has changed accidentally. You have two ways in this case:

• Optimistic approach - attempt to add log entry again with the updated term and wait for commit. Of course you need to analyze the semantics of the log entry and ensure that this action will not overwrite already committed data
• Pessimistic approach - inform the caller that the write operation is timed out and ask to retry again

[potrusil-osi]

I don't fully understand the optimistic approach. Is the updated term beneficial because the followers might accept the current node's leadership this way?

[sakno]

Updated term is needed because term of the next added log entry should not be less than the term of the log entry added later. Due to concurrency, there are few log entries may be added since the last attempt to commit the entry.

[potrusil-osi]

@sakno, I keep thinking about the two approaches you laid out here and I realized I still don't fully understand how WaitForCommitAsync should be used properly. It currently accepts only an index (and not a term), which means that it waits until a log entry with the provided index is committed, but it doesn't check the term of the committed entry. I can imagine that uncommitted log entries are overwritten by a new leader (if there was an election in the meantime) and the log entry with the index we are waiting for is replaced by a completely unrelated (but committed) entry (with a higher term). If that happens WaitForCommitAsync incorrectly returns with success. Is that really possible or am I missing something?

[sakno]

Yes, this is possible. WaitForCommitAsync is waiting for the index. It doesn't check the term. You need to add additional guard conditions to ensure that your log entry is committed. For instance:

PersistentState state = ...;
var term = state.Term;
IRaftLogEntry entry = ...; // create log entry with the current term which is equal to 'term' local variable
var index = await state.AppendAsync(entry);
await state.WaitForCommitAsync(index);
// ensure that the current term was not changed due to re-elections
if (term != state.Term)
{
  // check the term at index 'index'
}

The method for checking whether the log entry has the specified Term value already exist in the library but not available publicly: https://github.com/sakno/dotNext/blob/master/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/AuditTrail.cs#L27

I'll make this method as public in the next version.

[sakno]

The selected AppendAsync overload is for simplicity, yes. it's not accurate but acceptable because commit index increasing monotonically. This method is mostly used by Raft infrastructure for snapshot installation. For user code it's better to use more comprehensive overload:

ValueTask<long> AppendAsync<TEntryImpl>(ILogEntryProducer<TEntryImpl> entries, CancellationToken token = default(CancellationToken))

This method returns exactly what you want.

[potrusil-osi]

What is the main difference between the overload you proposed and the one used in RaftNode (ValueTask<long> AppendAsync<TEntry>(TEntry entry, CancellationToken token = default))? I looked at the source code and they both return state.LastIndex + 1L.

[sakno]

Let me clarify:

• ValueTask<long> AppendAsync<TEntryImpl>(TEntryImpl entry, CancellationToken token = default) - suitable for adding single entry point atomically. This method cannot be used for snapshot installation.
• ValueTask AppendAsync<TEntryImpl>(TEntryImpl entry, long startIndex) - suitable for adding single entry point atomically at the specified index. This method supports snapshot installation.
• ValueTask<long> AppendAsync<TEntryImpl>(ILogEntryProducer<TEntryImpl> entries, CancellationToken token = default) - suitable for adding multiple log entries atomically. Snapshot installation is not supported.

[sakno]

Snapshot installation typically is not needed by user code so the first and the last methods are more user-friendly.

[potrusil-osi]

I see what you mean. My original question was a bit different, though. If a write operation is invoked multiple times at the same time, then the corresponding method (for example UpdateValueAsync from RaftNode) will also be called multiple times concurrently. In such a case it would be better to use the index returned by one of the AppendAsync overloads and wait on it with WaitForCommitAsync, rather than what is currently in the RaftNode example where it waits on the committed log entry with the index that is one more than the last committed index (GetLastIndex(true) + 1L). In the latter case both concurrent operations could be waiting on the same index which would be wrong. Am I thinking about it right?

[sakno]

Yes, therefore it's better to rely on the returned index. AppendAsync operation is thread-safe as well as other public members of PersistentState class.