Testing distributed-store algorithms

This is a follow-up to my post on a datastructure for storing collections in Riak.
While I have been planning this follow-up, Kresten Krab Thorup has actually gone ahead and implemented the algorithm (on GitHub; see src/riak_column.erl) — which suites me nicely, as I haven't written a single line of implementation and am not really past the thinking stage yet :-)
To wit:

Shortly after writing the post, I discovered a couple of issues or finer points, both of which have to do with the fact that different rows in Riak live independent lives —  they are independently versioned, and no cross-row guarantees are given — specifically, nothing can be concluded from the read order: if one client A updates row X, then row Y, then another client B may see the Y update and later see the old version of row X.

(If Riak is set up with appropriate consistency settings, such that the majority of the replicas of an object are written synchronously, then you do have indirectly some sort of guarantee. Below a certain threshold of machine and/or network problems, that is.
But even so, nothing can be concluded from the read order if there several clusters set up with cluster-to-cluster replication: changes on different keys in one cluster arrive at the other clusters in arbitrary order.)

This leads to at least the following issues:

1. Firstly, when splitting a row into two, I included this step: "Write an empty row back under the original auxiliary row key".
   In fact, doing that would be wrong. Even writing some tombstone dummy value in the old row would be a bad idea; instead, one should simply mark the row as obsolete while keeping the old data. This is necessary because a client accessing the collection later on may see from the master row that the old row has been split, but not see the new rows. Or it may see the old row, but not the change in the main row. In the former case, it is necessary that at least the values in the old row be available (or the collection would suddenly have shrunk); in the latter case, it would not be apparent that the values in the old row are obsolete.
   
   
2. Secondly, after modifying an auxiliary row, the main row should be updated — even nothing in it has changed. This may sound silly, but is the easiest way to ensure that, in the event of a concurrent row split, the auxiliary row in question is taken into account at a subsequent merge (indeed, that the merge is triggered at all).
   
   
3. And even that is not enough if there are more than one cluster: At the time of the read repair of the main row, the update for the auxiliary row may not have arrived - and it may therefore be lost silently. It appears that some kind of versioning of the auxiliary rows is necessary; with such versioning, we can tell when repairing the main row that we're missing an update on the pre-split auxiliary row, and that the reference to it should therefore be kept in the main row, so that it can be repaired later when all information is available.

Ah, the perils and challenges of incomplete information.

Managing concurrent subtlety
As the saying goes,

Meddle not in the affairs of concurrent systems for their ways are subtle, and are quick to anger.
-- Tolkie(error: timeout)n

In a domain as subtle as this — with gotchas in the style of the above mentioned issues, how can we ever convince ourselves that a scheme like the one for Riak collections are implemented robustly and correctly?

As always, there are two ways: Formal verification — proving that the program is correct; and thorough testing — gaining confidence by exercising the program.
Both have their advantages and disadvantages.

If I were Dijkstra, I'd develop a formal proof (as he did in quite a few of his blog postings), and probably modify the algorithm appropriately along the way. I, however, am no Dijkstra, do not believe to have the time to develop a formal proof — nor do I have any delusion of being able to write anything worth reading in the process. Luckily, going the other way, through testing, have something going for it as well:

• It is not always clear which guarantees the components we build on actually provide. That mean that, regarding formal proof, the set of axioms may be uncertain.
• The same test can be applied to different implementations of a component.
• The same test can be applied to different versions of a component — i.e. if we modify the code, we can cheaply gain some confidence in the modified version.

For both formal verification and testing goes that the answers you get depend on the questions you ask.
When proving a property, you prove only that property; when you test a concrete call sequence with concrete values, you test with only those values.
Which is of course a good reason to get familiar with property-based testing — which lies somewhere in between the two in that it tests a property on a number of concrete call sequences — typically a few hundred, and typically different instances for each time a test is run.

This can be a nice compromise between formal verification and hand-rolled test cases — provided of course that the properties and the instance generators are chosen well.
As always with testing, paranoia and imagination is key. But you get more value for your paranoia when it's used to power random test case generation.

