Patent Description:
Cloud computing has increased in popularity as storage of large quantities of data in the cloud becomes more common. The need for robust storage of data has also grown to protect the increasingly large quantity of data stored in the cloud. Some cloud service providers increase the robustness of cloud data storage by asynchronously storing data at multiple storage locations across different geographical zones. For example, after writing the data to a first or primary storage location in a first zone, the data is then written to a second storage location in a second zone. Storing the data at these different storage locations allows users to recover their data in the event that one of the storage locations becomes inaccessible. For some data, such as for mission critical applications, data loss is intolerable.

<CIT> relates to fast failover with multiple secondary nodes.

<CIT> discloses a storage replication system with data tracking.

One aspect of the disclosure provides a method for synchronous replication of high throughput streaming data. The method includes receiving, at data processing hardware, a stream of data blocks for storage at a first storage location of a distributed storage system and at a second storage location of the distributed storage system. The first storage location is associated with a first geographical region and the second storage location is associated with a second geographical region different than the first geographical region. The method also includes synchronously writing, by the data processing hardware, the stream of data blocks to the first storage location and to the second storage location. While synchronously writing the stream of data blocks to the first storage location and to the second storage location, the method includes determining, by the data processing hardware, an unrecoverable failure at the second storage location that prohibits further writing of the stream of data blocks to the second storage location. The method also includes determining, by the data processing hardware, a failure point in the writing of the stream of data blocks. The failure point demarcates data blocks that were successfully written to the second storage location and data blocks that were not successfully written to the second storage location. The method also includes synchronously writing, by the data processing hardware, starting at the failure point, the stream of data blocks to the first storage location and to a third storage location of the distributed storage system. The third storage system is associated with a third geographical region different than the first geographical region and the second geographical region.

Implementations of the disclosure may include one or more of the following optional features. In some implementations, the method further includes, asynchronously writing, by the data processing hardware, the stream of data blocks to third storage location from a beginning point of the stream of data blocks to the failure point. In some examples, determining the unrecoverable failure at the second storage location that prohibits further writing of the stream of data blocks to the second storage location includes determining a failure of the writing of the stream of data blocks to the second storage location; in response to determining the failure of writing the stream of data blocks to the second storage location, retrying to write the stream of data blocks to the second storage location; and when retrying to write the stream of data blocks to the second storage location has failed, determining that the failure is an unrecoverable failure.

Optionally, determining the failure point in the writing of the stream of data blocks includes determining whether a first replication log is available which indicates the data blocks that have been successfully committed to the first storage location; determining whether a second replication log is available which indicates the data blocks that have been successfully committed to the second storage location; and when the first replication log and the second replication log are available, reconciling, based on a length of the first replication log and a length of the second replication log, the first replication log and the second replication log. In some examples, reconciling the first replication log and the second replication log includes determining an index of the second replication log associated with the unrecoverable failure; storing the index of the second replication log on memory hardware in communication with the data processing hardware; finalizing the second replication log to prohibit further writes to the second storage location; and generating a sentinel file to indicate a need for reconciliation. In other examples, when the first replication log is available and the second replication log is not available, the method further includes reconciling, by the data processing hardware, based on the length of the first replication log, the first replication log and the second replication log. When the first replication log is not available and the second replication log is available, the method may further include reconciling, by the data processing hardware, the first replication log and the second replication log based on the length of the second replication log.

In some implementations, the method further includes, generating, by the data processing hardware, a first replication log includes timestamps which indicates when each data block is written to the first storage location and generating, by the data processing hardware, a second replication log that includes timestamps which indicate when each data block is written to the second storage location. In these implementations, the method further includes, receiving, at the data processing hardware, a query request requesting return of a plurality of data blocks stored at the first storage location; reconciling, by the data processing hardware, based on a length of the first replication log and a length of the second replication log, the first replication log and the second replication log; and returning, by the data processing hardware, based on the reconciliation of the first replication log and the second replication log, the requested plurality of data blocks. Optionally, reconciling the first replication log and the second replication log includes determining that the length of the second replication log is not available and determining, within a threshold period of time, that a subsequent write is added to the first replication log.

Another aspect of the disclosure provides a system for synchronous replication of high throughput streaming data. The system includes data processing hardware and memory hardware in communication with the data processing hardware. The memory hardware storing instructions that when executed on the data processing hardware cause the data processing hardware to perform operations. The operations include receiving a stream of data blocks for storage at a first storage location of a distributed storage system and at a second storage location of the distributed storage system. The first storage location is associated with a first geographical region and the second storage location is associated with a second geographical region different than the first geographical region. The operations also include synchronously writing the stream of data blocks to the first storage location and to the second storage location. While synchronously writing the stream of data blocks to the first storage location and to the second storage location, the operations include determining an unrecoverable failure at the second storage location that prohibits further writing of the stream of data blocks to the second storage location. The operations also include determining a failure point in the writing of the stream of data blocks. The failure point demarcates data blocks that were successfully written to the second storage location and data blocks that were not successfully written to the second storage location. The operations also include synchronously writing, starting at the failure point, the stream of data blocks to the first storage location and to a third storage location of the distributed storage system. The third storage system is associated with a third geographical region different than the first geographical region and the second geographical region.

