DEV Community: Ditto

The Future of the Cloud? Make it Optional

Ryan Ratner — Wed, 13 Jul 2022 16:40:00 +0000

Today, most apps are cloud-only. Data must travel halfway around the world to a remote data center just for it to arrive at a device in the same room. These apps become entirely unusable if their data connection is slow or a Wi-Fi router breaks. Cloud-only apps have made it difficult for the "deskless workforce" to get their work done. They require database access on mobile devices in the field, where crucial on-the-ground operations are happening, but their cloud-only applications do not function 100% of the time. Operations halt while employees wait for a mobile phone to reconnect. It isn't a great user experience; it costs businesses money and can even put people in life-threatening situations when real-time data is needed for quick decision-making.

From Cloud-Only to Cloud-Optional

Despite these problems, most applications today are cloud-only. Why? If you want to send your friend a picture of a cat, you can use AirDrop, which sends the data directly between the two devices, peer-to-peer. So why don't all apps send other kinds of data directly, peer-to-peer, between devices? Why isn't the cloud optional?

The answer is that it's simply easier to build cloud-only applications. Over the past few decades, investors have funneled billions of dollars into cloud-only databases and tools. Because of this, developers don't have to think about TCP/IP networking, database
partitioning, or on-disk compression whenever they want to update a field in a database table. For years, this existing toolkit of cloud-only tools has made cloud-only applications the fastest way to deploy collaborative software.

Peer-to-peer is not new, but it is still hard to use

There are scalable peer-to-peer protocols for data that doesn't change, such as AirDropping cat pictures or torrenting movies and music – BitTorrent accounts for 27% of all upstream Internet traffic¹. But, developers need the reliability of a database for times when that data starts to change at the speed of Slack and requires the precision of an airplane safety checklist. Unfortunately, few engineers in the world can build a peer-to-peer database. It's a new software engineering paradigm that needs more education, tool development, and user experience improvements. Because of this, we haven't had a cloud-optional, peer-to-peer database. Until now.

Ideally, a peer-to-peer database needs to be:

User-friendly. Developers are users too! Instead of sending data to a remote server, the application needs to write data to its local database first in the form of changes, then listen for changes from other devices, and recombine them on the fly. Ditto provides an API on top of these details so developers can focus on their business logic instead of synchronization logic.
Cloud-optional. Devices can go into dead zones, routers can crash, or cloud services can go down. All devices must see the same query results given the same set of changes, even if the changes arrive in a different order. Ditto's Conflict-Free Replicated Data Types
(CRDTs) provide a consistent view of the data for every device. These data structures are still very much a new topic in computer science research.
Partitioned. Mesh networks can generate a tremendous amount of data that can overwhelm small devices if each node aggressively tries to sync every piece of data. Ditto provides a Big Peer and a Small Peer SDK with different replication strategies. The Small Peer is selfish, which means it only synchronizes data it explicitly requests, giving developers complete control over storage and bandwidth usage. The Big Peer is greedy, which means it synchronizes as much as possible.

Ad-hoc. Devices may join and leave the mesh at any time. Despite this churn, all devices should still see and have input on the same data: the same "source of truth." Routing messages between appropriate nodes becomes a mathematical challenge as the mesh network topology changes over time. Ditto's novel delta state CRDT is great for ad-hoc mesh networks because they are flexible and do not rely on having the full history of a database table to write or read the latest value.
Forward-Compatible. Since devices update at different times, they need to account for incoming data with different schema. For example, if a device is offline and therefore outdated, it should still be able to read new data and sync. Ditto is causally consistent, providing a reliable order of changes that can be inspected, including incorporating metadata about schema changes over time.

A shared set of tools

These challenges are straightforward to explain but bring up a sizable, entangled web of discussions. There are many different ways to design and implement caching, data synchronization, querying, and distributed architecture.

It is a lot of work for development teams to create a reliable peer-to-peer database that syncs data in a partially connected mesh. Organizations either can't hire the talent to complete the job or don't see the financial benefit of investing in building it themselves. It is a more efficient use of resources to support a database that many different companies use so that each individual company doesn't need to invest in building and maintaining their own cloud-optional database.

With Ditto, developers have a complete end-to-end cloud-optional solution that combines the best of cloud software with the best of peer-to-peer software. On the surface, the shift from cloud-only to cloud-optional seems subtle. But this is a more fundamental paradigm shift and provides tremendous business opportunities. Ditto is setting the standard as one of the first CRDT-based peer-to-peer databases that turn the existing plethora of mobile devices into a cloud-optional intelligent edge platform.

If you are a developer and want to make your cloud optional,
get started today for free. We're hiring, so look at our available listings and see if you think Ditto might be a good fit for you.

Footnotes

The Global Internet Phenomena Report, Sandvine; 2019. ↩

Testing CRDTs in Rust, From Theory to Practice

Ryan Ratner — Wed, 06 Apr 2022 16:24:24 +0000

Ditto is a database that puts theory to practice. Ditto's peer-to-peer
data-structures based on CRDTs are deployed into real-world production
use cases. As we make improvements to the underlying data-structures,
we leverage stateful property-based testing to ensure that Ditto's
core data-structures maintain their robustness and stability. In this
article, we show how this type of testing helped reveal a surprising
problem in an academic paper's optimized CRDT.

What Are CRDTs?

CRDTs are data-structures that can be used in distributed systems that
allow relaxed consistency models. They make handling conflicting
updates easier for programmers, and enable the development of
peer-to-peer and
"local-first"
applications. They're the primary data-structure that developers work
with in Ditto's Document Database. CRDTs allow Ditto to integrate
concurrent changes from clients into a single, deterministic, and
meaningful value.

