Quorum and transaction visibility in OrientDB 3.2

[timw]

Hi @tglman

We're working through testing 3.2 with a goal of replacing 3.1 in our production systems, and have run into a bit of a problem.

Background:

• We operate 3 node clusters of OrientDB nodes, which are accessed by multiple API service containers using stateless APIs, and the ROUND_ROBIN_CONNECT connection strategy.
• OrientDB transactions are predominantly insertions/updates of small numbers of records, and queries with projections across the record base.
• We have background tasks that run on the service containers for async processing, connected with AWS SQS queues.
• We are using writeQuorum of majority.
• We have readYourWrites enabled.
• We also use readQuorum of majority, but that's apparently not relevant.

What we've observed:

• With OrionDB 3.2 we observed integration tests failing, that are known to succeed reliably with 3.1.
• After investigation, we found the tests are failing because a record that was committed in one transaction is not visible in the background task processor when it receives the creation event notification via SQS. We have detailed logging and Time UUID based update timestamps on all records so this is definite.
• Examining the data remaining after the failing tests reveals that the transaction did commit successfully on all OrientDB nodes - i.e. the record is present and correct on all OrientDB servers, and can be queries from the API.

What we've learned from reviewing the code:

• It appears that readQuorum is not used in the query path - i.e. queries are all done at an effective readQuorum of 1 - it looks like read quorum is used in a limited number of distributed request paths, but not in the general client query requests.
• writeQuorum is evaluated in two primary places
  • In the first phase of the distributed commit, the writeQuorum is checked against the number of known/configured nodes - i.e. it checks that a quorum of known nodes is available.
  • In the distributed request layer (e.g. send2Nodes) the writeQuorum is checked against the available nodes - i.e. it checks that a quorum of the online nodes responds.
• In OrientDB 3.1, a write involving unique keys (which is 100% of our writes) escalates the quorum type from WRITE to WRITE_ALL_MASTERS at the distributed request layer.

Why we think OrientDB 3.1 passes these tests

• The writeQuorum of majority is upgraded to WRITE_ALL_MASTERS for all updates, which is effectively ALL when all nodes are online.
• All online servers commit the update as part of the distributed transaction.
• The queries are guaranteed to observe the updated record on all OrientDB servers that they can connect to.

The problem we observe with OrientDB 3.2

• The escalation to WRITE_ALL_MASTERS is now unnecessary and removed.
• Writes are committed at the distributed request layer at the majority quorum level.
• One of the 3 nodes may not have committed the transaction by the time the client is notified.
• There is then a race between the SQS notification and background task, and the commit completing on the third node.
• In a small, but reproducible over time, proportion of the time, the background task fails to observe the commit (apparently SQS is very fast).

The conundrum

• The obvious workaround is to upgrade the writeQuorum level to all, which will make the commit available on all nodes.
• The problem with this is that the all quorum level check applies at the distributed transaction commit level as well, which means that if an OrientDB node goes down we lose write availability in the entire system.
• Ultimately we're stuck between choosing a configuration with (non eventual) consistency or write availability

It's a fairly trivial task to modify the distributed request layer to wait for all nodes on commits (e.g. a readEveryonesWrites option), but we're wondering if there's anything we're missing in how OrientDB was intended to be used here, or pitfalls with that approach.

Some asides

• We found the readQuorum and writeQuorum configuration quite misleading (coming primarily from a background using sharded/replicated databases like Cassandra) and it took us a while to understand what OrientDB was actually meaning by those concepts. writeQuorum is overloaded in the OrientDB architecture to mean both a mechanism to avoid split brain, and a level of immediate consistency to apply, with the associated problems described above.
• The fact readQuorum isn't applied to reads isn't obvious at first glance - it's not even mentioned in the distributed architecture docs, but is in the default config file.
• The consistency of queries isn't mentioned in the docs, and while it "makes sense" that these don't obey readQuorum, it was still a bit surprising.
• The documentation for Conflict Resolution Policy was quite misleading. This functionality was removed in 3.x and replaced with an alternative conflict strategy system, and the docs are ambiguous about whether it applied to reads or writes - for a long time we thought it would reconcile reads base don these strategies.

[tglman]

Hi,

Your analysis is correct!

