# US 2014/0195486 A1 · Data recovery in multi-leader distributed systems > Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt - Number: US 2014/0195486 A1 - Document ID: US20140195486A1 - Status: Published application - Kind: A1 - Application number: 13/736,861 - Filed: 2013-01-08 - Published: 2014-07-10 - Earliest claimed: 2013-01-08 - Assignees: Facebook, Inc. - Applicants: Facebook, Inc. - Inventors: Sachin Kulkarni; Sanjeev Kumar; Harry Li; Laurent Demailly; Liat Atsmon Guz - Web page: https://patentgrep.com/patent/US20140195486A1 ## CPC - G06F 16/275: {Synchronous replication} - G06F 11/2082: {Data synchronisation} - G06F 11/2094: {Redundant storage or storage space (G06F11/2056 takes precedence)} - G06F 11/1662: {the resynchronized component or unit being a persistent storage device (re-synchronization of failed mirror storage G06F11/2082; rebuild or reconstruction of parity RAID storage G06F11/1008)} - G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N) ## Abstract Disclosed are a method and system for recovering a distributed system from a failure of a data storage unit. The distributed system includes a plurality of computer systems, each having a read-write computer and a data storage unit. Data is replicated from a particular data storage unit to other data storage units using publish-subscribe model. A read-write computer receives the replicated data, processes the data for any conflicts and stores it in the data storage unit. If a data storage unit fails, another data storage unit that has latest data corresponding to the failed data storage unit is determined and the latest data is replicated to other data storage units. Accordingly, the distributed system continues to have the data of the failed data storage unit. The failed data storage unit may be reconstructed using data from one of the other data storage units in the distributed system. ## Claims ### Claim 1 A method comprising: receiving, from a client and at a first read-write computer of a first computer system in a distributed system having a plurality of computer systems, data to be written into the first computer system, each of the computer systems including at least a read-write computer and a data storage unit; writing the data into a first data storage unit of the first computer system; and replicating the data from the first data storage unit to data storage units of other computer systems in the distributed system, the replicating including transmitting, via a publish-subscribe model, the data from the first data storage unit to read-write computers of the other computer systems that have subscribed to the data from the first data storage unit, and writing, by the read-write computers, the data to the corresponding data storage units. ### Claim 2 (depends on 1) The method of claim 1, wherein each of the computer systems is at a different location. ### Claim 3 (depends on 1) The method of claim 1, wherein the read-write computer of a computer system serves at least one of (a) read request or (b) a write request, from the client. ### Claim 4 (depends on 1) The method of claim 1, wherein the read-write computer of a computer system includes a high-speed cache. ### Claim 5 (depends on 1) The method of claim 1, wherein the first read-write computer caches data from the first data storage unit in a high speed cache. ### Claim 6 (depends on 5) The method of claim 5, wherein the first read-write computer serves a read request from the client by obtaining data from the high speed cache. ### Claim 7 (depends on 1) The method of claim 1, wherein writing the data to the first data storage unit includes determining whether the first data storage unit has failed; responsive to a determination that the first data storage unit has failed, routing, by the first read-write computer, all write requests from the client to a second computer system in the distributed environment. ### Claim 8 (depends on 7) The method of claim 7, wherein the second computer system is selected based on a predefined policy. ### Claim 9 (depends on 1) The method of claim 1 further comprising: determining whether a replication lag of the first data storage unit with respect to other data storage units of the computer systems exceeds a predefined threshold; and responsive to a determination that the replication lag of the first data storage unit exceeds a predefined threshold, changing a status of the first data storage unit to a catch up mode, indicating that the first data storage unit is (a) processing replicated data from other data storage units and (b) not accepting write requests from the client, routing, by the first read-write computer, all write requests from the client to a second data storage unit of a second computer system in the distributed system, allowing the first data storage unit to process the replicated data from other data storage units until the replication lag is below the predefined threshold, and changing the status of the first data storage unit from the catch up mode to active mode, indicating that the first data storage unit is ready to accept write requests from the client. ### Claim 10 (depends on 1) The method of claim 1, wherein replicating the data from the first data storage unit to data storage units of other computer systems includes receiving, at the first read-write computer and via the publish-subscribe model, replicated data from data storage units of other computer systems to which the first read-write computer has subscribed, determining whether a data item in the replicated data has been modified by more than one computer system in the distributed system, responsive to a determination that the data item has been modified by more than one computer system, resolving a conflict between copies of the data item from different computer systems based on a predefined criteria, and writing the replicated data into the first data storage unit. ### Claim 11 (depends on 10) The method of claim 10, wherein each of the data items in the replicated data has a format that includes at least (a) payload, (b) a timestamp of creation or modification of a data item, (c) a version of the data item, and (d) a region of a computer system from which the data item is replicated. ### Claim 12 (depends on 11) The method of claim 11, wherein the version and the timestamp are updated each time the payload of the data item is modified. ### Claim 13 (depends on 10) The method of claim 10, wherein the predefined criteria for resolving the conflicts include selecting a copy of the data item, from a particular computer system, that has a latest version among the copies from the different computer systems, if the version is same among the copies of the data item, selecting a copy of the data item, from a particular computer system, that has a latest timestamp among the copies from the different computer systems, and if the timestamp is same among the copies of the data item, selecting a copy of the data item from a computer system of a particular region based on a predefined policy. ### Claim 14 (depends on 1) The method of claim 1, wherein the data storage unit of each of the computer systems is split into a plurality of logical partitions, each of the logical partitions including a subset of data from the data storage unit. ### Claim 15 (depends on 14) The method of claim 14, wherein replicating the data from the first data storage unit to data storage units of the other computer systems includes transmitting, via the publish-subscribe model, the data from each of the logical partitions of the first data storage unit. ### Claim 16 (depends on 15) The method of claim 15, wherein transmitting the data via the publish-subscribe model includes publishing, by the first data storage unit, the data from each of the logical partitions to other computer systems that have subscribed to the data of a particular logical partition in the first data storage unit. ### Claim 17 A method comprising: determining whether a first data storage unit of a first computer system in a distributed system has failed, wherein the first computer system is one of a plurality of computer systems in the distributed system and wherein each of the computer systems includes at least a read-write computer and a data storage unit; responsive to a determination that the first data storage unit has failed, determining if the data from the first data storage unit is replicated to other computer systems in the distributed system that have subscribed to the data from the first computer system; responsive to a determination that the data from the first data storage unit is not replicated to the other computer systems, determining a particular data storage unit in the distributed system that has latest data corresponding to the first data storage unit; and replicating, via a publish-subscribe model, the latest data from the particular data storage unit to the other computer systems. ### Claim 18 (depends on 17) The method of claim 17, wherein replicating the latest data from the particular data storage unit to the other computer systems includes transmitting the latest data from the particular data storage unit to read-write computers of the other computer systems, and writing, by the read-write computers, the latest data into corresponding data storage units of the other computer systems. ### Claim 19 (depends on 18) The method of claim 18 further comprising: reconstructing the first data storage unit by copying data from a source data storage unit of one of the other computer systems to the first data storage unit. ### Claim 20 (depends on 19) The method of claim 19, wherein copying the data from the source data storage unit includes: copying at least transaction files and bin log files from the source data storage unit, and replicating, via the publish-subscribe model, data from the source data storage unit to the first computer system, to obtain the data that is not included in the transaction files and bin log files. ### Claim 21 An apparatus comprising: a distributed system having a plurality of computer systems, each of the computer systems having at least a read-write computer to serve a read request and/or a write request from a client, a data storage unit to store data forwarded from the read-write computer, and a replication module to replicate, via a publish-subscribe model, the data stored in the data storage unit of a computer system to data storage units of other computer systems, wherein the other computer systems are computer systems that have subscribed to data from the data storage unit and wherein the replication module transmits the data from the data storage unit to read-write computers of the other computer systems.