Over time the words that lend their initials to form the acronym CRDT
have often changed, it is usually accepted that CRDT stands for
Conflict-Free Replicated Datastructure, or in our case Convergent
Replicated Datastructures.

There has been a lot written about CRDTs, but a good starting place to
learn more is crdt.tech

Ditto uses a State-based CRDT. State-based CRDTs use metadata inside
the data-structure to track causality and determine a value. We
developed a novel type of Delta-State CRDTs for Ditto, which we will
explain in more detail at a later date (stay tuned!) where the
differences between peers is calculated and sent over the network,
rather than sending the full state.

Implementing A New CRDT

Ditto's database represents a document as a CRDT map with additional
nested CRDTs for values. Today, the Ditto Document currently only
supports Remove Wins behavior. My current task at Ditto is to create a
new CRDT Map that supports both Add Wins and Remove Wins
behavior. This is a novel data-type which allows the developer to
choose the behavior when performing a remove operation.

Throughout this article I write mainly about Sets, as they are simpler
to describe and reason about, but what is discussed can also be extended to
Maps.

Add Wins and Remove Wins are choices in CRDT Set behavior. Given
concurrent updates to the same element in a Set, where one peer
inserts the element and another removes it, you have a choice: either
the Add Wins and the inserted element is in the Set, or the Remove
Wins and the element is not in the Set. In order to enable Remove Wins
behaviour, State-based CRDT sets tend to use some metadata, called a
tombstone, that marks an element as removed. To understand the
difference between Add Wins and Remove Wins, check out the graphics
below.

Add-Wins Behavior

Remove-Wins Behavior

Ditto observes this situation often from its customers in
the airline industry. Before takeoff, two flight attendants
with tablets or phones, in this case peer 1 and peer 2,
may download backend data before syncing with each other.
When these two disconnected flight attendants make concurrent,
offline edits to the database before syncing, as shown in the
graphics above, merging of the two devices results in
different behavior depending on the CRDT set behavior.

Remove Wins behavior can be surprising for some users, however. In general it
makes sense, as it does what it says: given any pair of concurrent add
and remove operations, the remove will win. However, with the
following sequence enacted concurrently by two peers, everyone is
surprised by the result:

Peer 1 adds element X to the set
Peer 1 removes element X from the set
Peer 1 re-adds element X to the set
Peer 2 adds element X to the set
Peer 2 removes element X from the set
Peer 2 re-adds element X to the set
Peer 1 merges with Peer2 and vice versa
X is not in the set

To explain, the remove at step 2 on Peer One is concurrent with the
add (and re-add) on Peer Two. The remove at step 2 on Peer Two is
concurrent with the add (add re-add) on Peer One. When the sites
exchange data and merge into a single value the remove(s) win and the
Set is empty.

Given this peculiarity, Ditto wants to enrich the types available to
developers and offer both Add Wins and Remove Wins behavior. However,
designing and implementing CRDTs is a difficult task, we have an
impossibly large set of interleaving operations and orderings to test,
and we want to be certain that the result is deterministic across them
all. Property-Based Testing is an essential tool in the armory.

What Is Property-Based Testing?

If you have never heard of property-based testing before watch
this. John Hughes is
one of the (grand?)daddies (sorry John) of property-based testing. A
stone cold genius, he has probably saved countless millions of
dollars, and maybe even lives since inventing Quickcheck with Koen
Claessen

There are lots of descriptions out there of property-based testing. In
general, it means generating the test cases, usually by generating
inputs. The programmer declares the properties of the thing under
test and the library tries to generate cases that violate those
properties. When it does find some input that violates a property, it
will shrink the input into a minimal failing case. The classic
example is that for all generated arrays as inputs, the reverse of the
reverse of an array is equal to the original array. This is compared
to unit testing where you make a bunch of arrays (empty?), hand
code the result, and say "expected == actual". Another common example
is round tripping for serialisation/deserialisation libraries. That
the result of serialise then deserialise for any input is the same
as the original input is a classic property.

Stateful Property-Based Testing

My personal favorite way of testing complex things, like distributed
systems or distributed data-structures, is to use EQC's Statem. This
is a stateful property-based testing module, that allows me to model
the system under test, generate a sequence of operations to run on
that system, and check that the system matches the model,
throughout. If some sequence of operations leads to a difference
between the model and the system under test (hereafter SUT),
quickcheck shrinks the operations to a minimal sequence.

With data-structures (like CRDTs) this is great. We can take an
"obviously correct", but poorly optimised implementation of a CRDT
(like the classic Observe Remove Set) and use it as the model, or
specification, for a better implementation, like the Optimised Add
Wins Set.

We execute the same randomly generated set of operations on both
data-structures, and if there exists a difference between the behavior
of the implementations, quickcheck will tell us, and crucially SHRINK
the list of operations to a minimum. Then it is the programmer's job
to decide if the bug is in the model, the SUT, the test itself, or
some combination thereof.

Every time quickcheck finds a failing counter example, you save it,
and implement it as a unit test (for regression). Since the failing case was
generated randomly, you cannot be certain that exact case will be
generated again.

It is an amazing tool. I've used it with Rustler to test Rust
code. However, not everyone at Ditto has a licence, and not everyone
at Ditto wants to learn Erlang. Even if I still think EQC statem is
the gold standard, it isn't what I used for this work, but
understanding the idea of generating a series of operations, and
applying them to a model and a "real" implementation is all you need to
take from this section.

Poor Man's Rust Statem

A very basic attempt to copy the process of EQC statem would be to generate a
list of operations, and execute them. This process is described in a few places,
For example Tyler Neely's post.

