System and method for determining consensus within a distributed database

• US 10,614,098 B2
• Filed: 05/25/2017
• Issued: 04/07/2020
• Est. Priority Date: 12/23/2010
• Status: Active Grant

• Current Assignee: MongoDB, Inc.
• Original Assignee: MongoDB, Inc.
• Inventors: Horowitz, Eliot, Schwerin, Andrew Michalski, Zhou, Siyuan, Milkie, Eric Andrew
• Primary Examiner(s): Duncan, Marc
• Application Number:
  US15/605,372
• Publication Number:
  US 20170270176A1
• Time in Patent Office: 1,048 Days
• CPC Class Codes: G06F 11/0709 in a distributed system consisting of a plurality of standalone computer nodes, e.g. clusters, client-server systems
• CPC Class Codes: G06F 11/0757 by exceeding a time limit, i.e. time-out, e.g. watchdogs
• CPC Class Codes: G06F 11/0793 Remedial or corrective actions recovery from an exception in an instruction pipeline G06F9/3861; by retry G06F11/1402; for recovering from a failure of a protocol instance or entity H04L69/40
• CPC Class Codes: G06F 11/14 Error detection or correction of the data by redundancy in operation G06F11/16 takes precedence
• CPC Class Codes: G06F 11/1458 Management of the backup or restore process
• CPC Class Codes: G06F 11/1471 involving logging of persistent data for recovery
• CPC Class Codes: G06F 11/1658 Data re-synchronization of a redundant component, or initial sync of replacement, additional or spare unit
• CPC Class Codes: G06F 11/2028 eliminating a faulty processor or activating a spare
• CPC Class Codes: G06F 11/2048 where the redundant components share neither address space nor persistent storage
• CPC Class Codes: G06F 11/2094 Redundant storage or storage space G06F11/2056 takes precedence
• CPC Class Codes: G06F 11/2097 maintaining the standby controller/processing unit updated initialisation or re-synchronisation thereof G06F11/1658 and subgroups
• CPC Class Codes: G06F 16/2365 Ensuring data consistency and integrity
• CPC Class Codes: G06F 16/273 Asynchronous replication or reconciliation
• CPC Class Codes: G06F 2201/84 Using snapshots, i.e. a logical point-in-time copy of the data

Abstract

A system and method for determining consensus within a distributed database are provided. According to one aspect, a protocol is provided that reduces or eliminates heartbeat communication between nodes of a replica set. Nodes may communicate liveness information using existing database commands and metadata associated with the database commands. According to another aspect, improved systems and methods are provided for detection of node failures and election of a new primary node.

Claims

1. A computer-implemented method for maintaining consensus in a distributed database system comprising a plurality of nodes, the method comprising:

establishing a primary node having a primary node role within the distributed database system and the primary node hosting a database, wherein the distributed database system provides one or more responses to database requests from one or more client computer systems;

establishing a plurality of secondary nodes having a secondary node role, wherein each secondary node is associated with the primary node and hosts a replica of the database hosted by the primary node;

replicating, from the primary node, executed write operations to secondary nodes associated with the primary node;

communicating, via a normal database operation, heartbeat information as metadata within a write operation function; and

automatically recovering the primary node role in the distributed database system in response to a detected failure of the primary node by an absence of the heartbeat information.

2. The method according to claim 1, further comprising communicating the heartbeat information to one or more secondary nodes via chain of secondary nodes.

3. The method according to claim 1, further comprising identifying, within a message associated with an election, a new term identifier.

4. The method according to claim 3, further comprising storing the new term identifier in an operation log.

5. The method of claim 4, further comprising associating the new term identifier with at least one write operation of the operation log.

6. The method of claim 5, further comprising preventing replication of a write operation responsive to determining that a term identifier associated with the write operation does not match the new term identifier.

7. The method according to claim 3, further comprising an act of storing, by one or more of the plurality of secondary nodes, the new term identifier in durable storage.

8. The method according to claim 7, further comprising prohibiting the plurality of secondary nodes from voting more than a predetermined number of times in an election identified by the new term identifier.

9. The method according to claim 3, further comprising incrementing a value of the new term identifier for each election attempt when a secondary node transitions to a candidate status.