For the problem in question, and assuming an Erlang implementation like KKT's, property testing can be done with relative ease using a tool like Quviq QuickCheck or Proper. Their support for testing state machines comes in handy; I've not tried using it before, but this seems a good occasion.

How to test randomly
Randomized testing involves abstracting over usage scenarios.
How the abstraction is done determines what will be tested.
Knowledge of the problem domain is the primary guide; paranoia together with perhaps knowledge of implementation details should provide additional input.
What must be considered is:

• What is tested:
  Which API functions to exercise?
  Which invariants to verify?
• How it is tested:
  Test concurrent use to check for thread safety?
  Test with invalid inputs?
  Should certain special circumstances be simulated - e.g., file I/O errors, disk-full conditions, network delays, packet loss?
• With what it is tested:
  Test with abnormal (very long/short/high/low) inputs?
  Key collisions?
  Many values for the same key?
  Keys which are nearly identical?

This is of course where the imagination and paranoia enter the picture.
You can't assume that "since we generate the values randomly over a large domain, we exercise all code paths" — exercising e.g. a dictionary with a million randomly generated keys is of little use if the keys never collide.

How to test distributed collections
OK then, how to test an implementation of the distributed collections scheme?
Using the state machine testing support of Proper, we can maintain a "model" collection aside the subject-under-scrutiny collection (shortened to SUS in the following).

What to test:
This is the easiest part: We will test the usual collection functions: insert, delete, lookup, list keys.
The invariant is that the result is consistent with the same operation performed on the model collection — as defined below.

How to test:
We'll certainly need to test concurrent use. Let's say that there are three simultaneous users of the collection; we know that there are subtleties involving two, and there may be additional ones involving at least three.
The users won't be really simultaneous, though — We will test concurrent use in a way where the concurrency is made explicit. This provides repeatability and insight into why things fail, which is indispensable for this problem domain.
So, instead of having an actual underlying Riak store, we'll mock it up, in such a way that the mock provides just the guarantees we actually expect the real thing to provide.


Model representation
The structure I have in mind is: the mock remembers all versions of all values put into it. This is what the entire simulated multi-cluster store — let's call it the "cloud" — has ever seen.
The mock furthermore has a number of "views" of the store, corresponding to what you would see if you accessed the cloud at different points. It has a number of these; we'll make the assumption that in any of these views, for any given key, the version will only increase with time. I.e. we do not expect the version of any value in any view to evolve backwards. The mock keeps track of the view-to-version mapping for all keys.

One thing that can happen, then, beside operations on collections, is that value versions find their way from one view to another. Also, version merging can happen in transit — the versions being merged are not necessarily the latest value in any view, but may be any versions that have ever existed in the cloud.
For simplicity, let's say that this is modelled with views also — if we allow enough views, this won't reduce the generality.

The model thus consists, not just of a simple collection, but of an abstract model of the storage cloud.
Beside the externally visible events — the calls to the collection API — there are internal events: a given version of a given key finds its way to one view to another. Both kind of events go into the randomly generated test scenario specification.

The test
It was originally my intention to end with putting all of this together in a Proper-based Erlang unit test — it would then be a good occasion for me to get experience with the state-machine modelling support (link: state machine part is  from p.26).
However, it appears that I have a latency-vs.-completeness tradeoff to make, so I'd better stop here, publish what I have, and hope to return to the subject soon, hopefully with some concrete code. (While this posting has been under way, naturally other topics have come up which I'd like to write about; the order in which further postings arrive here is undetermined...)

Multi-version collections in Riak

The setting:
We have a distributed, transaction-less key-value store with multi-version awareness (vector clocks as version numbers, basically). More specifically, we have Riak.

This provides us with the ability to store and retrieve blobs by key — and to do decent conflict resolution, up to a point.

That is a great building block, but there are some shortcomings which means that we need to build on.

First, the stored values have (as far as Riak is concerned) no structure.
This means that if we store rows as values, and need to update only a part of a row, we still need to fetch and store the entire row.
There are both potential bandwidth and versioning issues with this: you may have to move more data than strictly needed, and the elements of the row are (at the outset) not individually versioned, which makes conflict resolution difficult.
For the latter problem, a solution has already been devised in the form of Vector Maps.