In this case I use Rust Proptest to generate a Vector of operations to run on a
CRDT set. ¹

A Model (Specification) and a SUT

As a first step to supporting multiple behaviors in Ditto's document
I looked for prior art or existing literature and found the work of
André Rijo et al, specifically the OAR-Set in Rijo's 2018
dissertation and the
optimised version in this paper
(also from 2018).

My idea was to use the Rijo OAR-Set as a model for the Map I was
making.

Remove Wins vs Observed Add

In order to get a feel for the Set's semantics I implemented the
OA-Set, the OR-Set, and the OAR-Set, testing them against each
other using the basic model vs SUT approach described above.

The Observed Add behavior was new to me. It isn't a Remove Wins in
the sense that any remove concurrent with any add always
wins. Instead, the Observed Add allows for the "removal of removes"
that is to say in the case that re-adding something after observing
its removal means that the remove itself will no long have an effect
on some as yet unseen concurrent add. In a true remove wins set you
can end up with the surprising "mutual destruction" of two Peers as
described above who both:

add element X to the set
remove element X from the set
re-add element X to the set
merge

In this case, with Observed Add, when Peer One re-inserts they are in
effect saying "I'm removing my remove!" and so it has no effect on
concurrent adds. If you want to learn more about how this is achieved,
you should read the paper. It is
neat, and I wish I'd known about it before I made the Remove Wins Map!.

I was surprised by the Observe Add behavior, and wanted to better
understand it. I understand by doing, so I implemented the Optimised
OAR-Set from the paper. Naturally I tested the Optimised version as
the SUT against the unoptimised version as the model. I was surprised
to see they diverged.

The Test; A Counter Example

The state of the system is modeled as a BTreeMap<usize, (Model, SUT)> where
the key usize is a "peer" in the distributed system.

The operations that the test performs on the system are:

    enum SetOp {
        Replicate { src_peer: usize, dst_peer: usize },
        Insert { peer: usize, elem: u64 },
        Remove { peer: usize, elem: u64 }, // this is the remove wins remove
        ObserveRemove { peer: usize, elem: u64 }, // this is the add wins remove
    }

The test generates a sequence of operations: a mutation operation happens at a
site, and replication merges the data from the first site to the second.
This models the replication of data from one site to another. Without any
distributed systems, networks, databases, replication protocols, threads or
parallelisation, we have a very basic but effective model of a distributed system.
Each operation is a step, or moment in time. We have a single set. We allow
concurrent updates to it. We arbitrarily replicate between pairs of sites.

After all operations are applied, we merge all sites into a single
value. This is the guarantee of eventual
consistency:
when the system is quiescent (i.e., all peers have seen all operations)
there is a single common value. At each merge we check that the model
and SUT match, and at the end, when ALL sites are merged, we check
again that the model and SUT match.

Proptest found a failing case and kindly shrank the failing case down
to:

let shrunk_ops = vec![
    // Peer Zero inserts 0 ⇒ [0]
    SetOp::Insert { peer: 0, elem: 0 },
    // Peer Zero does a remove wins remove of 0 ⇒ []
    SetOp::Remove { peer: 0, elem: 0),
    // Peer Zero replicates to Peer Two ⇒ []
    SetOp::Replicate { src_peer: 0, dst_peer: 2 },
    // Peer Two (re)-adds (and has observed the remove wins remove (in OARSet this "removes the remove")) ⇒ [0]
    SetOp::Insert { peer: 2, elem: 0 },
    // Peer One inserts 0 (it has _not_ seen the earlier remove) ⇒ [0]
    SetOp::Insert { peer: 1, elem: 0 },
    // Peer Two observe removes 0 ⇒ []
    SetOp::ObserveRemove { peer: 2, elem: 0 },
];

At some step the test failed because the model value was [0] and the
SUT was [].

The Difference

The job of the programmer is now to figure out why the model and the
SUT diverge. In this case I had a hunch, so I augmented my test to
merge the final state in all possible permutations, and this showed
that in some orderings the final sets matched.

If we start by merging Peer One into Peer Zero, and then merge Peer
Two into the result (i.e. (Peer Zero <- Peer One) <- Peer Two) the
divergence happens when Peer Two is merged into the result of the merge of
Peer One into Peer Zero. This is because when we merge Peer One into
Peer Zero, we lose the 0 element from the set. Peer One still has
the Observe Add tombstone (recall that it was Peer Two that removes
the remove by re-adding 0 to the set.) The tombstone at Peer Zero
acts on the element on Peer One.