Implementations of the disclosure may include one or more of the following optional features. In some implementations, the operations also include, asynchronously writing the stream of data blocks to third storage location from a beginning point of the stream of data blocks to the failure point. In some examples, determining the unrecoverable failure at the second storage location that prohibits further writing of the stream of data blocks to the second storage location includes determining a failure of the writing of the stream of data blocks to the second storage location; in response to determining the failure of writing the stream of data blocks to the second storage location, retrying to write the stream of data blocks to the second storage location; and when retrying to write the stream of data blocks to the second storage location has failed, determining that the failure is an unrecoverable failure.

Optionally, determining the failure point in the writing of the stream of data blocks includes determining whether a first replication log is available which indicates the data blocks that have been successfully committed to the first storage location; determining whether a second replication log is available which indicates the data blocks that have been successfully committed to the second storage location; and when the first replication log and the second replication log are available, reconciling, based on a length of the first replication log and a length of the second replication log, the first replication log and the second replication log. In some examples, reconciling the first replication log and the second replication log includes determining an index of the second replication log associated with the unrecoverable failure; storing the index of the second replication log on memory hardware in communication with the data processing hardware; finalizing the second replication log to prohibit further writes to the second storage location; and generating a sentinel file to indicate a need for reconciliation. In other examples, when the first replication log is available and the second replication log is not available, the operations further include reconciling, based on the length of the first replication log, the first replication log and the second replication log. When the first replication log is not available and the second replication log is available, the operations further include reconciling the first replication log and the second replication log based on the length of the second replication log.

In some implementations, the operations further include, generating a first replication log includes timestamps which indicates when each data block is written to the first storage location and generating a second replication log that includes timestamps which indicate when each data block is written to the second storage location. In these implementations, the operations further include, receiving a query request requesting return of a plurality of data blocks stored at the first storage location; reconciling, based on a length of the first replication log and a length of the second replication log, the first replication log and the second replication log; and returning, based on the reconciliation of the first replication log and the second replication log, the requested plurality of data blocks. Optionally, reconciling the first replication log and the second replication log includes determining that the length of the second replication log is not available and determining, within a threshold period of time, that a subsequent write is added to the first replication log.

As cloud data warehouses have increased in size and popularity, the amount of data that they ingest has increased exponentially. For example, some systems ingest (i.e., receive) hundreds of gigabytes a second if not more. This ingested data is often stored in multiple storage locations located in different geographical regions (i.e., "zones") such as a primary storage location and a secondary storage location to provide a backup for the data. The secondary storage location for the data acts as a failsafe that provides added protection for the data if the primary storage location is damaged or otherwise inaccessible. For example, when a natural disaster occurs in the geographical region of the first storage location the data is not accessible at the first storage location. In this example, the data remains accessible at the second storage location located in a different geographical region than the first storage location.

Conventional systems write data to the primary storage location upon data ingestion and acknowledge receipt of the data to the user. Then, an asynchronous background process replicates the data from the primary storage location to the secondary storage location. For example, after a period of time of inactivity at the first storage location (e.g., no new data has been received for fifteen minutes) or when a sufficient amount of data has been written to the primary storage location (e.g., <NUM> MBs), the asynchronous background process replicates and stores the same data from the primary storage location to the secondary storage location.

This asynchronous replication process leaves a window of vulnerability where a permanent failure that occurs at the first storage location can lead to irrecoverable data loss of the data. Specifically, when a failure at the first storage location occurs before the system writes the data via the asynchronous replication process to the second storage location, the data may be permanently lost because the data has not yet been stored at the secondary storage location.

In contrast, a synchronous replication process stores the streaming data received from a user at the primary storage location and secondary storage location simultaneously. Thus, if a failure occurs during or after writing the data to either the primary storage location or the secondary storage location, the data is still accessible at the other storage location. However, when failures occur during the synchronous replication to the primary storage location or the secondary storage locations the synchronous replication process failure handling must ensure that the data remains available and is still replicated properly. For example, when the secondary storage location becomes inaccessible while synchronously replicating the data to the primary storage location and the secondary storage location simultaneously, there is no complete backup for the data.