Second, there is no concept of "collections".
This leaves you with two possibilities when you need to store a collection in Riak: storing the entire collection under one Riak key, or storing it under several Riak keys.
(I'll henceforth use the term "row" for a value stored under one Riak key, both to avoid confusion with the keys of the individual element and because that's the term that has become stuck in my head.)

Storing all of a collection in one row has the above-mentioned drawbacks: it's a waste of resources (bandwidth, CPU, disk I/O) much of the time, and you lose much of the advantages of automatic version conflict resolution because only the version of the entire value is taken into consideration.

Storing the collection in different rows also has its drawbacks, though — most significantly, you lose data locality. To fetch a collection of 100 elements, spread over 100 different keys, 100 separate disk locations may have to be accessed. (Unless perhaps you're so clever that you can trick the hashing function, that is...)

A side note:
Riak has a "link" concept which enable you to link objects associated with different keys together; sadly, it has both the locality issue (which can't be helped) and the versioning issue (which is a pity, this being Riak).
For instance, consider an object which in version [] has one link, "foo"; in version [{a,1}] the foo link is deleted; and in version [{b,1}] another link, "bar", is added. Based on versions [{a,1}] and [{b,1}], with link sets Ø and {foo,bar}, respectively, you can't conclude anything about which links should be present in a merge. (Even if the version [] was accessible to Riak, it would be disregarded entirely.)
So, if you want to take advantage of Riak's multiversioning support, you'd better stay clear of links. Or else use them in a disciplined manner — i.e., only adding, never changing or deleting them.

The concerns
Summing up so far, we have these concerns when storing collections:

1. The ability to store an arbitrary amount of elements (obviously).
2. Data locality. — This concern alone would have all elements in one row.
3. Manageable row size. — This concern alone would have one element in each row.
4. Efficient element lookup.
   (In the following, the elements are assumed to be associated with some kind of key.)
5. Easy conflict resolution.
   (E.g., when creating a new row, expect that another Riak node might also decide to create a row at the same time, based on the same decisions; this may influence the way you choose keys. Also, don't move data from one row to another too much.)

(A non-concern, luckily, is hard limits e.g. on row size. The present problem has many features in common with the issue of how to structure database storage, but while DBs have to respect the size of a disk block, we have more freedom. A good thing, too — dealing with concurrency is quite enough.)

The conflict between concerns (2) and (3) is obvious.
One straightforward compromise between the two is "add elements to a row until it reaches a certain size (say, N elements), then create a new row for the next N elements, and so on", but this conflicts with concern (4).

Hash tables
So, we need a growable, block-based collection, in which elements can be looked up efficiently; what are our options?
A hash table as it is usually implemented — using hash value modulo bucket count to get the bucket number — is not so good a choice in light of (5); global redistribution of the data is a thing to be avoided.

Taking inspiration from database systems, however, we might consider the hash table variants used there, secondary-storage hash tables (link?).
These work by using a bitwise prefix of the hash value as the bucket indicator.

The scheme I have in mind differs from both of the variants described in my book on database systems, but the basic idea is the same — after all, the main concern of touching as few block as possible at each access is the same.
I'll sketch my scheme below.

Collection representation using secondary-storage hash table
First, a bit of basics and assumptions:
I'll assume that we have a Riak bucket to our disposal for collection storage; that each collection is associated with an alphanumeric key in that collection.

Each element is versioned individually, for reasons mentioned earlier:

VersionedElement :: {ElementKey, [{VClock, Data}]}

(I'll be using Erlang type specifications.)

After a version merge, multiple versions of the element may be present, with pairwise independent versions (i.e. none of them "happens before" any of the other); this is why there is a list of versioned values rather than a single value.

As mentioned, we use a prefix of the hash values for indexing the auxiliary rows.
The two approaches for extensible secondary-storage hash tables which are described in my database book ("Database Systems: The Complete Book", by Garcia-Molina, Ullman and Widom; interesting reading) are:

• "Extensible hash table" — in which a hash table's current structure is described by the number i of bits of suffix used for selecting the hash bucket. Drawback: the table extension from i to 2i has to be done all at once.
  
• "Linear hash table" — which employs an incremental growing approach, and in which the structure is described by (i,k) where i is as above and k is the current number of buckets, 2^i-1 < k <= 2ⁱ . If a suffix is below k, it's used as the bucket number; if it is k or larger, then the suffix of length i-1 is used (i.e., clear the MSB of the original suffix). When k is increased, the elements in preexisting bucket j are redistributed between j and k — where j is k with the MSB cleared.
  Quite elegant, really.

In our case, the concerns and constraints are similar but different; for one thing, we can get away with using less compact and simple data structures. More specifically, instead of a couple of numbers we're going to have a list of buckets:

RowPointer :: {BitCount :: integer(), BitSuffix :: integer()}

This allows us to split exactly the hash buckets that need splitting, rather than having to rely on a predetermined splitting order.
(This structuring turns out to be similar to the one wikipedia calls extendible hashing, except for the representation of the bit suffix table)
Consider a collection stored under Key. The collection's main row looks like this:

Key → MainRow

where

MainRow :: {[VersionedElement], [RowPointer]}

and VersionedElement and RowPointer are described above.

For locality, we allow a limited number of elements to be stored directly in the main row. This is an optimization, of course, which ensures that for small collections, one row access is needed rather than two; it can be easily omitted but I see no harm in it.

The RowPointer set must be internally consistent: For each possible bit pattern, there should be at most one RowPointer matching that pattern.

For auxiliary rows, we use the key format

AuxKey = Key#BitCount,BitSuffix

(where '#' is a character guaranteed never to be present in Key. Alternatively, use Key# as key for the main row.)
The auxiliary rows contain just elements:

AuxKey → [VersionedElement]

Consistency and conflict resolution
Let us define criteria for a reliable collection storing scheme, to guide the further design:

• Firstly, we can determine the key set for a given collection (subject to which revision versions are visible to us).
•  Secondly, given a collection and a key, we can determine the last version(s) of the associated value.

Part of the first criterion is that deleted elements stay deleted even when other versions of the collection is "versionally visible" where an older version of the element is present.
Part of the second criterion is, strictly speaking, that if a version is visible in which an elements was deleted, then that element is either not present at all in the key set, or the concluded "last version(s)" include a 'deleted' value — i.e., it will not be forgotten that it was once deleted.

Version-Merging Algorithm
When merging two versions of the collection, we have for each row a number of versions of that row available. In Riak we can't tell across rows which versions belong together.

An outline of a merge algorithm is as follows:

• First, determine the set of keys.
• Next, for each key,
• Determine the most recent version(s).

Determining the most recent versions of each element is a known and solved problem; the interesting part is therefore finding all of the relevant versions.

First attempt:

• Take the union of the RowPointers set from all versions of the main row.
• For each of these rows, take all element versions.

This doesn't take deletions into account, however.
To handle deletions, I see two options:

1: The tombstone approach. We represent deleted elements by explicit tombstone values (just as it is done in vector maps and, I believe, in Riak itself).

It would be nicer to have deleted elements be removed from the system altogether. Is this possible, within the constraints we've set up?

Consider a merge operation of two independent versions "D" and "P" of a row; a given element "X" is present in "P", but absent in "D". Assume further that the element modification version is later than the last common version ("LVC"). How do we know whether the "X" was present in "LVC", then deleted in "D" and modified in "P" (in which case we must include both options, present and deleted, in the merge result: [X_P, tombstone]), or if it was created since the LCV (in which case it must be present, not deleted, in the merge result: [X_P])?

           LCV
          / \
delete(X)/   \update(X)
        /     \
       D       P
 (X absent)  (X → X_P)

This question leads to:

2: The extra versioning approach. For each element, include both the version of the last modification and the version of its creation. Furthermore, include for each row the version of the last modification to the row (these versions must be comparable to the versions of the individual elements, by the way).

With this extra information, we can distinguish: If the row timestamp of "D" is later than the creation date, then it was deleted (result=[X_P, tombstone]); if not, then X was created since (result=[X_P]).

Interestingly, the two approaches (1) and (2) are not equivalent, but yield different results in certain cases.
Consider for instance the following history:

                V0
               /:
              / :
create(Y) - A1  V1 - create(X)
               /|
              / |
delete(X) - A2  V2 - delete(X)
               /:
              / :
create(Y) - A3  V3 - create(X)
               /|
              / |
delete(X) - A4  V4 - update(X)

and assume that the input versions available to the merge operation is V4 and one of {A2, A3}.
Then using the tombstone approach, A2 and A3 contains a tombstone value for X, and we must include it in the merge result.
Using the extra versioning approach, on the other hand, we can conclude that the X present in V4 is created at a point not "versionally visible" to either A2 or A3, and thus exists without question after the merger.

(If merging (V4 and A1) or (V4 and A4), the two approaches yield the same result.)

Anyway, I hope to have convinced you that we can resolve conflicts reliably.
Having gotten that out of the way, let's turn to how the data structure is actually manipulated.

Collection Manipulation Algorithms
In the insertion and deletion algorithm, the possibility of write conflicts has been ignored. This is intentional; Riak lets you detect the write conflict, and it can then be repaired — instantly or later, as read repair — using the principles described above.
Note that the organization of the data means that repairs caused by write conflicts are local, i.e. only a few rows need to be repaired, not the entire collection.

Lookup algorithm:
Given a collection key, and an element key, look up the key in the collection. Answer with the versioned element or "not found".
 - Fetch the main row of the collection: {Elements, RowPointers}.
- Is the element among Elements?
  - If yes: There's your answer.
  - If no: Is there a RowPointer matching the key?
    - If no: Conclude "not found".
    - If yes: Fetch the auxiliary row pointed to by RowPointer: Elements2.
      - Is the element in Elements2?
        - If yes: There's your answer.
    - If no: Conclude "not found".

Update algorithm:
Like the lookup algorithm, with the obvious additions. Left as an exercise to the reader.

Insertion algorithm:
Given a collection key, an element key and a value, insert the (key,value) pair in the collection.

 - Perform a Update of the key. If the key was found, we are done; if not, continue.
- Fetch the main row of the collection: {Elements, RowPointers}.
- Is there room for another element in Element, without exceeding the size threshold?
  - If yes: insert the versioned element in the main row.
  - If no: Is there a RowPointer matching the key?
    - If yes: fetch the row pointed to by RowPointer, and add the versioned element.
      - Is the size limit exceeded?
        - If no: Write the auxiliary row back.
    - If yes: Split the auxiliary row into two rows with a BitCount increased by one.
          - Write these rows.
      - Update the main row with the new row pointers.
      - Write an empty row back under the original auxiliary row key. (Optional?)
    - If no:
      - Create a RowPointer matching the key and having as small a BitCount as possible.
      - Insert an auxiliary row for that RowPointer, containing just the new element.
      - Update the main row, adding the RowPointer.

Deletion algorithm:
Given a collection key, and an element key, delete the element with that key from the collection (if it is present).
Done like Update with a 'tombstone' value — if the element is present; if it isn't, nothing should be done.

Conclusion
Riak is a good building block for a distributed storage system, but for certain applications you need some appropriate abstractions on top of the raw key-value store. There are for instance good reasons to put some thought into the organization of collections of values, and ensure that conflicting versions can be dealt with appropriately.
I have presented some of the issues involved and attempted to put together a  usable scheme.
At this point, it's all talk and no code, though — not a bad starting point, but I cannot claim much about the resulting scheme except that, given that this text ended up much longer and with many more asides that I originally envisioned, I must have put some thought into it.

(Update: There is now a follow-up. Still more thinking than code, though -- for my part, at least.)

Oh, and by the way, I hope to be able to find the time to explain why I'm writing about Riak, all of a sudden.