In the optimised specification a remove ACTUALLY REMOVES the element;
it is forgotten (that's part of the optimisation). When we then merge
Peer Two, the unoptimised set still has the element 0 in the set,
and notices that the tombstone itself was "removed" by the re-add at
Peer Two, and so the add on Peer One becomes concurrent with the
observe remove on Peer Two (Observe Remove means Add Wins).

Well done property testing! The paper passed review and was accepted
in the national informatics conference INForum 2018, but a six-step
run with 3 peers shows the specs don't match. That isn't a critique of
the work or the scientific process, but rather an endorsement of the power
of stateful property-based testing.

Not only do the specs not match, it also shows that the spec for the
Optimised OAR-Set is not actually a CRDT, as a State-based CRDT's
Merge function must be idempotent, commutative, and associative. For
example, if we merge the Peers in the order (Peer Two <- Peer Zero)<- Peer One
then the SUT matches the model.

The Fix?

I wanted to fix it for the sake of the blog post, and for my own
understanding. I'm not sure my fix is great, as it isn't really in the
spirit of "optimised." My changes are in merge and lookup of the
spec.

In merge we never remove/forget a value from the Add set, even
if it appears to be tombstoned by the merge. In lookup merely being
present in the Add set with timestamps is no longer enough, the
element must also NOT be present in the Remove map. That's it. Now
they match, the property test passes, and the Set is a CRDT.

Conclusion

I contacted André Rijo and Nunu Preguiça about this work, and they
were generous of their time in discussing it with me, agreeing that
I had found a genuine issue in the paper: that the optimised
specification does not match the OARSet specification.

Deletes in distributed systems are hard, and the Observed Add semantic
from these papers is a valuable alternative to the harsher Remove Wins
behavior.

This posts illustrates just how hard it is to make correct CRDTs, and
why tools like Stateful Property-Based Testing are so valuable.

Look out for the new types coming soon to the Ditto Platform.

Learn more by visiting our Docs

There is a PR open on Proptest that adds EQC Statem-like features to Proptest. 🤞 ↩

DO cross the streams! Introducing Combine Support in Ditto - build complex workflows from multiple reactive streams

Ryan Ratner — Thu, 10 Mar 2022 19:10:39 +0000

Historically, iOS developers had to use delegates, polling timers, notification centers, and callbacks to build reactive architectures. While these techniques are useful for simple callbacks, they falter when dealing with multiple events in robust event-driven apps. Creating complex chains, combinations, and permutations of multiple reactive streams is incredibly difficult. Thus, reactive libraries like the open source Rx suite ¹ and Apple's Combine were created to solve this exact issue.

Since version 1.1.1 we have several Combine.Publisher extension methods on our long-running callback APIs. Notably:

These new extension methods make it easy to use reactive techniques in your iOS or macOS apps without an additional library. We consider these APIs stable: they are available for iOS versions 13.0+ and macOS 10.15+. Of all the extension methods, you'll probably use LiveQueryPublisher the most. This Publisher is a method on the DittoPendingCursorOperation and facilitates most of the synchronization behavior of your app. Here's an example:

let cancellables = Set<AnyCancellables>()

// observe all documents in a collection
ditto.store["cars"]
  .findAll()
  .liveQueryPublisher()
  .sink { (documents, event)
    // do something with documents and events
  }
  .store(in: &cancellables)

// or you can observe documents matching a query
ditto.store["cars"]
  .find("color == $args.color && mileage > $args.mileage", args: [ "color": "red", "mileage": 5000])
  .sort("mileage")
  .limit(50)
  .liveQueryPublisher()
  .sink { (documents, event)
    // do something with documents and events
  }
  .store(in: &cancellables)

Stopping the live query is identical to stopping the publisher:

cancellable = nil // if you have a single cancellable, stop it by setting it to nil

cancellables.removeAll() // removeAll will stop all attached publishers and their respective live queries

Often, our users love to use Codable with Ditto Documents. Since Codables can fail to decode for some reason (type mismatch is the most common example), you may want to handle each decode error individually as they're streamed out. The liveQueryPublisher gives each emission as an Array<DittoDocument> each time the query results change.

struct Car: Codable {
  var _id: String
  var name: String
  var mileage: Float
  var isSold: Bool
}

ditto.store["cars"]
  .findAll()
  .liveQueryPublisher()
  .flatMap({ (docs, _) in docs.publisher })
  .tryMap({ try $0.typed(as: Car.self).value })
  .sink(receiveCompletion: { (error) in
    print("Decoding a document failed: \(String(describing: error))")
  }, receiveValue: { (car: Car) in
    print("Successfully decoded a car \(Car)")
  })

There may be times where your application would like to collect the final values of only successfully decoded Car objects. You may want this functionality if you're not interested in stale or malformed legacy documents. Here we use flatMap to turn the Publisher<[DittoDocument]> into a Publisher<DittoDocument> that will emit for each item in the array. Now for each of the single emitted DittoDocument we will use compactMap on the decoding function. Notice the try? here. If any Car fails to decode, this function will return a nil. compactMap will skip any nil values. At the end we will use collect to gather all of the emitted Car objects and repackage them into an Array.

let cancellable = ditto.store["cars"]
  .findAll()
  .liveQueryPublisher()
  .flatMap({ $0.documents.publisher })
  .compactMap({ try? $0.typed(as: Car.self).value })
  .collect()
  .sink { cars in
    print("Sucessfully decoded cars: \(cars). Failed to decode cars were removed from the array.")
  }

SQL `JOIN`-like behavior with Ditto using Combine

A question that we get all the time is "How do I perform SQL-like JOINs with Ditto"? While Ditto's current interface can't handle relationships like a traditional SQL database, our Combine support can help us achieve the same effect. ².

Let's say you're trying to build a menu view in SwiftUI like the first image shown below.

It's likely that you'll use a SwiftUI List with multiple Sections with the help of ForEach. Assume each document in their respective collection looks like the following:

// products
{
  "_id": "chicken-sandwich",
  "name": "Chicken Sandwich",
  "detail": "Grilled chicken, tomatoes, lettuce, and mustard",
  "categoryId": "entree"
}

// categories
{
  "_id": "entrees",
  "name": "Main Courses and Entrees"
}

We can create representable Codables for each Document type. Notice that we've added Identifiable to help ForEach iteration:

struct Category: Codable {
    var _id: String
    var name: String
    var isOnSale: Bool
}

extension Category: Identifiable {
    var id: String {
        return self._id
    }
}

struct Product: Codable {
    var _id: String
    var name: String
    var detail: String
    var categoryId: String
}

extension Product: Identifiable {
    var id: String {
        return self._id
    }
}

Notice that the product has a categoryId, this is our foreign key. Linking these foreign keys with our earlier APIs wasn't very straightforward. However, with our new Combine extensions, we can use the combineLatest function to emit a single callback for both the products and categories collections.

First we will need to create a JOIN-ed struct that will house our nested values:

struct CategorizedProducts {
  // This is the category
  var category: Category
  // This is the products filtered by the category above
  var products: [Product]
}

// We add `Identifiable` to help `ForEach` iteration.
// Since this is unique by inner category._id property, we ensure
// to return its value
extension CategorizedProducts: Identifiable {
  var id: String {
      return self.category._id
  }
}

To populate CategorizedProducts we initially need to create our combineLatest implementation. First we will need to get access to both categories and products publishers. We use Codable to map documents into concrete data types in the .tryMap operator.

let categoriesPublisher = categoriesCollection.findAll().liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Category.self).value }) })

