At most once, exactly once, and at least once

When handling failure strategies, you need to start getting clever and make sure you have all edge cases covered. There are three approaches you can take for every request, because how you handle requests maps to message delivery semantics across nodes in distributed systems. In our example in Figure 14-3, the only guarantee you have is that your request has been executed at least once. If you are logging on to the system and the first logic node is so slow that the front-end node tries another one and succeeds with it, the worst-case scenario is that you log on twice and two sessions are created, one of which will eventually expire.

Similarly, if you are sending an SMS or an instant message, you might be happy with the at most once approach. If your system sends billions of messages a day, the loss of a few messages is acceptable relative to the load and the cost associated with guaranteed delivery. You send your request and forget about it. In our example with no single point of failure, when the front-end node sends the request to the logic node, it also immediately sends a reply back to the client.

But what if you were sending money or a premium rate SMS? Losing money, making the transfer more than once, or sending and charging for the same premium SMS multiple times because of an error will not make you popular. Under these circumstances, you need the exactly once approach. A request can succeed or fail. If failure is in your business logic, such as where a user is not allowed to receive premium rate SMSs, we actually consider the failed request to be successful, as it falls within the valid return values. Errors that should worry us are timeouts, software bugs, or corrupt state causing a process or node to terminate abnormally, leaving the system in a potentially unknown or undefined state. As long as you use the exactly once approach in a single node, abnormal process termination can be detected. As soon as you go distributed, however, the semantics of the request cannot be guaranteed.

The successful case is when you send a request and receive a response. But if you do not receive a response, is it because of the request never reaching the remote node, because of a bug in the remote node, or because the acknowledgment and reply of the successful execution got lost in transit? The system could be left in an inconsistent state and need cleaning up. In some systems, the cleanup is executed automatically by a script that tries to determine what went wrong and address the problem. In other cases, cleaning up might require human intervention because of the complexity of the code or seriousness of the failed transaction. If a request to send a premium rate SMS failed, a script could start by investigating if the mobile device received the remote SMS, if the user was charged for it, or if the request ever made it to the system. Having comprehensive logs, as we show in “Logs”, becomes critical.

A common pattern in achieving exactly once semantics with at most once calls is to use unique sequence numbers in the client requests. A client sends a request that gets processed correctly (Figure 14-4, part 1). If the response from the front-end node is lost or delayed, a timeout in the client is triggered. The client resends the request with the same identifier, and the logic node identifies it as a duplicate request and returns the original reply, possibly tagging it as a duplicate (Figure 14-4, part 2). You are still not guaranteed success, as the connectivity between the client and the server might not come up again. But it will work in the presence of transient errors.

Figure 14-4. Duplicate requests

This approach relies on idempotence. The term describes an operation that the user can apply multiple times with the same effect as applying it once. For example, if a request changes a customer’s shipping address, whether the system performs the request successfully once or multiple times has the same result, assuming the shipping address is the same in each request. Such a request can actually be executed multiple times because the side effects of any second or subsequent executions essentially have no observable effect. With our request identification scheme, though, the second and subsequent executions never occur.

Imagine a billing system for premium rate SMSs. You need to guarantee that if you charge the user, you will do so exactly once, and only after the SMS is received. An approach typically taken to guarantee this result is reserving the funds in the recipient account before sending the SMS. When reserving them,  the billing system returns a unique identifier. The SMS is sent, possibly multiple times. The charge is made only when the first report notifying that it has been delivered is received by the billing system. The unique identifier is then used to execute the payment and charge the account. Subsequent attempts to use the same identifier, possibly when receiving multiple copies of the same delivery report, do not result in additional charges. And if the SMS never reaches its recipient, the reserved funds are eventually released after timing out. The timeout also invalidates the identifier.

At most once, at least once, and exactly once approaches all have advantages and tradeoffs. While deciding what strategy to use, keep in mind that requests and the messaging infrastructure that underpins them are unreliable. This unreliability needs to be managed in the business logic and semantics of every request. The easiest to implement and least memory- and CPU-intensive approach is the “at most once” approach, where you send off your request and forget about it. If something fails, you have lost the request, but without affecting the performance of all of the other requests that succeeded. The “at least once” approach is more expensive, because you need to store the state of the request, monitor it, and upon receiving timeouts or errors, forward it to a different node. Along with higher memory and CPU usage, it can generate additional network traffic. Theoreticians will argue that the at least once approach cannot be guaranteed to be successful, as all nodes receiving the request can be down. We’ll leave them scratching their heads and figuring out what double and triple redundancy are all about. The hardest strategy is the “exactly once” approach, because you need to provide guarantees when executing what is in effect a transaction. The request can succeed or fail, but nothing in between.

