Patent Publication Number: US-9411868-B2

Title: Passive real-time order state replication and recovery

Description:
BACKGROUND 
     1. Field 
     This disclosure relates generally to fault detection and recovery, and, more particularly, to a system and method for restoring a database, database record or associated metadata to a desired consistent order state. 
     2. Background 
     Performance requirements often demand persistent availability of computing resources. Computing systems designed to meet these demands are called “high-availability” computing systems. High availability systems utilize a number of diverse strategies to operate for long periods of time with a low rate of failure. When failures do occur, systems and strategies may be utilized to maintain the appearance of normal operation to external systems, recover a lost state or lost information, and restore normal operation as quickly as possible. 
     Traditionally, there are two primary replication and recovery solutions for high-availability computing systems. The first traditional replication and recovery solution relies on internal application logic to maintain persistent order state in memory in case of failure. However, this makes the system highly vulnerable to data corruption. If the host computer system crashes, there may be state and information losses upon recovery, leaving the application in an inconsistent and unpredictable condition. These outages are, of course, inevitable, and Information Technology (IT) personnel must often forego maintaining persistence to undergo time-intensive restoration of the application to a “clean” operating state, risking the loss of critical information. 
     The second traditional replication and recovery solution relies on active applications that are hosted on separate computer systems, tasked with providing data replication and recovery services. These applications are placed on the data flow path and intercept inbound data before passing it along to the supported computer system. While this solution avoids the pitfalls of relying on internal application logic, it necessarily introduces latency into the system, which may create unacceptable data bottlenecks, especially for systems that must handle large volumes of information. Furthermore, both traditional replication and recovery solutions typically require large amounts of processing time to recover the order state of the supported system. 
     BRIEF SUMMARY 
     In one aspect of this disclosure, a system and method for passive real-time order state replication and recovery is disclosed. Upstream data is received from an upstream system via a reliable transport, the upstream data also received by a supported system. Downstream data is received from the supported system via the reliable transport. Data acknowledgements are received from the supported system acknowledging receipt of the upstream and downstream data. A replicated current order state of the supported system is continuously updated in real-time based on the received upstream data, downstream data and the data acknowledgements. A recovery request is received after the supported system has experienced an outage. The current order state is restored to the supported system by transmitting a recovery message to the supported system containing the replicated current order state. 
     The foregoing has outlined rather generally the features and technical advantages of one or more embodiments of this disclosure in order that the following detailed description may be better understood. Additional features and advantages of this disclosure will be described hereinafter, which may form the subject of the claims of this application. 
    
    
     
       This disclosure is further described in the detailed description that follows, with reference to the drawings, in which: 
         FIG. 1  is a high level representation of an illustrative passive real-time order state replication and recovery system providing support for a supported data system; 
         FIGS. 1A-1F  are high level representations of an illustrative passive real-time order state replication and recovery system providing support for a supported data system; 
         FIG. 2  is a high level representation of an illustrative passive real-time order state replication and recovery system instance; 
         FIG. 3  is an illustrative sequence of steps for passive real-time order state replication; 
         FIG. 4  is an illustrative sequence of steps for real-time order state recovery; 
         FIG. 5  is a continuing illustrative sequence of steps for real-time order state recovery; 
         FIG. 6  is an illustrative table describing how different types of messages are handled during the recovery process; and 
         FIGS. 7A-7D  are high level representations of a supported system utilizing multiple illustrative passive real-time order state replication and recovery systems. 
     
    
    
     DETAILED DESCRIPTION 
     This application discloses a passive real-time order state replication and recovery system and method, which utilizes a separately hosted application to receive and process out-of-band upstream and downstream data to continuously update in real-time the order state of a supported data or computer system. By keeping the monitoring instances separate from the supported data system, the passive real-time order state replication and recovery system and method avoids the pitfalls of high-availability data systems that rely on internal application logic, in that an outage of the supported data system has no impact on the passive real-time order state replication and recovery system and method. Similarly, by utilizing particular low latency transmission protocols that are separate from the main data path to disseminate data, minimal latency is introduced to the system. 
     Moreover, instead of iterating through the historical sequence of inputs and outputs to restore application state, the passive real-time order state replication and recovery system continually maintains, in real-time, the same “order state” as the supported system. There are many different order states, examples of which include (but are not limited to) initial, acknowledgment, order is being processed, fill or kill, immediate, canceled, good for day, and closed. When the supported system experiences an outage, the preserved order state information is applied to the supported system to restore the order state of the supported system, minimizing the unavailability of the supported system due to the outage. 
       FIG. 1  is a high level representation of an illustrative passive real-time order state replication and recovery system and method providing support for a supported data system. The diagram illustrates the general operation of the entire system and method, and is described in detail through  FIGS. 1A-1F . 
       FIGS. 1A-1F  are high level representations of an illustrative passive real-time order state replication and recovery system (“OOM”)  100 .  FIG. 1A  illustrates OOM  100  during normal operation. OOM  100  provides replication and recovery support for a supported computer system  140 , which may be any type of computer system that processes orders and has stateful transactions. By way of example, the supported computer system  140  may be a trading engine or component in a trading network that receives orders, breaks up the received orders into multiple orders, and places the multiple orders onto multiple venues. Supported system  140  is tasked with receiving upstream message data  135  from an upstream server system  130  and transmitting downstream message data  145  to a downstream server system  150 . The upstream server system  130  may be, for example, an entry system, order router or a component that delivers an order to the supported system  140 . Similarly, the downstream server system  150  may be, for example, another trading engine or an exchange gateway that receives and delivers orders to multiple exchanges. 
     The upstream and downstream messages  135 ,  145  may be transmitted via a reliable transport  170 . The reliable transport  170  may be, for example, a topic-oriented data publication service, wherein systems subscribe to particular “topics” and receive only messages from the reliable transport  170  pertaining to the subscribed topic. Once an upstream message  135  is received by the supported system  140 , the received message  135  is processed by the supported system  140 . Upon successful processing of the upstream message  135 , the supported system  140  may send an acknowledgement  155  to the upstream server system  130  via reliable transport  170 . 
     OOM  100  supports the supported system  140  by receiving upstream messages  135 , downstream messages  145  and acknowledgements  155 , and continuously, in real-time, determining and storing a “snapshot” of the current order state of the supported system  140 . OOM  100  preferably receives the upstream message data  135  and downstream message data  145  via multicast transmission, ensuring that minimal latency is introduced to the transmission of the messages via the reliable transport  170 . The current order state of the supported system  140  may be maintained in real-time by OOM  100 , which receives the same upstream messages  135 , downstream messages  145  and acknowledgement messages  155  as the supported system  140 . 
     OOM  100  may have sublayers, including Administrative Layer  105 , Order Manager Layer  110  and Data Store  115 . The Administrative Layer  105  provides administrators with access to monitoring and control functions for the replication and recovery processes. The Order Manager Layer  110  may be used to create or alter order states upon receiving commands to do so. The Data Store  115  provides storage for order states or supported system states based upon the received upstream message data  135  and downstream message data  145 . 
       FIG. 1B  illustrates the supported system  140 , reliable transport  170  and OOM  100  after the supported system  140  has experienced an outage. The supported system  140  is rendered offline, and communication with the upstream server system  130  and downstream server system  150  is broken. However, the reliable transport  170  continues to function and OOM  100  temporarily continues to receive inbound transmissions from the upstream server system  130 , as long as the upstream server system  130  continues to transmit upstream messages  135 . Once the upstream server system  130  determines that the supported system  140  has gone offline, it stops transmitting messages to the supported system  140 . 
       FIG. 1C  illustrates the supported system  140 , reliable transport  170  and OOM  100  after the supported system  140  has been restarted. Once the supported system  140  has been reinitialized, the supported system  140  and OOM  100  may subscribe to particular recovery channels via the reliable transport  170  to prepare for recovery by transmission of data from OOM  100  to the supported system  140 . The supported system  140  may also reestablish a connection to the upstream system  130  for the purpose of acknowledging orders that were pending when the supported system  140  experienced the outage. Both of these connections may be created through the reliable transport  170  via subscription. 
       FIG. 1D  illustrates the supported system  140 , reliable transport  170  and OOM  100  during the recovery process. Once dependencies have been checked and all are ready for recovery, the supported system  140  may receive the relevant recovery messages  141  from OOM  100  necessary to rebuild the order state of the supported system  140  prior to the outage. The recovery message  141  is a snapshot of the current order state of the supported system  140  that is transmitted by OOM  100  over the reliable transport  170  in response to a recovery request from the supported system  140  after it comes back up after the outage. Recovery messages  141  may contain unacknowledged orders at the time the outage occurred. If the supported system  140  receives an unacknowledged order, it may send an acknowledgement message  142  to the upstream system  130  to acknowledge the received message. As will be described below, the different status between these two groups of messages requires different treatment during recovery. When the order state of the supported system  140  has been recovered, then, as depicted in  FIG. 1E , the supported system  140  is ready to resume normal operation. 
       FIG. 1F  illustrates the supported system  140  as it begins to resume normal operation. The supported system  140  re-subscribes to the appropriate reliable transport  170  topics to receive upstream message data  135 . After sending an indicator to the upstream server  130  that it is prepared to receive new messages, the supported system  140  begins receiving upstream message data  135  from the upstream server  130 , and normal operation may resume, as depicted in  FIG. 1F . 
       FIG. 2  is a high level representation of an illustrative passive real-time order state replication and recovery system instance or OOM  100 . OOM  100  may be operational with numerous other general purpose or special purpose computing systems, environments or configurations. A single OOM  100  may be implemented utilizing one or more computing systems of varying configurations. For instance, each OOM  100  may be contained within a single computing system (as depicted). Alternatively, multiple OOMs  100  may be operated on a single hardware platform. In other embodiments, a single computer system may house multiple processors, each hosting separate OOMs  100 . Any desirable configuration may be utilized as required. Each computing system preferably includes computing components for executing computer program instructions and processes. 
     As shown in  FIG. 2 , OOM  100  is illustrated in the form of a special purpose computer system. The components of OOM  100  may include (but are not limited to) one or more processors or processing units  200 , a system memory  205 , and a bus  218  that couples various system components including memory  205  to processor  200 . 
     Bus  218  represents one or more of any of several types of bus structures, including a memory bus or memory controller, a peripheral bus, an accelerated graphics port, and a processor or local bus using any of a variety of bus architectures. By way of example, and not limitation, such architectures include Industry Standard Architecture (ISA) bus, Micro Channel Architecture (MCA) bus, Enhanced ISA (EISA) bus, Video Electronics Standards Association (VESA) local bus, and Peripheral Component Interconnect (PCI) bus. 
     The processing unit  200  processes and executes computer program instructions stored in memory  205 . Any suitable programming language can be used to implement the routines of particular embodiments including C, C++, Java, assembly language, etc. Different programming techniques can be employed such as procedural or object oriented. The routines can execute on a single OOM  100  or multiple OOMs. Further, multiple processors  200  may be used. 
     System memory  205  can include computer system readable media in the form of volatile memory, such as random access memory (RAM)  230  and/or cache memory  232 . OOM  100  may further include other removable/non-removable, volatile/non-volatile computer system storage media. By way of example only, storage system  234  can be provided for reading from and writing to a non-removable, non-volatile magnetic media (not shown and typically referred to as a “hard drive”). Although not shown, a magnetic disk drive for reading from and writing to a removable, non-volatile magnetic disk (e.g., a “floppy disk”), and an optical disk drive for reading from or writing to a removable, non-volatile optical disk such as a CD-ROM, DVD-ROM or other optical media can be provided. In such instances, each can be connected to bus  218  by one or more data media interfaces. As will be further depicted and described below, memory  205  may include at least one program product having a set (e.g., at least one) of program modules that are configured to carry out the functions of embodiments described in this disclosure. 
     OOM software module  220  may be stored in memory  205  by way of example, and not limitation, as well as an operating system, one or more application programs, other program modules, and program data. The OOM software module  220  is executable by processor  200  to generally carry out the functions and/or methodologies of embodiments described herein. In particular, OOM software module  220  may be tasked with receiving and processing upstream and downstream message data  135 ,  145 , thereby maintaining the same order state as the supported system  140 . 
     OOM  100  may also communicate with one or more external devices  214  such as a keyboard, a pointing device, a display  224 , etc.; one or more devices that enable a user to interact with OOM  100 ; and/or any devices (e.g., network card, modem, etc.) that enable OOM  100  to communicate with one or more other computing devices. Such communication can occur via Input/Output (I/O) interface(s)  210 . 
     The network interface device  215  may provide the OOM  100  with access to a network, which may be a wireless or wired connection. The network may be, for example, the Internet, a corporate intranet, or any other computer network through which OOM  100  may connect to or otherwise communicate with other computers and databases, such as (but not limited to) other OOMs  100 , the upstream system  130 , the downstream system  150 , the messaging pipeline or reliable transport  170 , and any other computerized systems, networks or databases for specialized information that may be necessary for implementation of the passive real-time order state replication and recovery system and method. 
       FIG. 3  is an illustrative sequence of steps for passive real-time order state replication. One or more OOM instances  100  may be initialized to provide order state support for the supported system  140  (step  300 ). Once all systems are ready, the upstream server  130  may begin sending upstream messages  135  to the supported system  140  and OOM  100  using the reliable transport  170  (steps  305 ,  310 ). The supported system  140  receives and processes the upstream messages  135  to update the order state of the supported system, and upon processing, sends an acknowledgement  155  to the upstream system  130  and OOM  100  (step  320 ). The supported system  140  may also need to output downstream messages  145  in response to received upstream messages  135 . Therefore, a downstream message  145  may be sent to a downstream server  150  via the reliable transport  170 , which OOM  100  receives as well (step  325 ). The OOM instances  100 , subscribed to the same topics as the supported system  140 , receive and process the upstream messages  135  and downstream messages  145  via reliable transport  170  (step  310 ). Each OOM instance  100  continuously updates its own order state in accordance with the received upstream messages  135 , downstream messages  145  and acknowledgements  155 , maintaining in real-time the same order state as the supported system  140  (step  315 ). 
       FIGS. 4-5  illustrate a preferred sequence of steps for real-time order state recovery. Referring to  FIG. 4 , the supported system  140  may experience an outage, and stops receiving upstream messages  135  and sending downstream messages  145  (step  400 ). At the time the outage occurred, any number of upstream messages  135  may have gone unacknowledged by the supported system  140 . These “pending” upstream messages  135  will need to be acknowledged during the recovery process. The upstream system  130  may cease sending upstream messages  135  to the supported system  140  and OOM  100  when it detects the supported system  140  is offline (step  405 ). The reliable transport  170  may have a built-in heartbeat-based mechanism to detect the connected or disconnected status of the supported system  140 . 
     The supported system  140  may re-initialize, and both the supported system  140  and OOM  100  subscribe to appropriate topics via the reliable transport  170  for the purpose of sending and receiving recovery messages  141  and acknowledgements  142  (step  410 ). Subsequently, it is determined whether all dependencies are ready for recovery (step  415 ). That is, whether the system and infrastructure necessary for recovery (e.g., OOM  100 , supported system  140 , upstream server  130 , reliable transport  170 , etc.) are ready for recovery. If one of the dependencies is not ready for recovery, then the initialization of the supported system  140  is terminated and later restarted to retry recovery (step  420 ). If all dependencies are ready, recovery via OOM  100  is initiated (step  425 ). 
     Referring now to  FIG. 5 , the order state of the supported system  140  is restored by sending the relevant recovery messages  141  from OOM  100  to the supported system  140  (step  500 ). The recovery messages  141  contain the order states, which are based on messages  135 ,  145  and  155  received by OOM  100 , to duplicate the order state of the supported system  140 . During recovery, the supported system  140  may send acknowledgements  142  for messages  135  that were unacknowledged but already received (e.g., unacknowledged orders) (step  505 ). This information is included inherently in OOM  100 , because it receives acknowledgement messages  155 , in addition to upstream messages  135  and downstream messages  145 . Therefore, messages received by OOM  100  that do not have a corresponding acknowledgement message  155  (e.g., unacknowledged orders) will require transmission of an acknowledgement  142 . Once the order state of the supported system  140  has been restored, the appropriate subscriptions may be created using the reliable transport  170  (step  510 ), and standard operation may be resumed by enabling the receipt of new order state information, allowing the supported system  140  to receive and send messages and acknowledgements (step  515 ). 
       FIG. 6  illustrates some nuances of the recovery phase. Messages may take a variety of forms, including pending “new” messages, “replacement” messages and “cancellation” messages. Pending new messages are messages that have been received previously but have gone unacknowledged (presumably because the outage occurred before they could be processed). Pending replacement messages are messages that are meant to replace another message already extant within the supported system  140 . Pending cancellation messages are messages that are sent to cancel an extant message within the supported system  140 . Each message may be a combination of these forms of messages, and accordingly, must be handled differently in order to accurately rebuild the order state of the supported system  140  prior to the outage. 
     First phase recoveries are illustrated in scenarios  600 - 615 . All first phase recovery messages are already acknowledged, meaning they are not “new” messages, as indicated by the “false” markers in column  640 . Scenarios  600 - 615  are therefore the messages that are recovered during the first phase. Scenario  600  summarizes messages that are neither new, pending replacement nor pending cancellation. These messages are recovered normally and require no additional steps. Scenario  605  summarizes messages that are not new or pending replacement, but are pending cancellations. These messages are therefore recovered normally, but during the second phase, a special cancellation acknowledgement  155  is sent from support system  140  to the upstream system  130 , as would have been performed normally if the outage had not occurred. Scenario  610  summarizes messages that are not new or pending cancellation, but are pending replacement. These messages are recovered normally, but during the second phase, the replacement is attempted. Subsequently, either a successful acknowledgement  155  or a rejection acknowledgement  155  is sent by the supported system  140  depending on whether the replacement attempt was successful. Scenario  615  summarizes messages that are not new, but are pending replacement and pending cancellation. These messages are recovered normally during the first phase, but during the second phase, the replacement and cancellation actions are attempted. The supported system  140  may then send two sets of acknowledgements  155 , the first indicating whether the replacement attempt was successful, and the second indicating whether the cancellation attempt was successful. 
     Second phase recoveries are illustrated in scenarios  620 - 635 . All second phase recovery messages were pending but not yet processed when the outage occurred, and therefore remain unacknowledged. Scenario  620  summarizes pending “new” messages that are not pending replacement or pending cancellation. These messages are received by the supported system  140  as simple new messages, processed by the supported system, and an acknowledgement  155  is sent to the upstream system  130 . Scenario  625  summarizes pending “new” messages that are not pending replacement, but are pending cancellation. These messages are received by the supported system  140  and an acknowledgement  155  is sent in response. Subsequently, the cancellation is attempted, and a second acknowledgement  155  is sent indicating whether the cancellation was successful or unsuccessful. Scenario  630  summarizes pending “new” messages that are pending replacement, but are not pending cancellation. These messages are received by the supported system  140  and an acknowledgement  155  is sent in response. Subsequently, the replacement is attempted and another acknowledgement  155  is sent indicating whether the replacement was successful or unsuccessful. Scenario  635  summarizes pending “new” messages that are pending replacement and pending cancellation. These messages are received by the supported system  140  and an acknowledgement  155  is sent in response. Subsequently, the replacement and cancellation are attempted, and two acknowledgements  155  may be sent, one indicating whether the replacement was successful or unsuccessful, and the other indicating whether the cancellation was successful or unsuccessful. 
       FIGS. 7A-7D  are high level representations of a supported system  140  utilizing multiple OOMs  100   a - 100   d . Multiple OOMs  100   a - 100   d  may be used in conjunction to provide any desired benefit, such as redundancy or more specialized support. The supported system  140  may select an OOM instance  100   a - 100   d  to recover from based on the following process. Referring to  FIG. 7A , after the supported system  140  has experienced an outage and upon recovery, the supported system  140  sends a “poll” request  700  to recover lost data to all OOM instances  100   a - 100   d . Referring to  FIG. 7B , when each OOM instance  100   a - 100   d  receives the “poll” request, each OOM instance  100   a - 100   d  sends a response  705   a - 705   d  to the supported system  140 . The responses  705   a - 705   d  may have varying travel times to reach to the supported system  140 , based on the network topography. As a result, some may be received before others. Referring to  FIG. 7C , the supported system  140  sends a command to initiate data recovery  710  to, for example, OOM instance  100   a , because the response from OOM instance  100   a  happened to be received first by the supported system  140 . Finally, referring to  FIG. 7D , OOM instance  100   a  begins sending recovery data  715  to the supported system  140 . Recovery then proceeds as described above. 
     This application was described above with reference to flow chart illustrations and/or block diagrams of methods, apparatus (systems) and computer program products according to one or more embodiments. It is understood that some or all of the blocks of the flow chart illustrations and/or block diagrams, and combinations of blocks in the flow chart illustrations and/or block diagrams, can be implemented by computer program instructions. The computer program instructions may also be loaded onto the computing system to cause a series of operational steps to be performed on the computer to produce a computer implemented process such that the instructions that execute on the computer provide processes for implementing the functions/acts specified in the flowchart and/or block diagram block(s). These computer program instructions may be provided to the CPU of the computing system such that the instructions, which execute via the CPU of the computing system, create means for implementing the functions/acts specified in the flowchart and/or block diagram block(s). 
     These computer program instructions may also be stored in a computer-readable medium that can direct the computing system to function in a particular manner, such that the instructions stored in the computer-readable medium implement the function/act specified in the flowchart and/or block diagram block or blocks. Any combination of one or more computer usable or computer readable medium(s) may be utilized. The computer-usable or computer-readable medium may be, for example (but not limited to), an electronic, magnetic, optical, electromagnetic, infrared, or semiconductor system, apparatus, device, or propagation medium. More specific examples (a non-exhaustive list) of the computer-readable medium include the following: an electrical connection having one or more wires, a portable computer diskette, a hard disk, a random access memory, a read-only memory, an erasable programmable read-only memory (e.g., EPROM or Flash memory), an optical fiber, a portable compact disc read-only memory, an optical storage device, a transmission media such as those supporting the Internet or an intranet, or a magnetic storage device. Any medium suitable for electronically capturing, compiling, interpreting, or otherwise processing in a suitable manner, if necessary, and storing into computer memory may be used. In the context of this disclosure, a computer-usable or computer-readable medium may be any medium that can contain, store, communicate, propagate, or transport the program for use by or in connection with the instruction execution system, apparatus, or device. The computer-usable medium may include a propagated data signal with the computer-usable program code embodied therewith, either in base band or as part of a carrier wave. The computer usable program code may be transmitted using any appropriate medium, including (but not limited to) wireless, wire line, optical fiber cable, RF, etc. 
     Having described and illustrated the principles of this application by reference to one or more preferred embodiments, it should be apparent that the preferred embodiment(s) may be modified in arrangement and detail without departing from the principles disclosed herein and that it is intended that the application be construed as including all such modifications and variations insofar as they come within the spirit and scope of the subject matter disclosed.