let productsPublisher = productsCollection.findAll().liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Product.self).value }) })

Finally, we can combine the latest values of each publisher using .combineLatest and .map. In the .map function, we iterate over each category and use it to create a CategorizedProducts object and filter all products by the categoryId.

let cancellable = categoriesPublisher.combineLatest(productsPublisher)
    .map { (categories, products) in
        return categories.map({ category -> CategorizedProducts in
            let filteredProducts = products.filter { product in product.categoryId == category._id }
            return CategorizedProducts(category: category, products: filteredProducts)
        })
    }
    .sink { categorizedProducts in
      print("categories with their products", categorizedProducts)
    }

If any update, insert, or deletions occur to the products or categories collection, you'll always get a new set of categorizedProducts. To show this menu we can iterate over each categorizedProducts in SwiftUI like so:

List {
  ForEach(viewModel.categorizedProducts) { categorizedProducts in
    Section(categorizedProducts.category.name) {
      ForEach(categorizedProducts.products) { product in
        VStack(alignment: .leading) {
          Text(product.name)
              .bold()
          Text(product.detail)
              .font(.caption)
        }
      }
    }
  }
}

For an example project using this technique, checkout the source code on GitHub here.

Making SQL `JOIN`-like behavior more efficient

Using .combineLatest as a way to get a SQL JOIN-like can achieve both a reactive API as well as better management of relational models. However, it's important to know that this is just an approximation. Remember, Ditto live queries sync exactly what you tell it to sync from the mesh. The example above will sync all categories and all products. This behavior may be desirable for many use cases but let's take at some ways we can reduce what is synced over the mesh.

You can use the queries to limit what documents are of interest by specifying a more selective query. Let's say you're only interested in getting CategorizedProducts where the category is one of "appetizers", "entrees", or "desserts". We could now filter using a query with .find instead of .findAll on the categories collection.

let categoriesPublisher = categoriesCollection
    .find("contains([$args.categoryIds], _id)", args: ["categoryId": ["appetizers", "entrees", "desserts"]])
    .liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Category.self).value }) })

let productsPublisher = productsCollection
    .find("contains([$args.categoryIds], categoryIds)", args: ["categoryId": ["appetizers", "entrees", "desserts"]])
    .liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Product.self).value }) })

let cancellable = categoriesPublisher.combineLatest(productsPublisher)
    .map { (categories, products) in
        return categories.map({ category -> CategorizedProducts in
            let filteredProducts = products.filter { product in product.categoryId == category._id }
            return CategorizedProducts(category: category, products: filteredProducts)
        })
    }
    .sink { categorizedProducts in
      print("categories with their products where categoryId are appetizers, entrees, desserts", categorizedProducts)
    }

Now the device will only sync relevant CategorizedProducts by the specified category _ids.

Advanced Scenarios for SQL `JOIN`-like behavior

Let's say we have a scenario where we only want to show CategorizedProducts where the Category.isOnSale == true. To do a SQL JOIN-like behavior with Ditto and Combine is not as straightforward as it was in the example above. This is because we are querying on a property that only exists on the categories collection. Previously we were querying on both the primary key Category._id and the foreign key Product.categoryId. To do this is a bit harder than most and requires a decent amount of understanding of all the Combine operators.

First we will need to create a live query of categories where the isOnSale == true.

let categoriesPublisher = categoriesCollection
    .find("isOnSale == true")
    .liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Category.self).value }) })

Since we don't know the categories that are passed ahead of time to filter the products, we need to filter them after the values have returned from the categories live query publisher. Once we've received the categories that fit the isOnSale == true query, we can then create map it's result into an AnyPublisher<[CategorizedProducts], Never>. We use AnyPublisher<[CategorizedProducts], Never> for brevity so that the entire chain isn't convoluted with complex generics.

Once the category's publisher emits data, we retrieve an array of categoryIds: [String] to feed it to the products live query publisher by filtering on the categoryId foreign key property. Next, inside of the first categories publisher's map, we will use our .combineLatest technique to map and filter the CategorizedProducts.

The most important function in this chain is the switchToLatest right before the .sink. While it can be tricky for your eyes to follow, the switchToLatest will dispose of any publishers if the top level publisher changes. This is a critical operator because we absolutely want to dispose the product live queries if the categories change. Categories may change if a new isOnSale is added, an existing category that has isOnSale becomes false, or is deleted. Without switchToLatest we will get mismatched products from previous categories.

let categoriesPublisher = categoriesCollection
    .find("isOnSale == true")
    .liveQueryPublisher()
    .tryMap({ try $0.documents.map({ try $0.typed(as: Category.self).value }) })