Implementations herein are directed toward a system for synchronously replicating streaming data to a first storage location and a second storage location simultaneously. The system manages failures that occur during the synchronous replication process such that streaming data always maintains availability at the primary storage location and/or the secondary storage location. For example, while synchronously writing the data blocks to the first storage location and the second storage location, the system determines an unrecoverable failure that prohibits additional data blocks from writing to the second storage location. In response to the failure, the system synchronously writes the stream of data blocks to the first storage location and a third storage location starting at the failure point of the unrecoverable failure.

Referring now to <FIG>, in some implementations, an example system <NUM> includes a user device <NUM> associated with a respective user <NUM> in communication with a remote system <NUM> via a network <NUM>. The user device <NUM> may correspond to any computing device, such as a desktop workstation, a laptop workstation, or a mobile device (i.e., a smart phone). The user device <NUM> includes computing resources <NUM> (e.g., data processing hardware) and/or storage resources <NUM> (e.g., memory hardware).

The remote system <NUM> may be a single computer, multiple computers, or a distributed system (e.g., a cloud environment) having scalable / elastic resources <NUM> including computing resources <NUM> (e.g., data processing hardware) and/or storage resources <NUM> (e.g., memory hardware). A plurality of data block data stores <NUM>, 150a-c (i.e., remote storage devices <NUM>) are overlain on the storage resources <NUM> to allow scalable use of the storage resources <NUM> by one or both of the user <NUM> and computing resources <NUM>. The data block data stores <NUM> may each be configured to store a plurality of data blocks <NUM>, 22a-n within a data structure (e.g., a table). Each data block data store <NUM> is associated with a storage location <NUM>, 210a-c and each storage location <NUM> is associated with a different geographical region.

The user <NUM> transmits a stream of data blocks <NUM> using the user device <NUM> via the network <NUM> to the remote system <NUM> for storage at the remote system <NUM>. The stream of data blocks <NUM> may include any number of data blocks <NUM>. The remote system <NUM> receives the stream of data blocks <NUM> for storage at a first storage location 210a (e.g., a primary storage location <NUM>) of the remote system <NUM> and at a second storage location 210b of the remote system <NUM> (e.g., a secondary storage location <NUM>). The first storage location 210a is associated with a first geographical region and the second storage location 210b is associated with a second geographic region different than the first geographical region. For example, the first storage location 210a is located in New York City and the second storage location 210b is located in Los Angeles. Each storage location <NUM> of the remote system <NUM> includes one or more data block data stores <NUM> configured to store the stream of data blocks <NUM>.

The remote system <NUM> executes a data block replicator <NUM> configured to replicate the stream of data blocks <NUM> received from the user device <NUM>. That is, the data block replicator <NUM> duplicates each data block <NUM> of the stream of data blocks <NUM> into two separate streams of data blocks <NUM>. By replicating the stream of data blocks <NUM> into two streams of data blocks <NUM>, the data block replicator <NUM> may write the stream of data blocks <NUM> to the first storage location 210a and the second storage location 210b synchronously (i.e., simultaneously). For example, the stream of data blocks <NUM> represents some or all of a table that the remote system <NUM> stores at a data block data store 150a at the first storage location 210a (e.g., Los Angeles) and at a data block data store 150b at the second storage location 210b (e.g., New York City). The data block replicator <NUM> synchronously writes each data block <NUM> of the stream of data blocks <NUM> to both the first storage location 210a and the second storage location 210b at the same time or substantially at the same time. To ensure that the writes stay synchronous, the data block replicator <NUM> may transmit each pair of replicated data blocks <NUM> at or near the same time to the primary storage location 210a and the secondary storage location 210b and then wait to transmit the next data block <NUM> until the data block replicator <NUM> confirms that both receiving data block data stores <NUM> (e.g., the data block data store 150a and the data block data store 150b) received the previous data block <NUM>. That is, each pair of replicated data blocks <NUM> are transmitted simultaneously and thus the data block replicator <NUM> maintains synchronicity throughout writing the stream of data blocks <NUM>.

In some examples, the user <NUM> selects the primary storage location <NUM> (i.e., the first storage location 210a). For example, the user <NUM> selects that the primary storage location for the data is Los Angeles. In this example, either the user <NUM> or the remote system <NUM> selects the secondary storage location <NUM> (i.e., the second storage location 210b). For example, the remote system <NUM> prompts the user <NUM> to select a secondary storage location <NUM> that is different than the primary storage location <NUM>. Alternatively, the remote system <NUM> automatically selects a secondary storage location <NUM> that is different than the primary storage location <NUM> based on any number of factors (e.g., location, distance from primary storage location <NUM>, congestion, cost, latency, etc.). In other examples, the remote system <NUM> selects both the primary storage location <NUM> and the secondary storage location <NUM> automatically based on any of the same factors.