These guarantees are impossible with distributed systems, since failure can also mean a request being successfully executed but its acknowledgment and reply being lost. You need algorithms that try to retrace the call through the logs and understand where a failure occurred to try to correct it or compensate for it. In some systems, this is so complex or the stakes are so high that human intervention is required.

Until now, we’ve said, “Let it crash.” Yes, let it crash, and no matter which of the three strategies you pick, put your effort into the recovery, ensuring that after failure, your system returns to a consistent state. The beauty of error handling and recovery in Erlang is that your recovery strategy will be the same when dealing with all of your errors, software, hardware, and network faults included. If you do it right, there will be no need to duplicate code in a process recreating its state after a crash or recovering after a network partition or packet loss.

Share nothing

The share-nothing architecture is where no data or state is shared. This could be specific to a node, a node family, or a cluster. Once you have addressed the underlying infrastructure, such as hardware, networks, and load balancing, share-nothing architectures can result in linearly scalable systems. Because each collection of nodes has an independent copy of its own data and state, it can operate on its own. When you need to scale, all you need to do is add more infrastructure and reconfigure your load balancers.

Figure 14-5, part 1 shows two front-end and two logic nodes. Using a login request, Client1 and Client2 send their credentials to initiate a session. This request is forwarded to one of the two front-end nodes using the load-balancing strategy configured in the load balancer. In our example, each node gets a request that it forwards to its primary logic node. These nodes each check the client credentials and create a session, storing the session state in a database.

Client1 now sends a new request right after the node storing its session data has crashed, losing everything (Figure 14-5, part 2). The front-end node forwards it to its standby logic node, which rejects the request because it is unaware of the session. The client, upon receiving an unknown session error, sends a new login request that is forwarded and handled by the second logic node. All future requests from this client should now be forwarded to the logic node containing the session. If they aren’t and the node that crashed comes up again, the client will have to log on again (Figure 14-5, part 3), and we just assume that the session in the standby node will eventually time out and be deleted.

Figure 14-5. Share-nothing architecture

As we don’t have to copy our session state across nodes, we get better scalability, because we can continue adding front-end and logic nodes as the number of simultaneously connected users increases. The downside of this strategy is that if you lose a node, you lose the state and all of the data associated with it. In our example, all sessions are lost, forcing users to log on again and establish a new session in another node. You also need to choose how to route your requests across nodes, ensuring that each request is routed to the logic node that stores its matching session data. This guarantees continuity after a node failure and recovery.

Share something

What do we do if we want to ensure that users are still logged on and have a valid session after a node failure? The share-something architecture, where you duplicate some but not all of your data, might address some of your concerns. In Figure 14-6, we copy the session state across all logic nodes. If a node terminates, is slow, or can’t be reached, requests are forwarded to logic nodes that have copies of the session data. This approach ensures that the client does not have to go though a login procedure when switching logic nodes. But it trades off some scalability, because the session data needs to be copied across multiple nodes every time a client logs in and deleted when the session is terminated. Things get even more expensive whenever a node is added to the cluster or restarts, because sessions from the other nodes might have to be copied to it and kept consistent.

Figure 14-6. Share-something architecture

The strategy just described is called share something because it is a compromise: you copy some, but not all of the data and state associated with each session. The strategy reduces the overhead of copying while increasing the level of fault tolerance. Let’s return to our e-commerce site. The session data is copied, so if a node is lost, the user does not have to log on again. However, the contents of the shopping cart are not copied, so upon losing a node, the users unexpectedly have their carts emptied. When a user is checking out and paying for the selected items, only those items in the active logic node’s shopping cart will appear.

We have been assuming all along that if the logic node crashed, all of its data would be lost. What if the shopping cart were stored in a persistent key-value store which, upon restart, was reread? Or what if a network partition occurred, or the node was just slow in responding and, as such, presumed dead? When the node becomes available, you need to decide on your routing strategy—namely, which of the two logic nodes receives new requests. And because you have two shopping carts, they now need to be merged (Figure 14-7, part 1), or one of them has to be discarded.

How is this done? Do you join all of the items? What if we had added an item to the shopping cart in the second node? Will we end up with one or two copies of the item? Or what if we had sent a delete operation to the second node, but the operation failed because the item was not there? How you solve these problems depends on your business and your risks. Some distributed databases, such as Riak and Cassandra, provide you with options. In our example, the node that crashed becomes the primary again, and we move the contents of the second shopping cart to it (Figure 14-7, part 2).