categoriesPublisher
    .map({ categories -> AnyPublisher<[CategorizedProducts], Never> in
        // retrieve the categoryIds for all that were on sale.
        let categoryIds: [String] = categories.map{ $0._id }

        let productsPublisher = self.productsCollection
            .find("contains($args.categoryIds, categoryId)", args: ["categoryIds": categoryIds])
            .liveQueryPublisher()
            .tryMap({ try $0.documents.map({ try $0.typed(as: Product.self).value }) })
            .catch({ _ in Just([]) })
            .eraseToAnyPublisher()

        // we now create CategorizedProducts from the filtered categories and filtered products
        return Just(categories).combineLatest(productsPublisher)
            .map { (categories, products) -> [CategorizedProducts] in
                return categories.map({ category -> CategorizedProducts in
                    let filteredProducts = products.filter { product in product.categoryId == category._id }
                    return CategorizedProducts(category: category, products: filteredProducts)
                })
            }
            .eraseToAnyPublisher()
    })
    .switchToLatest() // extremely important so that we dispose of the products publisher if the categories change
    .catch({ _ in Just([]) })
    .sink { categorizedProducts in
      // completely filtered categorizedProducts
    }

Now if any category is added, removed, or updated to match isOnSale == true, we will instantly retrieve a new set of CategorizedProducts with live queries specifically limited to the matched Category._id
As you can see this is very complex and really shows off the power of Combine with Ditto. We understand that the last example is a very complex and verbose code sample to achieve SQL JOIN-like behavior and we are working hard on adding native support directly within Ditto.

We're extremely excited to see all the new iOS and macOS applications that leverage Combine with Ditto. Building reactive applications has always been a core tenet of Ditto's design philosophy. Now with Combine, you can have incredible control over Ditto's live query system to build complex and robust event-driven apps.

Footnotes

Rx is suite of libraries that really gained a tremendous foothold for designing Reactive streams. There are libraries for JavaScript - RxJS, Swift - RxSwift, Java - RxJava, Rx .NET - Reactive Extensions and more. If you're building Ditto applications beyond Swift, you should take a look at these libraries. Most of the operators available in Apple's Combine framework have equivalent functions. ↩
We get so many requests for relationships, aggregates, syncing specific fields and more that we're planning on adding this functionality directly in an upcoming iteration of our Ditto query language. This will likely look like SQL but will have some changes to accommodate working with Ditto's distributed database semantics. ↩

Mocking Time in Async Rust

Ryan Ratner — Tue, 15 Feb 2022 19:01:09 +0000

We love async Rust at Ditto. We were early adopters, using futures and streams in our networking code long before async/await became stable. Our product is a perfect use case. It's I/O-heavy, there are numerous peer-to-peer connections synchronizing data concurrently, and we need to use all cores efficiently on low-powered devices like mobile phones.

A common challenge in async is writing good tests. Take a simple example: below we create a hypothetical Connection. If there is no activity for 10 seconds it should time out internally and close its events channel. We can write a test verifying this:

#[tokio::test]
async fn test_timeout_occurs() {
    // Create the Connection
    let mut conn = Connection::new().await;
    // Wait for a while
    delay_for(Duration::from_secs(11)).await.unwrap();
    // The receive channel should have closed
    assert_eq!(conn.try_recv().unwrap_err(), TryRecvError::Disconnected);
}

One problem with this test is that it always takes 11 seconds to execute: 10 seconds to wait for the timeout plus a safety factor. Slow tests are a huge hassle for both developers and CI/CD. Imagine if it was 30 minutes instead. Your CI runs would take at least 30 minutes, all because of this one test!

A related challenge is testing for the absence of a timer expiry. Here we ensure the events channel is still open at 5 seconds:

#[tokio::test]
async fn test_timeout_does_not_occur() {
    // Create the Connection
    let mut conn = Connection::new().await;
    // Wait less than in the other test
    delay_for(Duration::from_secs(5)).await.unwrap();
    // The receive channel should still be open (but no value)
    assert_eq!(conn.try_recv().unwrap_err(), TryRecvError::Empty);
}

This test is also slow, weighing in at 5 seconds, but there is a more insidious problem. Imagine our Connection had a bug which caused it to close the channel after only 4 seconds. Would this test catch it? Probably… but not necessarily.

When that internal 4-second timer expires a series of steps must occur. A tokio worker thread must be unparked and poll the task which was waiting on that timer. Having noticed that the timer has expired it drops the events sender, which indicates to the receiving end that the channel is closed. It's only a small amount of work but it's not instantaneous, and it has to run concurrently with the main test task.

If you run this on your high-end development machine (which is what Rust developers use) then that expiry-and-channel-close process will occur in a matter of milliseconds. By the time the test task performs the assertion 1 second later, the channel will almost certainly be closed and the bug will be detected. On a heavily-loaded CI server, however, there may be fierce contention between threads. Processing the inner timeout might be delayed by 1–2 real-world seconds and the test would appear to pass. The previous test has the same problem in reverse; if the inner timeout gets delayed it will fail, even though the code being tested is correct.

These tests are non-deterministic and will result in flaky CI: intermittent build failures where you're not sure whether you can trust the results. This is suboptimal.

Conventional wisdom says that you shouldn't structure your code this way. Any logic you want to test should be expressed as synchronous code which is fully deterministic and abstracted away from the system clock. This is okay for small unit tests but it becomes a headache in larger integration tests. Some of our business logic is tied up in async behavior. Why can't we test that, the way our customers will experience it? If we test only the sync parts we're losing coverage.

What we want is a way to test our time-driven code with fewer compromises.