While synchronously writing the stream of data blocks <NUM> to the first storage location 210a and to the second storage location 210b, the remote system <NUM>, in some implementations, determines an unrecoverable failure at the second storage location 210b that prohibits further writing of the stream of data blocks <NUM> to the second storage location 210b. For example, the second storage location 210b is subject to a natural disaster that physically damages the data block data store 150b of the second storage location 210b. In another example, the data block data store 150b of the second storage location 210b encounters network connection issues that prevent communication with the second storage location 210b and/or data block data store 150b. An unrecoverable error is defined as an error that persists for a threshold period of time and/or persists after a threshold number of retry attempts.

The unrecoverable failure may occur such that the data block replicator <NUM> only writes a portion of the data blocks <NUM> in the stream of data blocks <NUM> to the second storage location 210b. That is, the data block replicator <NUM> writes only a portion of the stream of data blocks <NUM> before the unrecoverable failure prohibits any further communication with the second storage location 210b and/or data block data store 150b. The remote system <NUM> determines a failure point <NUM> (<FIG> and <FIG>) in the writing of the stream of data blocks <NUM>. In particular, the failure point <NUM> indicates at which data block <NUM> in the stream of data blocks <NUM> the unrecoverable failure occurred at. That is, the failure point <NUM> demarcates the data blocks <NUM> that the remote system <NUM> successfully wrote to the second storage location 210b from the data blocks <NUM> that the remote system <NUM> did not successfully write to the second storage location 210b. While the examples herein describe the unrecoverable failure prohibiting further writing to the second storage location 210b, it is understood that the unrecoverable failure may occur between either the primary storage location <NUM> (e.g., the first storage location 210a) or the secondary storage location <NUM> (e.g., the second storage location 210b) and still remain within the scope of the implementations herein.

After determining the failure point <NUM> at the second storage location 210b, the data block replicator <NUM>, starting at the failure point <NUM>, synchronously writes the stream of data blocks <NUM> to the first storage location 210a and to a data block data store 150c of a third storage location 210c. The third storage location 210c is associated with a third geographical region (e.g., Miami) different than the first geographic region (e.g., Los Angeles) and the second geographical region (e.g., New York City).

Referring now to <FIG>, in some implementations, the data block replicator <NUM> receives the stream of data blocks <NUM> that include data blocks 22a-j. The data block replicator <NUM> replicates the stream of data blocks <NUM> (e.g., data blocks 22a-j) to write to both the first storage location 210a and the second storage location 210b simultaneously. That is, the data block replicator <NUM> replicates the stream of data blocks <NUM> into two streams of data blocks <NUM> (e.g., two streams of data blocks 22a-j) that allows the data block replicator <NUM> to synchronously write the stream of data blocks <NUM> to both the first storage location 210a and the second storage location 210b. That is, instead of writing the data blocks <NUM> asynchronously, such as by starting the writes to the second storage location 210b a period of time after starting the writes to the first storage location 210a, the data block replicator <NUM> synchronously writes each pair of replicated data blocks <NUM> to the first storage location 210a and the second storage location 210b.

After replicating the stream of data blocks <NUM>, the data block replicator <NUM> writes each data block <NUM> of the stream of data blocks <NUM> to the data block data store 150a at the first storage location 210a and to the data block data store 150b at the second storage location 210b. While synchronously writing the stream of data blocks <NUM> to the first storage location 210a and second storage location 210b, the remote system <NUM> determines the unrecoverable failure at the second storage location 210b. The unrecoverable failure prohibits further writing of the stream of data blocks <NUM> to the second storage location 210b. In some implementations, the remote system <NUM> determines that a write of one of the data blocks <NUM> has failed. For example, the second storage location 210b waits for confirmation of the write of each data block <NUM> for a threshold period of time. Here, the write of the data block 22f to the second storage location 210b failed while the write of the data block 22f to the first storage location was successful. When the data block replicator <NUM> cannot determine that a write for each of a pair of replicated data blocks <NUM> (i.e., the data block <NUM> for the first storage location 210a and the corresponding data block <NUM> for the second storage location 210b) has occurred within the threshold period of time, the remote system <NUM> determines that the write of the data block <NUM> has failed.

In response to determining the failure of writing one of the data blocks <NUM> to the second storage location 210b (or, in other examples, the first storage location 210a), the remote system <NUM> may retry to write the stream of data blocks <NUM> to the second storage location 210b. For example, the second storage location 210b loses network connection to the remote system <NUM> momentarily but recovers network connectivity within the threshold period of time. Thus, the data block replicator <NUM> may attempt to transmit the failed data block <NUM> any number of times prior to determining that the failure is unrecoverable. Because the data block replicator writes the data blocks <NUM> to both the first storage location 210a and the second storage location 210b synchronously, the data block replicator <NUM> may refrain from writing any additional data blocks <NUM> to the first storage location 210a while retrying the failed write to the second storage location 210b.

