>> Hi, everyone. My name is Tomek Masternak,
I'm an engineer on the particular software,
when I build distributed systems and
platform before building distributed systems.
In this presentation, I will be talking about,
checking safety in Exactly-Once,
which is another library that
I built together with my friend Szymon Pobiega.
Let me start with describing the problem and
the challenge distributed system builders face,
when they use messaging infrastructure.
What is currently at the factor standard,
both when it comes to the infrastructure which is
available On-prem or in the Cloud,
is At-Least-Once message delivery guarantee.
That changed quite sometime in the past,
before we used to have a two-phase commit
available and distributed transactions.
That's pretty much no longer the case
and currently At-Least-Once message delivery
is that reality that everyone has to work with.
Now, At-Least-Once message delivery
can be challenging from the system builder perspective
especially when writing logic for handling messages that
arrive or are being pushed from
the messaging infrastructure to the receivers.
Let us go through our hypothetical situation just to
see what kind of situations I'm talking about.
Let's assume that there are some messages
in flight one, two and three and that is the order
in which they are stored in the
messaging infrastructure or in the queue.
Because of the way,
how the At-Least-Once message delivery
works is perfectly possible for
a message to be read, delivered, or
delivered multiple times to the receiver.
This is usually a simple consequence of
the fact that messaging infrastructure will
not consider a message to be delivered
until it gets an acknowledgment from the receiver.
Whenever the acknowledgment does not arrive
or does not arrive in a given period of time,
the message will reappear in the queue and
will be processed or delivered once again.
From the processing perspective,
what is possible is that some messages
get duplicated in our situation, it's message two.
What's also possible, is reordering and this usually
happens when a messaging infrastructure has
some delivery or processing timeout.
Whenever our messages beat out by the receiver,
the timeout starts and if it reaches a threshold,
that message will reappear in the queue
one more time and potentially in the meantime,
other messages might be successfully processed,
so the effective ordering of processing that the receiver
had will be different than the original ordering.
Finally, both of those situations can happen
independently at the same time,
so via BN situation is perfectly fine for the messages
to reorder and it's perfectly fine for messages to
be duplicated and now that can be
a challenge from the system builder perspective.
A very common and standard solution to
that situation is item potency,
which basically means that whenever
a duplicate of the message is received,
the end result of processing that message should be the
same as for the first processing of that message.
Very often this is understood or this boils down
to the message processing logic being deterministic.
This is often the case in remote procedure
call type of systems when there is
only a single message in flight
or a single call in flight and that call will be
repeated over and over again until
a successful acknowledgment comes in
from the remoting location.
Now, that might be not enough if there are
more than one messaging flight and this is
very often the case when we are building
and modeling business processes.
This is a very simple example which
shows worldwide what might be the problem.
What we are looking at here is a simple
sequence diagram showing the behavior in
a system which models
or implements shooting range game.
There is a handlers shooting range
which is responsible for holding,
a piece of spade which is
the current target position of the target.
The players can attempt to fire
at the given position and the shooting range
is responsible for sending back a response
saying whether the attempt was successful or fail.
So what we can see here is a situation
in which we have a shooting range
and the shooting range is set to target position 42
and then we have a fire that fires at position 42.
The response from that call is a hit.
After that, the target is being moved to position one
and finally, an interesting situation happens
because we get a duplicate and duplicated fire
at message which was already processed once,
but then it's being reprocessed one more time.
Logic is deterministic but the response that we've had is a miss.
What happened in that scenario is that we
have a single logical message being delivered twice
and each of the processing of this
message results in a contradictory result,
contradictory side effect, which is usually
far from what we are expecting
from the systems that we're building.
The obvious problem that we have here is that,
even though the logical is deterministic,
the state changed significantly in between
the first processing and the duplicate
which resulted in that behavior.
What it shows is that the deterministic logic is not enough,
we also need to think about the state.
Either we need to capture the side effects
and basically we'll do them whenever a duplicate
comes in or we need to be
able to get a handle at the historical state,
as it was when the message first arrived.
What we are assuming in the library is that
the distributed system that we are talking
about is field of bunch of handlers,
each handler has a separate dedicated piece of state.
That handler is sole holder and
the only possibility to communicate
between the handlers is via sending messages.
What each handler is doing is that it
picks up the message from the input queue,
executes the business logic,
which results in state update
and puts a message in the output queue.
The main idea or goal of the library was, as stated here,
to make sure that the observable side effects
correspond to some serial execution of
input messages with atomic commit guarantee
between the business storage and the output queue.
Basically what we wanted to make sure is
that whenever a duplicate comes in,
the side effects that we'll produce on the state
and in terms of sending of the output messages,
is the same as if there was a single processing and
the ordering is corresponding to some possible serial execution.
The implementation of the library is based on the Azure Stack.
If they run in Azure Function host,
which basically means now the whole will be messaging
infrastructure available for Azure functions
is available there as well.
The state is stored in Azure Cosmos DB.
Specifically, what do we rely on in terms of
Azure Cosmos DB is that it
provides optimistic concurrency control,
which is pretty important or the key part
of the approach that we are taking.
We store our infrastructural library-driven data
in a separate logical partition,
which also comes with a consequence which is that
there is no atomic transactions between
the business state and the infrastructural state.
We assume that the users are using
Azure Cosmos DB with session consistency model.
Finally, high performance scenarios are out of scope.
It's not something that we focused on.
Good enough performance, whatever that means, was our goal.
Now, the general idea of the algorithm is as follows.
Whenever a message comes in,
we check whether we already processed it.
If not, we load the state.
We run the business logic
and we do not send out the output messages,
but we capture them and store in this form.
Now, we attach a single correlation ID
with the business piece of data,
and then we commit it to the business logic or partition.
This is where we do the optimistic concurrency trade.
We do that only if that entity did not
change since we loaded it from the store.
If that is successful,
this is basically the point of no return.
The atomic commit protocol, once that is done,
the both parts, which is the business state which is
already committed and the output messages
will be committed eventual.
Now, an important thing here to
notice is that during some failure situations,
what can happen is that the output messages
might have been sent out multiple times.
However, because the duplicates can
happen with At-Least-Once message delivery already,
we can say that we are hiding behind that failure mode.
Basically, because the duplicates can be already there,
we can produce duplicates because
whoever is processing them should be
and is expected to cope with that situation.
That was the general idea
about the problem and the solution
that we seek and wanted to implement.
Now I wanted to talk about the TLA+ Model
and how we use that model to validate the safety.
The goals of the model was to make sure that
our algorithm works not only in
the happy path but also on in the things to implement.
As we all know, it's very easy to
show that the algorithm sometimes works,
but it's actually very tricky to figure
out all the further case scenario.
We wanted to make sure that we have some validation,
some model checking down to make sure that
we actually are convinced ourselves that
the algorithm does what we wanted it to do.
Specifically what we wanted to check was the safety properties.
Those safety properties were described as atomic commit.
Basically we wanted to make sure
that when we commit to the business logic store we
eventually push out the changes to the output queue,
and also made changes to be the input queue,
and also we wanted to make sure that
the side effects are consistent in a way that,
even for duplicates, there is only one change
in the business state and the output messages
and the business state change were made over
the same version of the business state, which I think
will be a bit clearer when we get to the formulas.
Finally, we wanted to make sure that everything
works as expected even without atomic transaction
between different logical partitions.
Because as I stated,
we start the business logic and
the infrastructural data in a separate partitions.
Obviously, there were most of the things that were
in the implementation did not make it to the model,
and specifically we did not model
the cleanup logic which is responsible
for clearing the infrastructural data
that is stored by the library.
We did not model the model resistance that are used
to lower the chance of concurrency and collisions,
optimistic concurrency check failures basically.
We also did not model the exponential backlogs
and processing retries which are there
to make the library a bit
more performing in the failure scenarios.
Let's look at the specification.
This is the main part of
the process that models the single handler.
Most of the labels are folded
and that shows the main structural specification.
What we have here is the infinite while loop
and in that loop the process is trying to
input message forms the input queue.
Then it checks by looking at the transaction field on the state,
whether it needs to redo some of the transactional steps to
basically push out the changes that
are still there and not pushed to the output queue.
After doing that, we check
whether we already processed the message.
If we did not process the message,
we model the business logic execution
capturing the side effects storing them,
the optimistic concurrency commits
to the business store and then pushing
out the transaction until it's finished.
The main reason why we use plus calc was that
this syntax was more familiar for us and secondly,
that it actually corresponds pretty well
when it comes to the implementation that was done in C#.
We were able to look at the code side-by-side
and the specification corresponded
pretty well with what we could see on
the screen looking at the real implementation.
When it comes to state modeling,
there wasn't that much happening.
Two important notices is
that the input queue was modeled as a set,
but that set had a records with two fields,
One which was the logical message ID
and duplicate message ID.
The way how we model the duplicates was
basically by starting off with duplicates already put queue.
When it comes to the business stack modeling,
we were a starting a sequence which was a history of
snapshots of the states as they
were throughout the execution of the algorithm,
and that was pretty useful when writing the invariants.
We also modeled the diversion
for the optimistic concurrency check.
All the other bits were used to model
the infrastructural storage
and the output to which was also set.
Another interesting thing is that we did not use is
a default termination formula.
What we said was that
the processes were running in the infinite loops.
And the power termination of
the algorithm was a situation in which
all the processes were in the lock-in
message and the input queue was empty.
Basically, the termination was modeling
the situation in which the input cue is trained
and none of the handler is doing anything in specific.
But waiting for anything to appear in that input.
Finally, the safety invariants.
This is what we ended up when it comes to safety.
That basically models or expresses the safety property,
which makes sure that for any logical message
there is AtMostOneStateChange happening to
the business store and most once message
being produced and sent to the output cue.
Those are the first two parts,
one on the top and one in the middle.
Finally, the one which is called consistent state
and output says that if the message is fully processed,
the version on which the diversion for the business state
and the output message should be the same.
Basically we make sure that those two pieces of side-effects
are consistent in a way that they operated and were
used by a business logic operating on
the same version of the business set.
Another interesting bit is how we modeled failures.
What we ended up was a combat role,
but basically was a single instruction
which was filled two main loop,
which basically meant that model
the situation in which either the process
restarts or some exception is being
thrown and we basically should have the top of
the stack and go back looking in that infinite fail loop.
The way that we used it was that whenever
a failure was possible or we wanted to model at failure,
we would do either or expression.
For instance, in the commit state macro we can see that if
the concurrent optimistic concurrency check results in intro,
we either commit right to the store or we fail.
That was a general approach
to monitoring those kinds of failures.
Okay, so now a few words about the results.
There were expected and unexpected results.
When it comes to the expected results,
those were obviously bugs to be
found in various parts of the algorithm.
Specifically in the post-failure commits and also
in the logic that is signs, the transaction id.
As I said, that was something that we
wanted to get from the model check-ins.
So that's why I called it expected.
I don't want to deprecate the value of that
because that was exactly what we want.
However, there were also unexpected bits.
I would say that the main unexpected bit was
the fact that when we started off writing the model,
we realized that we don't really
understand what is it that we are actually doing.
So it actually forced us to distill the algorithm to
its essentials and to make sure
whether the assumptions that
we are making which are pretty important and
also what are the most important steps in that algorithm.
As a follow-up of doubt, we realized that there are
actually some extensions that we could also use.
For instance, I already said that one of
the main assumptions that we are making is
that we don't need any kind of
a preparer phase for the output queue,
because we can always commit to that output queue.
We can always send a message there.
The duplicates can happen there because
there's already that failure mode,
which we need to cope with.
But what we realized is that there might be
some other resources that are always able to commit.
So for instance, if we have
some write once store, so for instance,
a blobs store with a [inaudible] blobs that can be
written only once that could be
an expression to that algorithm as well.
That could be a supported side effect
that we could apply the same way
as we are sending out the messages.
Finally, what we realized is that we did not know
the cost of [inaudible] well enough and all
the memory model implications.
So we had to go back to the drawing board,
and understand what doesn't mean that we are
running in a session memory model.
Also that there are no atomic transactions
between the logical preparations.
Finally, another unexpected thing
was that in order to create a meaningful model,
we had to understand what are
the concurrency characteristics of the algorithm.
What are the parts that can happen
independently and concurrently?
But even more importantly, we had to learn
what are the failure modes or what are
the failure situations that may happen in
different parts of the system as well.
So it actually forced us to understand the technology
that we're using are better than we knew it beforehand.
So we did the validation, we did the model checking.
An interesting note here is that whenever there was a bug
in a specific location or in the algorithm,
we were able to find it on very small models.
And by very small models,
I mean a single message with a single duplicate
and a single process. Almost that.
In that sense, the running time of model checking
was never a problem for us because whenever
there were bugs, we would get a feedback immediately.
Now that being said, after we settled on
an algorithm and we checked it with a smaller model.
We also went with a bigger one.
The bigger that we did was
two processes with six messages in the input cue,
those are numbers in terms of
number of space that we ended up with
and that was running roughly two-and-a-half hours
on a laptop, my personal laptop, that I would consider
a high-end laptop based on current standards.
That's it that I prepared for you in this presentation.
If anyone wants to know more about
the work that we did around TLA class and the library.
We are posting about that at exactly-once.github.io.
I'm also available on Twitter at Masternak.
If anyone wants to chat, please reach out to me.
I'd really like to share ideas. Thank you very much.