To reduce the scope to only the issue, what's happening is:

application1 do some write operations and then send a message to application2
applciation2 try to read the data written by application1 and it does not see it, because it may be reading from a node that did not participate to the quorum
In 3.1.x this was not the case, because OrientDB for some operations did require all the nodes to commit the transaction before confirming it, in 3.2.x all the transaction can be run as quorum.

So

• read your own write would not help in this case even if it was implemented correctly(which is not the case as today), because there are two different applications, so they start from two different contexts.
• read quorum (which is not implemented right now) it may help but this would add a big latency on all the query you execute, which is the reason why is not there anymore, we were trying to move to a different design.
• write quorum all is what works now, but it does reduce the availability is enough for one node to be offline to block all the writes, what is needed is something like "quorum all online" which is a bit strange but can be a viable solution in 3.2.x
• A final solution would be a "read your own write" solution, which we are planning to implement (even though not in 3.2.x), with an addition of the ability to transfer between session the "minimum state", so when sending a task to another application you can send also a state got from the session so the new session on the other side know the baseline of transactions and wait for them when is needed.

This sounds correct for you, and what would be the best solution for you?

[timw]

Those are all the solutions I've considered as well.

• I'd be interested in more details of why read your writes isn't implemented correctly (I can see it always elects the local node to be in the quorum, so it's not obvious to me why that would be broken).
• read quorum in the OrientDB architecture would indeed be hard to implement (which is why it "makes sense" that queries didn't do it). In a columnar database it's more straightforward to implement, but OrientDB would have to either compare projection outputs or capture underlying record+versions to reconcile, as well as the additional latencies involved, so a more involved solution.
• I think some variant of "all online" is my preferred solution, but it should be "all online and at least majority". My suggestion is an "readAllWrites" flag, or alternatively a rework of "readYourWrites" to a "readWrites: own/all" option. That way the write quorum level is about managing consistency in the case of partition, and the "consistency visibility" options are an additional modification.
• A client mediate solution that involves passing information from one session to another won't be viable for us. I don't like it in general, as it feels like a crutch, but also our entire architecture is built around stateless APIs, so data access (even across multiple processes) is unconnected as far as the code that talks to OrientDB is concerned.

I can try out the "readAllWrites" solution in our fork and see how the results look.

[tglman]

Hi,

• So for read your write issue, if you open a session with ROUND_ROBIN_REQUEST you can stumble on the same problem you are having now, with only one application with one session.

• Yes exactly you need to keep the information of the record version to reconcile the content.

• Actually thinking to an "all online" variant, what you do actually need is not that the quorum is all, but that the response of your write request is returned when all nodes have finished (successfully or failing) but keeping the quorum as success criteria, this will slow down a bit the writes but guarantee that when the result of a write is returned you, you can read the value from all the nodes (that are still available), this could be achievable modifying ODistributedTxResponseManagerImpl adding another method similar to waitForSynchronousResponses that wait the finish state, and then call it here right after the confirmation is sent, and only if successful (avoiding pointless delay in case of failure), this could be a good solution for your use case, even though I think is not a good default solution so if implemented should be optional.

• I understand your point on the stateless case, anyway the design of the distributed consistency in OrientDB has evolved in the last years, as today to detect and keep the nodes all at the same point of transactions we do use a "vector clock" from which we can detect if there are missing transactions on a node and what they are, the long term idea is to keep this vector clock in the session as well to describe what was the last minimum state viewed by that session, and use that vector clock state when switching between nodes to wait(or select node) for the minimum transaction to guarantee "readYourWrites" and as well a constant forward reading flow (it could happen when switching from a node to another the transaction that you see on one node you don't see on other nodes because of quorum based consensus). if this information is passed between session/process you can achieve the same consistency everywhere, with the advantage to need to wait only for quorum write, and only for transactions that your current session need to be aware.

Also as context before introducing the vector clock (in 3.0.x) you could experience irreconcilable inconsistencies for transactions quorum majority based, to keep the system consistent you needed a quorum all or you needed to be lucky enough that you had in fact a quorum all(all nodes online and replying on time) even though you would be running on quorum majority, this was somewhat a known problem and the merge strategies where introduced to try to solve this by delegating to the user what to do when a consistency failure happened.