When one of the retries to write the failed data blocks <NUM> is successful, the remote system <NUM> may determine that the failure is not an unrecoverable failure and continues to write the stream of data blocks <NUM> to both the first storage location 210a and the second storage location 210b. Where the remote system <NUM> determines that the one or more retries of the failed write is unsuccessful, the remote system <NUM> determines that the failure is an unrecoverable failure that prohibits further writing of the stream of data blocks <NUM> to the second storage location 210b. The data block replicator <NUM> determines the failure point <NUM> based on, for example, which data block <NUM> failed to write or based on the last successful data block <NUM> write. In the example of <FIG>, the data block replicator <NUM> determines that the failure point occurred between data blocks 22e (which the data block replicator <NUM> successfully wrote to the second storage location 210b) and the data block 22f (which the data block replicator <NUM> failed to write to the second storage location 210b).

In some examples, after the remote system <NUM> determines the unrecoverable failure at the second storage location 210b, the remote system <NUM> returns an error or an indication of the failure to the user <NUM>. The indication may include details such as how many data blocks <NUM> were successfully written prior to the failure, the failure point <NUM>, which storage location <NUM> the write failed on, etc..

Referring now to <FIG>, in some examples, after the remote system <NUM> determines the unrecoverable failure at the second storage location 210b and the failure point <NUM>, the data block replicator <NUM>, starting at the failure point <NUM>, synchronously writes the stream of data blocks <NUM> to the first storage location 210a and the third storage location 210c. In particular, after the unrecoverable failure at the second storage location 210b, the data block replicator <NUM> stops attempting to write the data blocks <NUM> to the second storage location 210b and instead begins writing the data blocks <NUM> to the third storage location 210c synchronously with the first storage location 210a. In this example, after determining the failure point <NUM> is in between the data block 22e and the data block 22f of the stream of data blocks 22a-j, the data block replicator <NUM> stops attempting to write the data block 22f to the second storage location 210b. The data block replicator <NUM> begins synchronously writing the stream of data blocks <NUM> starting at the failure point <NUM> (i.e., data block 22f) to the first storage location 210a and the second storage location 210b until the end of the stream of data blocks <NUM> (i.e., data block 22j). Thus, the data block replicator <NUM> maintains a synchronous write of the stream of data blocks <NUM> throughout writing the entire stream of data blocks <NUM>.

Referring now to <FIG>, in some implementations, the data block replicator <NUM> asynchronously writes, from a beginning of the stream of data blocks <NUM> (i.e., the first data block <NUM> in the stream of data blocks <NUM>) to the failure point <NUM>, the stream of data blocks <NUM> to the third storage location 210c. For example, after the data block replicator <NUM> completes the synchronous writing of the stream of data blocks <NUM> to the first storage location 210a and the third storage location 210c (i.e., after data block 22j), the first storage location 210a includes data blocks 22a-j and the third storage location includes data blocks 22f-j. To complete the replication of the stream of data blocks <NUM> at the third storage location 210c, the data block replicator <NUM> asynchronously writes the remaining data blocks 22a-e to the third storage location 210c (i.e., the data blocks <NUM> that the data block replicator <NUM> successfully wrote to the second storage location 210b). After the data block replicator <NUM> asynchronously writes the stream data blocks <NUM> to the third storage location 210c, both the first storage location 210a and third storage location 210c include the entire stream of data blocks <NUM> (e.g., data blocks 22a-j).

In some examples, the data block replicator <NUM> asynchronously writes the stream of data blocks <NUM> to the third storage location 210c after the data block replicator <NUM> synchronously writes the stream of data blocks <NUM> to the third storage location 210c. In other examples, the data block replicator <NUM> asynchronously writes the stream of data blocks <NUM> to the third storage location 210c while the data block replicator <NUM> synchronously writes the stream of data blocks <NUM> to the third storage location 210c with the first storage location 210a.