Figure 14-7. Network partitions with the share-something architecture

The Dynamo paper discussed in “Riak Core” describes the Amazon way of merging its shopping cart when recovering from failures. If, during the merge, there is uncertainty over the deletion of an item, it gets included, leaving the responsibility to the shopper to either remove it when reviewing the final order or return it for a refund if it actually gets shipped. How many times, upon checkout, have you found an item in your Amazon shopping cart you were sure you had deleted? It has happened to us a few times.

The share-something architecture is ideal for use cases where you are allowed to lose an occasional odd request but need to retain state for more expensive operations. We used a shopping cart example. Think of an instant messaging server instead. The most expensive operation, and biggest bottleneck, is users logging in and starting a session. The session needs to access an authentication server, retrieve the user’s contact list, and send everyone a status update. Imagine a server handling a million users. The last thing you want as the result of a network partition or a node crash is for a million users to be logging back on, especially when the system is still recovering from having sent 30 million offline status updates (assuming 60 contacts per user, of whom half are online).

One good solution is to distribute the session record across multiple nodes. What you do not share, however, are the status notifications and messages. You let them go through a single node with the “at most once” approach for sending messages in order to preserve speed. You assume that if the node crashes or is separated from the rest of the cluster through a network error, you either delay the delivery of the notifications and messages or lose some or all of them. How many times have you sent an SMS to someone close to you, only for them to receive it hours (or days) later, or never at all?

Share everything

Share-nothing and share-something architectures might work for some systems and data sets, but what if you want to make your system as fault tolerant and resilient as possible? While it is not possible to have a distributed system where losing requests is not an option, you might be dealing with money, shares, or other operations where inconsistency or the risk of losing a request is unacceptable. Each transaction must execute exactly once, its data has to be strongly consistent, and operations on it must either succeed or fail in their entirety. Although you can get away with not receiving an SMS or instant message, finding an equity trade you thought had been executed missing in your portfolio or money missing from your bank account is indefensible. This is where the share-everything architecture comes into the picture. All your data is shared across all of the nodes, any of which might, upon hardware or software failure, take over the requests. If there is any uncertainty over the outcome of a request, an error is returned to the user. When things go wrong, they have to be reconciled after the fact. For example, if you try to withdraw from multiple ATMs more funds than you have in your account, you get the money, but then later the bank penalizes you for overdrawing your account. But with no single point of failure, using redundant hardware and software, the risk for this error should be reduced to a minimum.

In Figure 14-8, we duplicate the sessions and shopping cart contents in two logic nodes, each handling a subset of clients and duplicating the session state and shopping cart to the other nodes. If a node terminates, the other one takes over. Should the node recover, it will not accept any requests until all of the data from the active node has been copied over and is consistent with other nodes.

Figure 14-8. Share-everything architecture

We call this primary-primary replication. This contrasts with primary-secondary replication, where a single primary node is responsible for the data. The secondary nodes can access the data, but must coordinate any destructive operations such as inserts or deletes with the primary if they wish to modify the data. If the primary is lost, either the system stops working entirely, or it provides a degraded service level where writes and updates are not allowed, or one of the secondaries takes over as primary.

The share-everything architecture is the most reliable of all data-sharing strategies, but this reliability comes at the cost of scalability. It tolerates the loss of nodes without impacting consistency of data, but if some nodes go wrong, it also loses availability. This strategy is also the most expensive to run and maintain, because every operation results in computational overhead and multiple requests across the network to ensure that the data is kept replicated and consistent. Upon restarting, nodes must connect to a primary and retrieve a copy the data to bring them back in sync with the primary node, ensuring they have a correct and current view of the state and the data.

Although share-everything architectures do not necessarily require distributed transactions across nodes, you will need them when dealing with data such as money or shares you cannot afford to lose. This contrasts with the requirements we saw for messaging, where duplicating the messages through eventual consistency will greatly reduce the risk of them getting lost if you lose a node, but with no strong guarantee that you will never lose a message.

When you have decided on your data-sharing strategy, you need to go through each request in your API, trace the call, and try to map everything that can go wrong. Within the node itself, for synchronous calls, you need to consider behavior timeouts and abnormal process termination. If dealing with asynchronous messaging, ensure that message loss (when the receiving process has terminated) is handled correctly. Across nodes, you need to consider network errors, partitions, slow nodes, and node termination. When you’re done, pick the recovery strategy that best suits the particular calls. This needs to be done for every external call, and will often result in a mixture of the three recovery strategies, depending on the importance of the state change.