# US 11,445,019 B2 · Methods, systems, and media for providing distributed database access during a network split

> Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt

- Number: US 11,445,019 B2
- Document ID: US11445019B2
- Status: Granted
- Kind: B2
- Application number: 16/826,900
- Filed: 2020-03-23
- Published: 2020-11-12
- Granted: 2022-09-13
- Earliest claimed: 2015-06-15
- Assignees: Redis Ltd.
- Applicants: Redis Ltd
- Inventors: Yossi Gottlieb; Yoav Steinberg; Yiftach Shoolman
- Web page: https://patentgrep.com/patent/US11445019B2

## CPC

- H04L 67/1095: Replication or mirroring of data, e.g. scheduling or transport for data synchronisation between network nodes
- G06F 16/2379: {Updates performed during online database operations; commit processing}
- G06F 16/27: Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor
- G06F 16/278: {Data partitioning, e.g. horizontal or vertical partitioning}
- H04L 67/02: based on web technology, e.g. hypertext transfer protocol [HTTP]
- H04L 67/10: in which an application is distributed across nodes in the network (software deployment G06F8/60; multiprogramming arrangements G06F9/46)
- H04L 67/56: Provisioning of proxy services (store-and-forward switching systems in data switching networks H04L12/54)
- H04L 69/40: for recovering from a failure of a protocol instance or entity, e.g. service redundancy protocols, protocol state redundancy or protocol service redirection (management of faults, events, alarms or notifications in data switching networks H04L41/06)
- H04L 41/30: {Decision processes by autonomous network management units using voting and bidding}
- H04L: TRANSMISSION OF DIGITAL INFORMATION, e.g. TELEGRAPHIC COMMUNICATION (arrangements common to telegraphic and telephonic communication H04M)
- G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N)

## Abstract

Methods, systems, and media for providing distributed database access during a network split. In some embodiments, the method comprises: receiving, at a node of N nodes, configuration information of the database which indicates which of the N nodes stores replicated shards of the database including master shards and corresponding slave shards, N being an odd number greater than one; identifying, after a network split, a subset of the N nodes which are capable of communicating with each other; receiving, by a proxy, a request to perform a write operation to an object in the database; determining that a second node included in the subset of the N nodes stores a master shard including the object; determining that the subset includes at least (N+1)/2 nodes; and causing the request to perform the write operation to be routed to the second node.

## Claims

### Claim 1

A method for providing access to information stored in a replicated distributed database, the method comprising:
receiving, using a hardware processor of a first node of N nodes, a first request to perform a first write operation to an object in the database, wherein N is an odd integer number greater than one;
determining that a second node of the N nodes stores a master shard that includes the object;
determining that the second node is included in a subset of the N nodes;
determining that the subset includes at least (N+1)/2 nodes; and
in response to determining that the subset includes at least (N+1)/2 nodes, causing the first request to perform the first write operation to be routed to the second node.

### Claim 2 (depends on 1)

The method of claim 1, further comprising:
electing, with the N nodes, a node as a cluster manager node before a network split occurs; and
generating, by the node elected as the cluster manager node, configuration information of the database.

### Claim 3 (depends on 2)

The method of claim 2, further comprising:
determining, with the subset of the N nodes after the network split has occurred, that the node elected as the cluster manager node is not among the subset of N nodes; and
in response to determining that the that the node elected as the cluster manager node is not among the subset of the N nodes, electing a new node from the subset of the N nodes as a new cluster manager node.

### Claim 4 (depends on 3)

The method of claim 3, further comprising:
determining, by the new node elected as the new cluster manager node, that a slave shard is stored by one of the subset of the N nodes and that the master shard to which the slave shard corresponds is not stored by one of the subset of the N nodes; and
updating the configuration information to indicate that the slave shard is to be re-designated as a new master shard, and that the master shard to which the slave shard corresponds is to be re-designated as a new slave shard.

### Claim 5 (depends on 1)

The method of claim 1, further comprising:
receiving, at a third node of the N nodes, configuration information;
identifying, after a network split has occurred, a second subset of the N nodes, including the third node, which are capable of communicating with each other;
receiving a second request to perform a second write operation to a second object in the database;
determining that the second subset includes less than (N+1)/2 nodes; and
in response to determining that the subset includes less than (N+1)/2 nodes, inhibiting the second write operation from being performed.