Referring now to schematic view 300a of <FIG>, in some implementations, the remote system <NUM> generates a replication log <NUM>, 310a-b for each storage location <NUM> and/or each data block data store <NUM>. The replication log <NUM> may be the physical file that stores the data blocks <NUM>. Alternatively, the replication log <NUM> may be a separate file. In this example, the remote system <NUM> generates a first replication log 310a that includes timestamps that indicate a write time for each data block <NUM> successfully written to the first storage location 210a and/or the data block data store 150a. That is, when the data block replicator <NUM> successfully writes a data block <NUM> (i.e., commits) to the storage location <NUM>, the remote system <NUM> updates the corresponding replication log <NUM> to reflect the write (e.g., with a corresponding timestamp). On the other hand, when the data block replicator <NUM> fails to successfully write a data block <NUM> to the storage location <NUM>, the remote system <NUM> does not update the corresponding replication log to reflect the write. Because each storage location <NUM> includes a corresponding replication log <NUM>, the remote system <NUM> may use the replication logs <NUM> to ensure coherency between the replications.

In some examples, the remote system <NUM> executes a reconciler <NUM>. The reconciler <NUM> may determine the failure point <NUM> in the stream of data blocks <NUM> by determining whether the first replication log 310a that indicates which data blocks <NUM> have been successfully committed to the first storage location 210a are available and by determining whether the second replication log 310b that indicates the data blocks <NUM> that have been successfully committed to the second storage location 210b are available. That is, the remote system <NUM> (or, in some examples, the user <NUM>) notifies the reconciler <NUM> of an error and the reconciler <NUM> attempts to obtain both the first replication log 310a from the first storage location 210a and the second replication log 310b from the second storage location 210b. In the example of <FIG>, the reconciler <NUM> determines that both the first replication log 310a and the second replication log 310b are available.

When the first replication log 310a and the second replication log 310b are both available, the reconciler <NUM> reconciles the first replication log 310a and the second replication log 310b based on a length of the first replication log 310a and a length of the second replication log 310b. In some examples, the reconciler <NUM> determines that the replication logs <NUM> are the same length (i.e., the same size) and the replication logs <NUM> each include records of data blocks <NUM> after the failure point <NUM>. This means that the remote system <NUM> recovered from the failure and the stream of data blocks <NUM> committed to both the first storage location 210a and the second storage location 210b completed successfully. In this scenario, the reconciler <NUM> informs the user <NUM> that the replication was successful.

In other examples, the reconciler <NUM> determines that the length of the second replication log 310b is less than the length of the first replication log 310a (i.e., the size of the second replication log 310b is less than the size of the first replication log 310a). This scenario indicates that either there is currently an "inflight" write between the data block replicator <NUM> and the second storage location 210b causing the different lengths in the replication logs 310a, 310b or there is no inflight write and the second storage location 210b has suffered a failure. That is, the data block replicator <NUM> successfully wrote a data block <NUM> to the first storage location 210a and failed to write the corresponding replicated data block <NUM> to the second storage location 210b. Because typically the remote system <NUM> and/or user <NUM> does notify the reconciler <NUM> of a failure until after waiting a threshold period of time, the reconciler <NUM> may assume that there is no inflight write. Continuing the example of <FIG>, the reconciler <NUM> determines that the length of the first replication log 310a (which includes entries for the writes of data blocks 22a-f) is longer than the length of the second replication log 310b (which includes entries for the writes of data blocks 22a-e).

After the reconciler <NUM> determines that the length of the first replication log 310a is different than the length of the second replication log 310b, the reconciler <NUM> performs a reconciliation process (e.g., a forced finalization). The reconciler <NUM> may reconcile the first replication log 310a and the second replication log 310b by determining an index of the second replication log 310b associated with the unrecoverable failure (e.g., a location of the unrecoverable failure or the failure point <NUM>). After the reconciler <NUM> determines the index of the second replication log 310b, the reconciler <NUM> may store the index of the second replication log 310b at the memory hardware <NUM>. That is, the remote system <NUM> may store the length of the replication log for future reference. In some implementations, the reconciler <NUM> finalizes the second replication log 310b to prohibit further writes to the second storage location 210b and generates a sentinel file <NUM> to indicate a need for reconciliation.

In some examples, the data block replicator <NUM> engages a separate stream server or worker (i.e., a processing task or thread) that is responsible for writing the data blocks <NUM> to the storage location <NUM>. In the event that the worker temporarily fails to write a data block <NUM>, but remains active (i.e., a "zombie" worker), the worker may attempt to continue writing data blocks <NUM> to the second storage location 210b even after the data block replicator <NUM> has determined an unrecoverable failure and began writing the stream of data blocks <NUM> to the third storage location 210c. When this zombie worker encounters the finalized replication log <NUM>, the worker is incapable of actually committing any additional writes to the replication log <NUM>. Furthermore, the sentinel file <NUM> serves as a notice to the worker to abort ownership of the log file.

