System for maintaining consistency across a decentralized database cluster and method therefor

• US 10,083,201 B2
• Filed: 09/22/2015
• Issued: 09/25/2018
• Est. Priority Date: 09/22/2015
• Status: Active Grant

• Current Assignee: Walmart Apollo, LLC (WalMart Inc.)
• Original Assignee: Walmart Apollo, LLC (WalMart Inc.)
• Inventors: Guney, Ergin, Zheng, Yan
• Primary Examiner(s): LEWIS, CHERYL RENEA
• Application Number:
  US14/861,672
• Publication Number:
  US 20170083562A1
• Time in Patent Office: 1,099 Days
• CPC Class Codes: G06F 16/2365 Ensuring data consistency and integrity

Abstract

A system and method for maintaining consistency across a decentralized database cluster is presented. A database cluster can contain a plurality of database servers, each hosting a copy of the same database. In a system without a centralized coordinating server, a message queue can be used to simultaneously provide updates to each database server in the database cluster. A When a database server receives an index update message, it adds message to a queue indicating it has received the message and watches an index update node. Once every server in the cluster has added a message to the queue, it is known that every database server has received the message. The update can be performed for each server and the changes committed simultaneously. In such a manner, all database servers in the cluster perform an update at approximately the same time. Other embodiments also are disclosed.

Claims

1. A system comprising:

a plurality of database servers each hosting a copy of data stored in a database, each database server of the plurality of database servers comprising one or more processing modules and one or more non-transitory storage modules storing computing instructions configured to run on the one or more processing modules and perform acts of;

receiving a stream of document update messages broadcast, via a messaging system, to the plurality of database servers, wherein a checkpoint marker message is embedded into at least one of the document update messages with an associated checkpoint ID; and

when a database server of the plurality of database servers is active;

adding a node representing the database server of the plurality of database servers to a list of active database servers stored on an external data store; and

receiving one of the document update messages comprising the checkpoint marker message with the associated checkpoint ID, wherein the checkpoint marker message causes the database server to pause further processing of other messages until after a roll call of the each database server on the list of active database servers is completed to indicate that the each database server on the list of active database servers has checked in with its own node under a checkpoint roll call node;

after receiving the one of the document update messages comprising the checkpoint marker message with the associated checkpoint ID, determining whether the checkpoint roll call node exists at the external data store for the associated checkpoint ID;

when the checkpoint roll call node is determined to exist at the external data store for the associated checkpoint ID, checking in to the external data store by adding a node representing the database server under the checkpoint roll call node;

when the checkpoint roll call node is determined to not exist at the external data store for the associated checkpoint ID, first creating the checkpoint roll call node and a checkpoint flag node at the external data store for the checkpoint ID, and then checking in to the external data store by adding the node representing the database server under the checkpoint roll call node;

after the node representing the database server is added under the checkpoint roll call node, determining whether the roll call is complete, wherein;

the roll call is complete when the each database server on the list of active database servers has processed the checkpoint marker message as indicated by an addition of their own node under the checkpoint roll call node at the external data store at which point the external data store deletes the checkpoint roll call node, nodes under the checkpoint roll call node, and the checkpoint flag node; and

when the roll call is not complete, monitor the checkpoint flag node at the external data store for the associated checkpoint ID until the each database server on the list of active database servers has processed the checkpoint marker message indicated by the addition of its own node under the checkpoint roll call node;

after the roll call is complete and after the checkpoint roll call node and the checkpoint flag node are deleted, receiving an update broadcasted to each database server on the list of active database servers so that the database server can update its data with the update during a common point in time with all other database servers on the list of active database servers;

updating the data on the database server during the common point in time in response to receiving the update; and

reactivating the database server to resume normal processing of the document update messages until the database server receives another one of the document update messages comprising another checkpoint message marker.

2. The system of claim 1 wherein:

determining whether the checkpoint roll call node exists at the external data store for the associated checkpoint ID further comprises;

when the checkpoint roll call node does not exist at the external data store, creating the checkpoint roll call node, wherein multiple servers creating the checkpoint roll call node at a same time can be avoided with a use of an exclusive lock mechanism.

3. The system of claim 2 wherein the computing instructions are further configured to perform acts of:

placing a watch on the checkpoint flag node for the associated checkpoint ID so that the database server receives a push notification when the checkpoint flag node is deleted.

4. The system of claim 3, wherein

determining when the roll call is complete further comprises:

comparing the nodes belonging to the checkpoint roll call node to the nodes belonging to the list of active database servers; and

if there are no nodes missing from the checkpoint roll call node, then updating the checkpoint flag node.

5. The system of claim 4 further comprising:

before receiving the update, receiving a notification that the checkpoint flag node has been updated.

6. The system of claim 1 wherein the database contains a shard of a larger database.

7. The system of claim 1 wherein the computing instructions are further configured to perform an act of:

removing the database server from the list of active database servers when the database server becomes inactive.

8. The system of claim 1 wherein:

the database contains a shard of a larger database;

determining if the checkpoint roll call node exists for the associated checkpoint ID further comprises;

if the checkpoint roll call node does not exist, creating the checkpoint roll call node; and

creating the checkpoint flag node;

wherein the computing instructions are further configured to perform an act of;

placing a watch on the checkpoint flag node;

wherein determining if the roll call is complete further comprises;

comparing the nodes belonging to the checkpoint roll call node to the nodes belonging to the list of active database servers; and

if there are no nodes missing from the checkpoint roll call node, then updating the checkpoint flag node;