### Claim 6 (depends on 5)

The method of claim 5, further comprising:
receiving a request to perform a read operation to a third object in the database;
determining which of the N nodes stores a master shard that includes the third object;
determining which of the N nodes stores a slave shard that includes the third object;
determining whether both the master shard storing the third object and the slave shard storing the third object are stored by nodes in the second subset of the N nodes; and
in response to determining that both the second master shard and the slave shard are stored by nodes in the second subset of the N nodes, causing the read operation to be routed to the node storing the master shard that includes the third object or the node storing the slave shard that includes the third object.

### Claim 7 (depends on 1)

The method of claim 1, further comprising:
receiving, at the first node during a period of time subsequent to a network split occurring, a message from each node in the subset of the N nodes, wherein the message indicates at least that the first node is in communication with the node that sent the message; and
determining, at the first node, that a message has not been received from each node not in the subset of the N nodes during the period of time subsequent to the network split.

### Claim 8 (depends on 1)

The method of claim 1, wherein the first request to perform the first write operation to the object in the database includes a key corresponding to the object.

### Claim 9 (depends on 1)

The method of claim 1, wherein the database is a non-relational database.

### Claim 10 (depends on 9)

The method of claim 9, wherein the database is an in-memory non-relational database.

### Claim 11

A system for providing access to information stored in a replicated distributed database, the system comprising:
at least one hardware processor that is collectively programmed to:
receive, at a first node of N nodes, a first request to perform a first write operation to an object in the database, wherein N is an odd integer number greater than one;
determine that a second node of the N nodes stores a master shard that includes the object;
determine that the second node is included in a subset of the N nodes;
determine that the subset includes at least (N+1)/2 nodes; and
in response to determining that the subset includes at least (N+1)/2 nodes, cause the first request to perform the first write operation to be routed to the second node.

### Claim 12 (depends on 11)

The system of claim 11, wherein the at least one hardware processor is further collectively programmed to:
elect, with the N nodes, a node as a cluster manager node before a network split occurs; and
generate, by the node elected as the cluster manager node, configuration information of the database.

### Claim 13 (depends on 12)

The system of claim 12, wherein the at least one hardware processor is further collectively programmed to:
determine, with the subset of the N nodes after the network split has occurred, that the node elected as the cluster manager node is not among the subset of N nodes; and
in response to determining that the that the node elected as the cluster manager node is not among the subset of the N nodes, elect a new node from the subset of the N nodes as a new cluster manager node.

### Claim 14 (depends on 13)

The system of claim 13, wherein the at least one hardware processor is further collectively programmed to:
determine, by the new node elected as the new cluster manager node, that a slave shard is stored by one of the subset of the N nodes and that the master shard to which the slave shard corresponds is not stored by one of the subset of the N nodes; and
update the configuration information to indicate that the slave shard is to be re-designated as a new master shard, and that the master shard to which the slave shard corresponds is to be re-designated as a new slave shard.

### Claim 15 (depends on 11)

The system of claim 11, wherein the at least one hardware processor is further collectively programmed to:
receive, at a third node of the N nodes, configuration information;
identify, after a network split has occurred, a second subset of the N nodes, including the third node, which are capable of communicating with each other;
receive a second request to perform a second write operation to a second object in the database;
determine that the second subset includes less than (N+1)/2 nodes; and
in response to determining that the subset includes less than (N+1)/2 nodes, inhibit the second write operation from being performed.

### Claim 16 (depends on 15)

The system of claim 15, wherein the at least one hardware processor is further collectively programmed to:
receive a request to perform a read operation to a third object in the database;
determine which of the N nodes stores a master shard that includes the third object;
determine which of the N nodes stores a slave shard that includes the third object;
determine whether both the master shard storing the third object and the slave shard storing the third object are stored by nodes in the second subset of the N nodes; and
in response to determining that both the second master shard and the slave shard are stored by nodes in the second subset of the N nodes, cause the read operation to be routed to the node storing the master shard that includes the third object or the node storing the slave shard that includes the third object.

### Claim 17 (depends on 11)