Referring now to <FIG>, in some examples, when the first replication log 310a is available and the second replication log 310b is not available (or vice versa), the reconciler <NUM> again reconciles the first replication log 310a and the second replication log 310b based on the length of the first replication log 310a. For example, when the unrecoverable failure is such that the second storage location 210b is unreachable, the second replication log 310b is similarly unreachable. In this instance, the reconciler <NUM> cannot compare the lengths of the first replication log 310a and the second replication log 310b and therefore relies on the length of just the first replication log 310a. Similarly, when the first replication log 310a is not available and the second replication log is available, the reconciler <NUM> may reconcile, based on the length of the second replication log 310b, the first replication log 310a and the second replication log 310b.

The reconciler <NUM> may rely on only the length of the available replication log <NUM> because the reconciler <NUM> will successfully reconcile whether the available replication log <NUM> is longer or shorter than the unavailable replication log <NUM>. When the available replication log <NUM> is shorter than the unavailable replication log <NUM>, this can only occur when the failure occurs after the data block <NUM> is written to the storage location <NUM>, but before the data block <NUM> is committed to the storage location. Thus, the reconciler <NUM> is safe to assume that the "extra" data block <NUM> that increased the length of the unavailable replication log <NUM> is not committed.

On the other hand, when the available replication log <NUM> is longer than the unavailable replication log <NUM>, the additional data block <NUM> that was written to the available storage location <NUM> corresponds to a write that failed (e.g., a write that failed with an internal error such as an HyperText Transfer Protocol (HTTP) <NUM> error). The reconciler <NUM> may safely assume this data is committed as long as the remote system <NUM> consistently serves the data block <NUM> on all reads.

Referring now to <FIG>, in some examples, the remote system <NUM> receives a query request <NUM> requesting return of a plurality of data blocks <NUM> stored at the first storage location 210a (i.e., the primary storage location <NUM>). The query request <NUM> may include one or more parameters that define what data block <NUM> to return such as one or more keywords. The parameters include a snapshot read timestamp <NUM>. The snapshot read timestamp <NUM> specifies a point in time for the remote system <NUM> to read the data block data store <NUM>. The remote system <NUM> correlates the snapshot read timestamp <NUM> with the timestamp associated with each data block <NUM> in the replication log <NUM> as the timestamp in the replication log <NUM> indicates at what point in time the data block <NUM> was committed to the data block data store <NUM>. For example, the remote system <NUM> does return data block <NUM> that the data block replicator <NUM> wrote to the data block data store <NUM> after the snapshot read timestamp <NUM>. When the query request <NUM> does not include a snapshot read timestamp <NUM>, the remote system <NUM> may default the snapshot read timestamp <NUM> to the current time.

Because the timestamps added to the replication log <NUM> as data blocks <NUM> are committed to the data block data store <NUM> monotonically increase, when the remote system <NUM> reads the replication log <NUM>, once the remote system <NUM> encounters a data block <NUM> with a commit timestamp that is greater than the snapshot read timestamp <NUM>, the remote system <NUM> can safely assume that all further records in the replication log <NUM> will also have a timestamp larger than the snapshot read timestamp <NUM> and thus the remote system <NUM> may stop reading the replication log <NUM>. However, when the last record in the replication log <NUM> is associated with a data block <NUM> with a timestamp that is not greater than the snapshot read timestamp <NUM>, the remote system <NUM> may not make any assumptions about the last data block <NUM>. This is because the last data block <NUM> may be the result of a failed write and thus not have been successfully replicated to the other storage location <NUM>. Thus, the remote system <NUM>, prior to serving the last data block <NUM>, must first ensure that any other subsequent reads with a snapshot read timestamp <NUM> equal to or greater than the current snapshot read timestamp <NUM> will also consistently serve the last data block <NUM>. To this end, the reconciler <NUM> may reconcile reads from query requests <NUM>.

In some implementations, the reconciler <NUM> reconciles, based on a length of the first replication log 310a and a length of the second replication log 310b, the first replication log 310a and the second replication log 310b. In some examples, the reconciler <NUM> first determines that the length of the second replication log 310b is not available and the reconciler <NUM> determines that, within a threshold period of time, a subsequent write is added to the first replication log 310a. That is, to reconcile the state, the reconciler <NUM> determines the length of the replication logs 310a, 310b and when both are the same, the reconciler <NUM> determines that the last data block in the first replication log 310a is committed and complete. However, when either replication log 310a, 310b is unavailable or unreachable, the reconciler <NUM> may wait for a small amount of time (e.g., for fifty milliseconds) for a subsequent write to appear in the available replication log <NUM>. In this example, the remote system (e.g., the stream servers of the data block replicator <NUM>) performs periodic "keep-alive" writes to inactive replication logs <NUM> (e.g., every fifty milliseconds).