It should be possible to test async code "as-is" in unit tests.
In a test context, passage of time should be abstracted so that tests execute near-instantly no matter how long the delays are.
Tests should be fully deterministic: an earlier timer always gets to work to completion before a timer scheduled later.
As a unit test advances mock time, Instant::now() should return the correct intermediate time during the triggering of each timer.
The processing work attached to a timer should allow new timers to be registered along the way.

Ditto has built an internal library to address all of these problems. This crate, ditto_time, abstracts over std::time and the tokio timer functions. Here is the first test again, except now using this library.

#[test]
fn test_timeout_occurs_fast() {
    let (time_control, _guard) = register_new_control();
    let rt = build_instrumented_runtime(&time_control);
    rt.block_on(async {
        let mut conn = Connection::new().await;
        time_control.advance(Duration::from_secs(10)).await; // <----
        assert_eq!(conn.try_recv().unwrap_err(), TryRecvError::Disconnected);
    });
}

This test is reliable and completes immediately. First there is a little boilerplate which ensures the code under test will use mock timers instead of real ones. For now, focus on the call to time_control.advance(). Notice that it moves time ahead precisely 10 seconds. Then on the next line, the channel is guaranteed to be closed.

This is the most difficult part. It is not sufficient to trigger the timer. Somehow we have to know when the code which received that timer event has finished doing all of its associated work. This isn't possible using conventional timer futures. We don't know when the associated task will get scheduled, let alone when it's finished. Therefore the futures in ditto_time are different—they output a FrozenTimeControlGuard.

{
    let _guard = ditto_time::delay_for(Duration::from_secs(10)).await.unwrap();
    // handle timer event
    // guard dropped; time can advance
}

It turns out that in almost all situations, timer-driven code does all of the relevant processing in a single block immediately following the await. Therefore all we have to do is create a binding like _guard, and now we have a value that can notify us when the end of the block has been reached—and therefore time can continue. In unusual cases the developer can preserve the guard as long as needed.

pub struct FrozenTimeControlGuard(Option<oneshot::Sender<()>>);

In unit test code the guard contains a oneshot sender. On drop the channel closes and the call to advance() knows that it can move on to the next timer. advance() is an async loop over all the currently-registered timers, triggering them one at a time and waiting for the corresponding channel to close. In non-test code there is no channel at all—no extra processing is required.

The last trick is correctly selecting real vs mock timers, and ensuring that they are associated with the correct TimeControl. Rust runs tests in parallel so there may be many tokio runtimes executing tests simultaneously. If we tried to use global storage, different tests will clobber each other.

The solution is thread-local storage (TLS). Tokio offers the ability to run code when each worker thread is started, which is an opportunity to store an Arc<TimeControl>. This is the purpose of the build_instrumented_runtime boilerplate in the test. When the code under test calls delay_for() it will dynamically create a Delay future of the correct variant:

thread_local!(
    static MOCK_CONTROL: RefCell<Option<Arc<TimeControl>>> = RefCell::new(None)
);
pub fn delay_for(duration: std::time::Duration) -> Delay {
    MOCK_CONTROL.with(|it| match it.borrow().as_ref() {
        Some(control) => {
            let deadline = control.now() + duration;
            Delay::new_mock(control, deadline)
        }
        None => Delay::Real(Box::pin(tokio::time::sleep(duration))),
    })
}

With all that explanation, let's review the test in full.

#[test]
fn test_timeout_occurs_fast() {
    let (time_control, _guard) = register_new_control();
    let rt = build_instrumented_runtime(&time_control);
    rt.block_on(async {
        let mut conn = Connection::new().await;
        time_control.advance(Duration::from_secs(10)).await;
        assert_eq!(conn.try_recv().unwrap_err(), TryRecvError::Disconnected);
    });
}

A TimeControl instance is created that the test will use to advance time, and will also collect registrations of mock timers.
A custom Runtime is created which injects the Arc<TimeControl> into TLS.
Inside that runtime, the Connection is created and a 10-second mock timer is registered.
advance() is called with a 10 second duration. This will advance to 10 seconds, permitting all of the code inside Connection to run. The channel gets closed.
The channel is definitely closed and reports Disconnected.

The second test can be adapted similarly. If the channel closes in 5 seconds or less, we're guaranteed to catch it.

#[test]
fn test_timeout_does_not_occur_fast() {
    let (time_control, _guard) = register_new_control();
    let rt = build_instrumented_runtime(&time_control);
    rt.block_on(async {
        let mut conn = Connection::new().await;
        time_control.advance(Duration::from_secs(5)).await;
        assert_eq!(conn.try_recv().unwrap_err(), TryRecvError::Empty);
    });
}

This technique has enabled Ditto to build fast-running unit tests on exactly the same async code our customers use, with only modest code modifications to account for the guards. Testing gives us the confidence to include business logic directly in our async layer, avoiding the overhead of structuring timer-driven code into sync and async components. We believe async Rust has a bright future and our customers are already seeing the benefits.

Visit us at www.ditto.live for more content

A database always capable of reading or writing information: Ditto Local Store

Ryan Ratner — Tue, 08 Feb 2022 22:45:44 +0000

If you aren’t familiar with Ditto, we are a real-time database for mobile, web, IoT, and server apps that can magically sync data with or even without the internet.

But what's really interesting about our database is that you can use Ditto in your apps as a regular document database. When we first created Ditto, we took three parallel tracks to building our entire system:

The database portion: the primary way that customers use Ditto today
Replication and CRDTs: how customers can mutate data in a distributed way and allow for automatic conflict resolution
The mesh network: how devices can talk to each other in a heterogenous way

