You Cannot Have Exactly-Once Delivery

"People often bend the meaning of “delivery” in order to make their system fit the semantics of exactly-once, or in other cases, the term is overloaded to mean something entirely different. State-machine replication is a good example of this. Atomic broadcast protocols ensure messages are delivered reliably and in order. The truth is, we can’t deliver messages reliably and in order in the face of network partitions and crashes without a high degree of coordination. This coordination, of course, comes at a cost (latency and availability), while still relying on at-least-once semantics."

This reminds me of the problem of reliable data transfer over an unreliable network. It's theoretically impossible, but TCP is still a practical, useful abstraction.

When communication is reestablished, that issue can be sorted out. See "two-phase commit".

Because it restates that, without timeouts, in theory, you could be waiting forever for a message to arrive. It's formally proved and whatnot, but it's not a super-surprising result when you restate it in common terms.

For message delivery, it seems that FLP says that you need to give up at-least-once, because at-least-once ostensibly requires an infinite number of retries. I'm not sure how it would apply to at-most-once.

Exactly-once is, so far as I can tell, a bit of a strawman. But it's a strawman that we all fall for, the first time we break stuff into different systems.

As a disclaimer, I am still wrapping my head around this stuff. Don't rely on me. Not even if your name is Salvatore.

DD: I myself shared many of these misconceptions, so I try not to demean or dismiss

EE: but rather educate and enlighten, hopefully while sounding less preachy than that just did.

FF: I continue to learn only by following in the footsteps of others.

In other words, of course you can deliver something exactly one time. Just not every time.

I wrote about this issue with a lot of APIs some time ago. When a connection is broken unexpectedly, the protocol simply has no way to recover the state of play and there's a risk of duplicate transactions popping up.

http://blog.hackensplat.com/2014/07/is-your-api-broken.html

Here is how SSB achieves EOIO:

- initiator establishes intent to communicate using BEGIN DIALOG[1] statement (SSB dialogs are the equivalent of a durable, long lived, TCP session). This creates the necessary state in the database by creating a row in sys.conversation_endpoints, with initial send_sequence_number 0.

- sender issues SEND[2] statement. The message is assigned the current send_sequence_number (0), the sys.conversation_endpoint send_sequence_number is incremented to 1, and the message is inserted in sys.transmission_queue

- after commit of SEND transaction a background transmitter reads the message from sys.transmission_queue, connects to destination, delivers the message over wire (TCP).

- target reconstructs message from wire, in a single transaction creates a row in it sys.conversation_endpoints (receive_sequence_number is 0) and delivers the message into the destination queue

- after commit of the message delivery, the target constructs an acknowledgement message and sends it back to initiator over the wire

- the sender gets the acknowledgement of message 0 and deletes the message from sys.transmission_queue

- sender may retry delivery periodically if it does not receive the ack

- target will send back an ack immediately on receipt of a duplicate (message sequence number is less than current receive_sequence_number)

What this protocol achieves is idempotency of delivery, hidden from the messaging application. Database WAL ensures stability in presence of crashes. Eg. if the target crashes in the middle of enqueueing the message into destination queue then the entire partial processing of the enqueue is rolled back on recovery and next retry from sender will succeed. If target crashes after processing is committed but before sending the ack then on recovery the enqueue is successful and the next retry from initiator will immediately send ack an ack, allowing initiator to delete the retried message and make progress (send the next message in sequence). Note that there is no two-phase-commit involved.

Retries of unacknowledged messages occur for the dialog lifetime, which can be days, months, even years. Databases are good at keeping state for so long. SSB uses logical names for destination (from 'foo' to 'bar') and routing can be reconfigured mid-flight (ie. the location where 'bar' is hosted can be changed transparently). Long lived state and routing allow for transparent reconfiguraiton of network topologies, handle downtime, manage disaster (target is lost and rebuild from backups). Most of the time this is transparent to the SSB application.