When a subsequent write does appear in the replication log <NUM>, the reconciler <NUM> may assume that replicated writes are occurring and thus that the previous data block <NUM> write was successful. Because this write will be associated with a timestamp that is greater than the snapshot read timestamp of the query request <NUM>, the reconciler <NUM> may safely assume that all data blocks <NUM> (i.e., the last data block <NUM> in the replication log <NUM>) are safe to serve. When a subsequent write does not appear after waiting for the small amount of time, the reconciler <NUM> retrieves any stored replication log length from previous reconciliations to determine whether the last data block <NUM> should be served. That is, when the data block <NUM> is at an index that is greater than the stored length from the previous reconciliation, the remote system <NUM> will not serve the last data block <NUM>. However, when the last data block <NUM> has an index that is the same or less than the stored length from the previous reconciliation, the remote system <NUM> will serve the last data block <NUM>. After any necessary reconciliation, the remote system <NUM> returns, based on the reconciliation of the first replication log 310a and the second replication log 310b, the requested plurality of data blocks <NUM>.

In some implementations, the remote system <NUM> receives multiple query requests <NUM> that each require reconciliation of the last data block <NUM>. In this scenario, only one of the query requests <NUM> will determine the final committed length of the replication log <NUM> and all of the other query requests <NUM> will be processed by the remote system <NUM> using the same final committed length. Thus, the remote system <NUM> offers consensus replication.

<FIG> is a flowchart of an exemplary arrangement of operations for a method <NUM> for synchronous replication of high throughput streaming data. The method <NUM>, at step <NUM>, includes receiving, at data processing hardware <NUM>, a stream of data blocks <NUM> for storage at a first storage location 210a of a distributed storage system <NUM> and at a second storage location 210b of the distributed storage system <NUM>. The first storage location 210a is associated with a first geographical region and the second storage location is associated with a second geographical region different from the first geographical region. The method <NUM>, at step <NUM>, includes synchronously writing, by the data processing hardware <NUM>, the stream of data blocks <NUM> to the first storage location 210a and to the second storage location 210b. While synchronously writing the stream of data blocks <NUM> to the first storage location 210a and to the second storage location 210b, the method <NUM>, at step <NUM>, includes determining, by the data processing hardware <NUM>, an unrecoverable failure at the second storage location 210b that prohibits further writing of the stream of data blocks <NUM> to the second storage location 210b.

The method <NUM>, at step <NUM>, includes determining, by the data processing hardware <NUM>, a failure point <NUM> in the writing of the stream of data blocks <NUM>. The failure point <NUM> demarcating data blocks <NUM> that were successfully written to the second storage location 210b and data blocks <NUM> that were not successfully written to the second storage location 210b. The method <NUM>, at step <NUM>, includes synchronously writing, by the data processing hardware <NUM>, starting at the failure point <NUM>, the stream of data blocks <NUM> to the first storage location 210a and to a third storage location 210c of the distributed storage system. The third storage location 210c is associated with a third geographical region different than the first geographical region and the second geographical region.

Claim 1:
A method (<NUM>) comprising:
receiving, at data processing hardware (<NUM>), a stream of data blocks (<NUM>) for storage at a first storage location (<NUM>) of a distributed storage system (<NUM>) and at a second storage location (<NUM>) of the distributed storage system (<NUM>), the first storage location (<NUM>) associated with a first geographical region and the second storage location (<NUM>) associated with a second geographical region different than the first geographical region;
synchronously writing, by the data processing hardware (<NUM>), the stream of data blocks (<NUM>) to the first storage location (<NUM>) and to the second storage location (<NUM>);
while synchronously writing the stream of data blocks (<NUM>) to the first storage location (<NUM>) and to the second storage location (<NUM>), determining, by the data processing hardware (<NUM>), an unrecoverable failure at the second storage location (<NUM>) that prohibits further writing of the stream of data blocks (<NUM>) to the second storage location (<NUM>);
determining, by the data processing hardware (<NUM>), a failure point (<NUM>) in the writing of the stream of data blocks (<NUM>), the failure point (<NUM>) demarcating data blocks (<NUM>) that were successfully written to the second storage location (<NUM>) and data blocks (<NUM>) that were not successfully written to the second storage location (<NUM>); and
synchronously writing, by the data processing hardware (<NUM>), starting at the failure point (<NUM>), the stream of data blocks (<NUM>) to the first storage location (<NUM>) and to a third storage location (<NUM>) of the distributed storage system (<NUM>), the third storage location (<NUM>) associated with a third geographical region different than the first geographical region and the second geographical region,
further comprising, asynchronously writing, by the data processing hardware (<NUM>), from a beginning point of the stream of data blocks (<NUM>) to the failure point (<NUM>), the stream of data blocks (<NUM>) to the third storage location (<NUM>).