Parts 2 and 3 only work if you call ditto.tryStartSync(). However, we wanted to design a system where the database portion was always capable of reading or writing information. That means if your application never sets up any network configuration permissions or calls ditto.tryStartSync(), the instance will only act as a local database.

The usage is incredibly simple. Here's an example using Swift

val ditto = Ditto()
// inserting a document
ditto.store["people"].upsert([
  "name": "Alice",
  "score": 21
])
// finding documents
let docs = ditto.store["people"].find('name == $args.name', [
  "name": "Alice"
])

and on Android

val ditto = Ditto(androidDependencies)
// inserting a document
ditto.store["people"].upsert(
  mapOf(
    "name" to "Alice",
    "score" to 21
    )
)

// finding documents
val docs = ditto.store["people"].find('name == $args.name', mapOf(
  "name" to "Alice"
))

For more information on our APIs and other supported programming languages checkout the concepts section of our documentation. All of the data store operations will work exactly the same as a syncing instance of Ditto. That means your application is completely capable of running a live query on a local instance of Ditto as well.

let ditto =  Ditto()
let liveQueries = ditto.store["cars"].findAll().observe({ documents in
  updateUI(documents)
})

Taking advantage of a local-only Ditto instance in the same app as a syncing Ditto instance

Each ditto instance is self-contained. That means that an app can run multiple ditto instances at the same time. All you need to do is to take a bit of care

// constructing a local only instance
let dittoLocalOnly = Ditto(path: '/foo1')

// constructing a syncing version of Ditto
let dittoSyncing = Ditto(identity: .onlinePlayground(appID: "my-cool-app"), path: '/foo2')
try! dittoSyncing.tryStartSync()

// notice that the path values are different to ensure that there are no file system collisions.

Why would you want a syncing instance and a non-syncing instance in the same app? There could be many reasons. Perhaps, your application may need to simply store data locally without replication, still using the same interface of a regular Ditto instance that you're accustomed to, while at the same time requiring to sync another instance. Syncing Ditto instances can control specific data that syncs or does not sync between devices, but some users may want the added simplicity of managing multiple Ditto instances with distinct functionality.

Looking Ahead

Ditto is an end-to-end platform product that can sync data regardless of connectivity from the edge to the cloud. That said, this is really our bread and butter, so a lot of our engineering efforts are focused on the performance of the replication, the mesh network, and our Big Peer (that lives in the cloud). This means that a lot of comparable database features aren't up to par with, say, something like SQLite or Realm. But don't worry, we are definitely getting there!

What are we lacking right now?

No indexing - Currently, we don't have any functionality for indexing your data. Don't worry, we are architecting this functionality as we speak.
Non-CRDT serializatation - Remember we built a syncing-database, so every document is a fully composable nested CRDT. Even if you use it as a local database, we still create and serialize a CRDT to disc.

What about JavaScript?

Currently, running Ditto as a local database using our JavaScript SDK comes with some caveats. Our web experience currently is in-memory only. That means if you were to use the npm package @dittolive/ditto in the web browser, your bundler will use our WebAssembly version. As a result, nothing in the store is persisted and a simple refresh of the web page will have all data evicted. We are actively working on a backing storage interface using the File System Access API for modern browsers and a fallback to IndexedDB this year.

The good news is that today you can get persistence if you are using our JS library in Electron or NodeJS. If you're curious why: unlike the browser, these platforms actually have native API access. So, when running Ditto in NodeJS or Electron environments, the NPM package will use a compiled native library as it's backing storage layer, and you will actually see Ditto files created!

A real app using Ditto as a local database - Vaxx

We wanted to test out how it would be to use Ditto as a local database in an app that would actually provide some use to the world. Since some cities have mandated that restaurants must check for vaccination records to receive service, we noticed that a lot of customers were endlessly scrolling on their phones to pull up a picture of their records. This was wasting a lot of the maitre d's time and slowing down seating. So we thought, what if we could create an app with a simple (yet edgy) name that would allow customers to immediately pull up their vaccine card? Well, with Ditto this is actually quite simple. We built it in just about 1 day and released it to the app store! Vaxx is running on Ditto, using it only as a local database, and doesn't sync anything at all. Vaxx is currently enjoying 55,000 active users with no promotion at all and still growing! In addition, if you just search for "Vaccine Card" in the U.S. Apple App Store, it's the first result.

We hope to create an Android version sometime in the coming months, so stay tuned!

DEV Community: Ditto

The Future of the Cloud? Make it Optional

From Cloud-Only to Cloud-Optional

Peer-to-peer is not new, but it is still hard to use

A shared set of tools

Footnotes

Testing CRDTs in Rust, From Theory to Practice

What Are CRDTs?

Implementing A New CRDT

Add-Wins Behavior

Remove-Wins Behavior

What Is Property-Based Testing?

Stateful Property-Based Testing

Poor Man's Rust Statem

A Model (Specification) and a SUT

Remove Wins vs Observed Add

The Test; A Counter Example

The Difference

The Fix?

Conclusion

DO cross the streams! Introducing Combine Support in Ditto - build complex workflows from multiple reactive streams

SQL JOIN-like behavior with Ditto using Combine

Making SQL JOIN-like behavior more efficient

Advanced Scenarios for SQL JOIN-like behavior

Footnotes

Mocking Time in Async Rust

A database always capable of reading or writing information: Ditto Local Store

Taking advantage of a local-only Ditto instance in the same app as a syncing Ditto instance

Looking Ahead

What about JavaScript?

A real app using Ditto as a local database - Vaxx

SQL `JOIN`-like behavior with Ditto using Combine

Making SQL `JOIN`-like behavior more efficient

Advanced Scenarios for SQL `JOIN`-like behavior