10. The method according to claim 3, further comprising communicating, via the normal database operation, a value of a current term identifier to a downstream node.

11. The method according to claim 3, further comprising determining, by the primary node, a last operation in a operation log of the primary node, and determining that the last operation was written by a majority of replica set members.

12. The method according to claim 1, further comprising:

establishing one of the plurality of secondary nodes as a new primary node having the primary node role;

deleting replicated write operations received by the new primary node from the original primary node; and

establishing a point of consistency within the database between the new primary node and remaining secondary nodes as a new current state of the database.

13. The method of claim 12, further comprising communicating a message to the remaining secondary nodes announcing that the one of the plurality of secondary nodes is the new primary node.

14. The method of claim 1 further comprising communicating, during the normal database operation, database state information as metadata within the write operation function.

15. The method of claim 1, further comprising:

communicating a no-operation command from the plurality of secondary nodes to the primary node, wherein the no-operation command is a valid command that causes the primary node to perform the valid command as a non-operation and treats the receipt of the no-operation command as a heartbeat signal; and

communicating, in a response to the no-op command, database state information as metadata to the plurality of secondary nodes.

16. A system for maintaining consensus in a distributed database system comprising a plurality of nodes, the system comprising:

at least two processors configured to;

establish a primary node having a primary node role within the distributed database system and the primary node hosting a database, wherein the distributed database system provides one or more responses to one or more database requests from one or more client computer systems;

establish a plurality of secondary nodes having a secondary node role, wherein each secondary node is associated with the primary node and hosts a replica of the database hosted by the primary node;

replicate, from the primary node, executed write operations to the plurality of secondary nodes;

communicate, via a normal database operation, heartbeat information to the plurality of secondary nodes as metadata within a write operation function; and

automatically recover the primary node role in the distributed database system in response to a detected failure of the primary node by an absence of the heartbeat information.

17. The system of claim 16, wherein the at least two processors are is further configured to communicate the heartbeat information to one or more secondary nodes via a chain of secondary nodes.

18. The system of claim 16, wherein the at least one of the at least two processors is further configured to identify, within a message associated with an election, a new term identifier.

19. The system of claim 18, wherein the at least one processor is further configured to store the new term identifier in an operation log.

20. The system of claim 19, wherein the at least one processor is further configured to associate the new term identifier with at least one write operation of the operation log.

21. The system of claim 20, wherein the at least one processor is further configured to prevent replication of a write operation responsive to determining that a term identifier associated with the write operation does not match the new term identifier.

22. The system of claim 18, wherein one or more of the plurality of secondary nodes are configured to store the new term identifier in durable storage.

23. The system of claim 22, wherein the at least two processors are further configured to prohibit the plurality of secondary nodes from voting more than a predetermined number of times in an election identified by the new term identifier.

24. The system of claim 18, wherein the at least one processor is further configured to increment a value of the new term identifier for each election attempt when a secondary node transitions to a candidate status.

25. The system of claim 18, wherein the at least one processor is further configured to communicate, via the normal database operation, a value of a current term identifier to a downstream node.

26. The system of claim 18, wherein the at least one processor is further configured to:

determine a last operation in an operation log of the primary node, and

determine that the last operation was written by a majority of replica set members.

27. The system of claim 16, wherein the at least two processors are further configured to:

establish one of the plurality of secondary nodes as a new primary node having the primary node role;

delete replicated write operations received by the new primary node from the original primary node; and

establish a point of consistency within the database between the new primary node and remaining secondary nodes as a new current state of the database.

28. The system of claim 27, wherein the at least two processors are further configured to communicate a message from the one of the plurality of secondary nodes to the remaining secondary nodes announcing that the one of the plurality of secondary nodes is the new primary node.

29. The system of claim 16, wherein the at least two processors are further configured to communicate, during the normal database operation, primary state information as metadata within the write operation function.

30. The system of claim 16, wherein the at least two processors are further configured to:

communicate a no-operation command from the plurality of secondary nodes to the primary node, wherein the no-operation command is a valid command that causes the primary node to perform the valid command as a non-operation and treats the receipt of the no-operation command as a heartbeat signal; and

communicate, in response to the no-operation command, database state information as metadata to the plurality of secondary nodes.