The system of claim 11, wherein the at least one hardware processor is further collectively programmed to:
receive, at the first node during a period of time subsequent to a network split occurring, a message from each node in the subset of the N nodes, wherein the message indicates at least that the first node is in communication with the node that sent the message; and
determine, at the first node, that a message has not been received from each node not in the subset of the N nodes during the period of time subsequent to the network split.

### Claim 18 (depends on 11)

The system of claim 11, wherein the first request to perform the first write operation to the object in the database includes a key corresponding to the object.

### Claim 19 (depends on 11)

The system of claim 11, wherein the database is a non-relational database.

### Claim 20 (depends on 19)

The system of claim 19, wherein the database is an in-memory non-relational database.

### Claim 21

A non-transitory computer-readable medium containing computer executable instructions that, when executed by a processor, cause the processor to perform a method for providing access to information stored in a replicated distributed database, the method comprising:
receiving, at a first node of N nodes, a first request to perform a first write operation to an object in the database, wherein N is an odd integer number greater than one;
determining that a second node of the N nodes stores a master shard that includes the object;
determining that the second node is included in a subset of the N nodes;
determining that the subset includes at least (N+1)/2 nodes; and
in response to determining that the subset includes at least (N+1)/2 nodes, causing the first request to perform the first write operation to be routed to the second node.

### Claim 22 (depends on 21)

The non-transitory computer-readable medium of claim 21, wherein the method further comprises:
electing, with the N nodes, a node as a cluster manager node before a network split occurs; and
generating, by the node elected as the cluster manager node, configuration information of the database.

### Claim 23 (depends on 21)

The non-transitory computer-readable medium of claim 21, wherein the method further comprises:
determining, with the subset of the N nodes after the network split has occurred, that the node elected as the cluster manager node is not among the subset of N nodes; and
in response to determining that the that the node elected as the cluster manager node is not among the subset of the N nodes, electing a new node from the subset of the N nodes as a new cluster manager node.

### Claim 24 (depends on 23)

The non-transitory computer-readable medium of claim 23, wherein the method further comprises:
determining, by the new node elected as the new cluster manager node, that a slave shard is stored by one of the subset of the N nodes and that the master shard to which the slave shard corresponds is not stored by one of the subset of the N nodes; and
updating the configuration information to indicate that the slave shard is to be re-designated as a new master shard, and that the master shard to which the slave shard corresponds is to be re-designated as a new slave shard.

### Claim 25 (depends on 21)

The non-transitory computer-readable medium of claim 21, wherein the method further comprises:
receiving, at a third node of the N nodes, configuration information;
identifying, after a network split has occurred, a second subset of the N nodes, including the third node, which are capable of communicating with each other;
receiving a second request to perform a second write operation to a second object in the database;
determining that the second subset includes less than (N+1)/2 nodes; and
in response to determining that the subset includes less than (N+1)/2 nodes, inhibiting the second write operation from being performed.

### Claim 26 (depends on 25)

The non-transitory computer-readable medium of claim 25, wherein the method further comprises:
receiving a request to perform a read operation to a third object in the database;
determining which of the N nodes stores a master shard that includes the third object;
determining which of the N nodes stores a slave shard that includes the third object;
determining whether both the master shard storing the third object and the slave shard storing the third object are stored by nodes in the second subset of the N nodes; and
in response to determining that both the second master shard and the slave shard are stored by nodes in the second subset of the N nodes, causing the read operation to be routed to the node storing the master shard that includes the third object or the node storing the slave shard that includes the third object.

### Claim 27 (depends on 21)

The non-transitory computer-readable medium of claim 21, wherein the method further comprises:
receiving, at the first node during a period of time subsequent to a network split occurring, a message from each node in the subset of the N nodes, wherein the message indicates at least that the first node is in communication with the node that sent the message; and
determining, at the first node, that a message has not been received from each node not in the subset of the N nodes during the period of time subsequent to the network split.

### Claim 28 (depends on 21)

The non-transitory computer-readable medium of claim 21, wherein the first request to perform the first write operation to the object in the database includes a key corresponding to the object.

### Claim 29 (depends on 21)

The non-transitory computer-readable medium of claim 21, wherein the database is a non-relational database.

### Claim 30 (depends on 29)

The non-transitory computer-readable medium of claim 29, wherein the database is an in-memory non-relational database.