wherein determining if the roll call is complete further comprises;

receiving a notification that the checkpoint flag node has been updated; and

wherein the computing instructions are further configured to perform acts of;

pausing execution of the computing instructions after receiving the checkpoint marker message with the associated checkpoint ID;

resuming execution of the computing instructions after committing the update to the database server.

9. The system of claim 1, wherein

the stream of the document update messages comprises a normal document update message, wherein pending index updates are committed when a roll call process is complete.

10. A method comprising:

receiving, by a computer system with one or more processors, a stream of document update messages broadcast, via a messaging system, each database server of a plurality of database servers hosting a copy of data stored in a database, wherein a checkpoint marker message is embedded into at least one of the document update messages with an associated checkpoint ID;

when a database server of the plurality of database servers is active;

adding, by the computer system, a node representing the database server of the plurality of database servers, to a list of active database servers stored on an external data store; and

receiving, with the computer system, one of the document update messages comprising the checkpoint marker message with the associated checkpoint ID, wherein the checkpoint marker message causes the database server to pause further processing of other messages until after a roll call of the each database server on the list of active database servers is completed to indicate that the each database server on the list of active database servers has checked in with its own node under a checkpoint roll call node;

after receiving the one of the document update messages comprising the checkpoint marker message with the associated checkpoint ID, determining, with a checkpoint roll call node detection system, whether the checkpoint roll call node exists at the external data store for the associated checkpoint ID;

when the checkpoint roll call node is determined to exist at the external data store for the associated checkpoint ID, checking in to the external data store by adding a node representing the database server under the checkpoint roll call node;

when the checkpoint roll call node is determined to not exist at the external data store for the associated checkpoint ID, first creating the checkpoint roll call node and a checkpoint flag node at the external data store for the associated checkpoint ID, and then checking in to the external data store by adding the node representing the database server under the checkpoint roll call node;

after the node representing the database server is added under the checkpoint roll call node, determining, with a roll call node completion detection system, whether the roll call is complete, wherein;

the roll call is complete when the each database server on the list of active database servers has processed the checkpoint marker message as indicated by an addition of their own node under the checkpoint roll call node at the external data store at which point the external data store deletes the checkpoint roll call node, nodes under the checkpoint roll call node, and the checkpoint flag node; and

when the roll call is not complete, monitor the checkpoint flag node at the external data store for the associated checkpoint ID until the each database server on the list of active database servers has processed the checkpoint marker message indicated by the addition of its own node under the checkpoint roll call mode;

after the roll call is complete and after the checkpoint roll call node and the checkpoint flag node are deleted, receiving an update broadcasted to the each database server on the list of active database servers so that the database server can update its data with the update during a common point in time with all other database servers on the list of active database servers;

updating the data on the database server during the common point in time in response to receiving the update; and

reactivating the database server to resume normal processing of the document update messages until the database server receives another one of the document update messages comprising another checkpoint message marker.

11. The method of claim 10 wherein:

determining, with the roll call node completion detection system, whether the checkpoint roll call node exists at the external data store for the associated checkpoint ID further comprises;

when the checkpoint roll call node does not exist at the external data store, creating the checkpoint roll call node, wherein multiple servers creating a checkpoint roll call node at a same time can be avoided with a use of an exclusive lock mechanism.

12. The method of claim 11 further comprising:

placing, by a checkpoint update flag node system, a watch on the checkpoint flag node for the associated checkpoint ID so that the database server receives a push notification when the checkpoint flag node is deleted.

13. The method of claim 12 wherein:

determining, with the roll call node completion detection system, when the roll call is complete further comprises;

comparing the nodes belonging to the checkpoint roll call node to the nodes belonging to the list of active database servers, wherein the checkpoint roll call nodes are created on the external data store; and

if there are no nodes missing from the checkpoint roll call node, then updating the checkpoint flag node.

14. The method of claim 13 further comprising:

deleting, with the roll call node completion detection system, the checkpoint roll call node after updating the checkpoint flag node.

15. The method of claim 13 wherein:

determining, with the roll call node completion detection system, if the roll call is complete further comprises;

receiving a notification that the checkpoint flag node has been updated.

16. The method of claim 10 wherein the database contains a shard of a larger database.

17. The method of claim 10 further comprising:

removing, with the computer system, the database server from the list of active database servers when the database server becomes inactive.

18. The method of claim 10 further comprising:

removing the database server from the list of active database servers if a load on the database server becomes greater than a first threshold; and

adding the database server to the list of active database servers if a load on the database server becomes less than a second threshold.

19. The method of claim 10 wherein:

the database contains a shard of a larger database;

determining if the checkpoint roll call node exists for the checkpoint ID further comprises;

if the checkpoint roll call node does not exist, creating a checkpoint roll call node; and

creating a checkpoint flag node;

computing instructions are configured to perform an act of;

placing a watch on the checkpoint flag node;

determining if the roll call is complete further comprises;

comparing nodes belonging to the checkpoint roll call node to nodes belonging to the list of active database servers; and

if there are no nodes missing from the checkpoint roll call node, then updating the checkpoint flag node;

determining if the roll call is complete further comprises;

receiving a notification that the checkpoint flag node has been updated; and

the computing instructions are further configured to perform acts of;

pausing execution of the computing instructions after receiving the checkpoint marker message with the associated checkpoint ID; and

resuming execution of the computing instructions after committing the update to the database server.

20. The method of claim 10, wherein

the stream of the document update messages further comprises a normal document update message, wherein pending index updates are committed when a roll call process is complete.