Furthermore, the guarantees can be extended to application semantics as well. Applications dequeue messages using RECEIVE[3] statement. In a single transaction the application would issue a RECEIVE to dequeue the next available message, lookup app state perteining the message, modify the state, send a response using SEND[2], commit. Again WAL guarantees consistency, after a crash everything is rolled back and the application would go again through exactly the same sequence (the response SEND cannot communicate anything on the wire until after commit, see above).

So EOIO is possible.

One has to understand the trade offs implied. Something like SSB will trade off latency for durability. Applications need not worry about retries, duplicates, missing messages, routing etc as long as they are capable of handling responses comming back hours (or maybe weeks) after the request was sent. And application processing of a message is idempotent (RECEIVE -> process -> crash -> rollback -> RECEIVE -> re-process) only as long as the processing is entirely database bound (update states in some app tables, not make REST calls). Yet such apps are not unusual: they use some database to store state and communicate with some other app that also uses a database to store state. Unlike most messaging systems, SSB stores the messages in the database thus achiving WAL consitency along with the app state. Many SSB applications are entirely contained in the database, the code itself is contained. They use SSB activation [4] to react to incoming message, without keeping any state in memory.

In SSB both the initiator and the sender are monolitic, SMP systems (not distributed). Together the two form a distributed system. It trades of availability over partitioning, but one has to understand how this trade off occurs. In case of parittioning (target is unreacheable) the application continues to be available locally (the SEND statement succeeds). If the netowrk paritioning is not resolved over the lifetime of the dialog, then the applicaiton will see an error. If the paritioning is resolved then message flows resumes and the applicaiton layer responses start showing up in the queue. Again, activation and durable state make this easy to handle, as long as a latency of potentially days makes sense in the business. Shorter lifetimes (hours, minutes) are certainly possible and in such cases, if network partitioning is not resolved in time, the timeout error will occur sooner.

  [0] https://msdn.microsoft.com/en-us/library/bb522893.aspx
  [1] https://msdn.microsoft.com/en-us/library/ms187377.aspx
  [2] https://msdn.microsoft.com/en-us/library/ms188407.aspx
  [3] https://msdn.microsoft.com/en-us/library/ms186963.aspx
  [4] https://technet.microsoft.com/en-us/library/ms171617.aspx

I believe what it really comes down to is that people new to distributed processing think they want "exactly once delivery", but later they (hopefully) learn they really want "exactly once processing". For example:

> We send him a series of text messages with turn-by-turn directions, but one of the messages is delivered twice! Our friend isn’t too happy when he finds himself in the bad part of town.

This is easily resolved by amending each message with an ordinal indicator, e.g. "1. Head towards the McD", "2. Turn left at the traffic lights", etc. The receiver then dedupes and the instructions follow "exactly once processing". Processing messages exactly once is a "business rule" and the responsibility for doing so lies in the business logic of the application.

This example also brings up another typical point of confusion in building distributed systems: people actually want "ordered processing", not "ordered delivery". The physical receiving order does not matter: your friend will not attempt to follow instruction #2 without first following instruction #1. If instruction #2 is received first, your friend will wait for instruction #1.

It's also important to note that the desired processing order of the messages has nothing to do with the physical sending order: I could be receiving directions to two different places from two different people, and it doesn't matter what order they are sent or received, just that all the messages get to me! A great article covering these topics in more detail with other real world examples is "Nobody Needs Reliable Messaging" [1].

I think it is useful to try understand why new distributed system builders run into these difficulties. I suspect they try to apply local procedure call semantics to distributed processing (a fool's errand), and message queue middleware works well enough that a naive "fire and forget" approach is the first strategy they attempt. When they subsequently lose their first message (hopefully before going into production), it's natural to think in terms of patching (e.g. distributed transactions, confirmation protocols, etc.) rather than to consider if the overall design pattern is appropriate.

Oddly, there is at least one very well designed solution that addresses these challenges - the Data Distribution Service (DDS) [2] - but I almost never see or hear about it at any of my clients.

[1] http://www.infoq.com/articles/no-reliable-messaging [2] http://portals.omg.org/dds/

I mean if you used the blockchain as a message queue it could be pretty reliable, no?