Bismuth

Algebraic Replicated Data Types

This manual introduces Algebraic Replicated Data Types (ARDTs): a systematic approach to building conflict-free replicated data types by composition of product, and sum types.

The merge semantics of a compound data type are automatically derived from the merge semantics of its components.

This document contains compilable code, you can download the full file.

Lattices

Every data type in this module is a join-semilattice. A lattice for a type A is a trait with a single operation: 

trait Lattice[A]:
  def merge(left: A, right: A): A

We usually will just import it 

import rdts.base.Lattice

The merge operation must be associative, commutative, and idempotent. Merging two states yields a consistent result containing all information from both sides.

If left `merge` right == right then left is subsumed by right: merging left into right is a no-op.

Lattices for existing types

To use a type as a lattice, we need to provide a given instance for that type. A safe way to do this is to use an existing ordering: 

given Lattice[Int] = Lattice.fromOrdering

Merging based on an ordering will discard the smaller of the two values.

Safe in this case means that we must be sure that the lattice properties (associative, commutative, idempotent) are guaranteed.

The library comes with existing instances for built-in Scala types that you can just import. 

import rdts.base.Lattice.given
import rdts.datatypes.*

Now you can merge any two values of a type that has a given Lattice instance: 

test("set lattice via union"):
    val setA: Set[Int] = Set(1, 2, 3)
    val setB: Set[Int] = Set(3, 4, 5)
    val mergedSet      = setA `merge` setB
    assertEquals(mergedSet, Set(1, 2, 3, 4, 5))

Maps merge key-wise using the lattice of their values: 

test("map lattice key-wise"):
    val mapA: Map[String, Int] = Map("x" -> 1, "y" -> 2)
    val mapB: Map[String, Int] = Map("y" -> 3, "z" -> 4)
    val mergedMap              = mapA `merge` mapB
    assertEquals(mergedMap("x"), 1)
    assertEquals(mergedMap("y"), 3)
    assertEquals(mergedMap("z"), 4)

Options form a lattice where None is smaller than Some: 

test("option lattice"):
    val optA: Option[Int] = None
    val optB: Option[Int] = Some(5)
    assertEquals(optA `merge` optB, Some(5))
    assertEquals(optB `merge` optA, Some(5))

Bottom

Bottom[A] provides the empty value of a lattice. It is the identity element of merge: merging an empty value into any other value leaves it unchanged. 

import rdts.base.Bottom
import rdts.base.Bottom.given

test("bottom is the identity of merge"):

    val set: Set[Int] = Set(1, 2, 3)
    val emptySet      = Bottom.empty[Set[Int]]
    assert(emptySet.isEmpty, "the empty set is a bottom value")

    // merging nothing into a state is a no-op
    assertEquals(emptySet `merge` set, set)

The empty set, the empty map, and None are all examples of bottom values.

Some Operations Have No Effect

Because merge only ever adds information, removing information is impossible through merges alone: operations that try to forget previously merged state are no-ops.

A plain grow-only set, for example, cannot forget an element. It can compute a smaller value, but merging that value back into the original set immediately re-adds what was removed: 

test("removal has no effect in a set lattice"):
    val original: Set[Int] = Set(1, 2)
    // removing 1 gives a smaller set ...
    val removed: Set[Int] = original.excl(1)
    assertEquals(removed, Set(2))

    // ... but merging it back into the original set re-adds 1
    assertEquals(original `merge` removed, original)

The removed value is Set(2), yet merging it back yields Set(1, 2) merge Set(2), which is Set(1, 2) again: the removal is lost because set union is monotone.

To support removals, replicated data types need more elaborate representations (such as tombstones).

Automatic Lattice Derivation

The core idea of ARDTs is that lattices can be automatically derived from the structure of data types. No manual merge logic is needed.

Product Lattices

For a product type like a case class, merge is component-wise. Each field is merged independently using its own lattice. You get the derivation with a single line: 

test("product lattice derives component-wise merge"):
    case class Data(counter: Int, items: Set[Int], mapping: Map[String, Int], flag: Option[Int])

    // automatically derive the lattice instance
    given Lattice[Data] = Lattice.derived

    val left  = Data(1, Set(1, 2), Map("a" -> 4), Some(1))
    val right = Data(2, Set(2, 3), Map("a" -> 1, "b" -> 2), None)

    val merged = left `merge` right
    assertEquals(merged.counter, 2)          // max(1, 2)
    assertEquals(merged.items, Set(1, 2, 3)) // union
    assertEquals(merged.mapping, Map("a" -> 4, "b" -> 2)) // key-wise merge
    assertEquals(merged.flag, Some(1)) // Some > None

Sum Lattices

For sum types (enums, sealed classes), we rely on their inherent total ordering to derive the lattice instance. Constructors that are defined later are considered larger. Sum lattices are not as common as product lattices, but do allow modeling state machines. Consider the following example, where we have a Workflow that first prepares some documents, then drafts a contract, and then finalizes the process. 

enum Workflow:
    case Init()
    case Documents(hasStaffSheet: Boolean = false, hasHourConfirmation: Boolean = false)
    case Contract(signed: Boolean = false)
    case Complete()

given Lattice[Workflow] = {
  // each case of the sum is a Product type, so we need to derive lattices for each
  given Lattice[Boolean]            = Lattice.fromOrdering
  given Lattice[Workflow.Init]      = Lattice.derived
  given Lattice[Workflow.Documents] = Lattice.derived
  given Lattice[Workflow.Contract]  = Lattice.derived
  given Lattice[Workflow.Complete]  = Lattice.derived
  Lattice.sumLattice
}

test("sum lattice for state machine"):
    val wf0: Workflow = Workflow.Init()
    val wf1           = wf0 `merge` Workflow.Documents()
    assertEquals(wf1, Workflow.Documents())

The ordering is Init < Documents < Contract < Complete. Once a workflow reaches Complete, merging any earlier state leaves it at Complete. By itself, the Workflow state machine does not really guarantee that steps are taken in the correct order; we could just skip directly from Init to Complete, without ever verifying the documents or the contract. We can fix this by using update methods that check preconditions. 

import Workflow.*
extension (wf: Workflow) {

  def newProcess: Workflow          = Documents()
  def addStaffSheet: Workflow       = Documents(hasStaffSheet = true)
  def addHourConfirmation: Workflow = Documents(hasHourConfirmation = true)

  def createContract: Workflow = wf match
      case Documents(hasStaffSheet = true, hasHourConfirmation = true) => Contract()
      case other                                                       => Init()

  def signContract: Workflow = wf match
      case Contract(signed = false) => Contract(true)
      case other                    => Init()

  def complete: Workflow = wf match
      case Contract(signed = true) => Complete()
      case other                   => Init()
}

test("workflow"):
    import rdts.syntax.DeltaBuffer

    val db     = DeltaBuffer(Workflow.Init())
    val result = db
      .mod(_.newProcess)
      .mod(_.addStaffSheet)
      .mod(_.addHourConfirmation)
      .mod(_.createContract)
      .mod(_.signContract)
      .mod(_.complete)

    assertEquals(result.state, Workflow.Complete())

If we now leave out any of the steps above, then we will not complete the workflow. Note that it’s not important which replica calls any of the above methods, only that it is in the corresponding state. So anyone could progress the workflow. We could further restrict this to specific replicas/roles by adding those roles to the precondition. If we don’t trust all of the replicas, we need some form of access control, which is currently an open research topic.

Designing Replicated Data Types

Now that we can derive lattices from products and sums, we can design replicated data types.

The key motivation for ARDTs is that they provide automatic convergence: no matter in which order replicas merge states, they always reach the same final result. This solves the hard problem of distributed convergence, but the resulting value may not be semantically useful.

Consider counting votes in an election. If we use a plain Int, two replicas each counting one vote would produce values 1 and 1. Merging with max gives 1, losing a vote! The replicas agree on 1, but the result is wrong.

ARDTs shift the problem from ensuring convergence (which is hard) to designing good data types (which is much easier). We need to design data types where concurrent operations do not overwrite each other.

A common trick is to use replica identities. Each replica writes to its own part of the state, so concurrent writes never conflict. The library provides a built in class to track unique IDs and the specific Uid of the local replica (which is a different type, to prevent accidential misuse). 

import rdts.base.Uid
import rdts.base.ReplicaId

A replicated counter

Let us design a counter that counts votes correctly. Each replica tracks its own contribution in a map from replica id to count. The total is the sum of all entries. Since each replica only writes to its own key, no information is lost on merge. 

// A Counter is just a map: each replica tracks its own count
case class Counter(counters: Map[Uid, Int] = Map.empty):
    def value: Int = counters.values.sum

    def add(amount: Int)(using replicaId: ReplicaId): Counter =
      Counter(Map(replicaId.uid -> (counters.getOrElse(replicaId.uid, 0) + amount)))

// We can derive the lattice automatically
given Lattice[Counter] = Lattice.derived

Note how the methods only return delta states: small values that need to be merged into the state to take effect. 

test("Using a counter"):

    given ReplicaId = ReplicaId.predefined("replica-alice")

    var counter = Counter()

    // inc() returns a delta that we merge into the state
    counter = counter `merge` counter.add(1)
    assertEquals(counter.value, 1)

    // merging deltas from multiple replicas adds their contributions
    val deltaAlice = counter.add(1)
    val deltaBob   =
        given ReplicaId = ReplicaId.predefined("replica-bob")
        counter.add(1)

    val total = counter `merge` deltaAlice `merge` deltaBob
    assertEquals(total.value, 3)

Our Counter only grows. If we try to add a negative value, the map merge with max would ignore it because the existing positive value is already larger: 

test("Counter cannot decrease"):
    given ReplicaId = ReplicaId.predefined("replica-alice")

    var counter = Counter()
    counter = counter `merge` counter.add(1)
    assertEquals(counter.value, 1)

    // trying to "un-count" by adding -1 has no effect
    counter = counter `merge` counter.add(-1)
    assertEquals(counter.value, 1) // still 1, because max(1, -1) == 1

We need a counter that can go both up and down. The solution is a product of two Counters: one for positive increments, one for negative increments. The total is the difference. The product lattice derives component-wise merge automatically. 

// A PosNegCounter is a product of two Counters: one for positive, one for negative
case class PosNeg(pos: Counter, neg: Counter):
    def value: Int                                = pos.value - neg.value
    def add(amount: Int)(using ReplicaId): PosNeg =
      if amount >= 0 then PosNeg(pos.add(amount), Counter())
      else PosNeg(Counter(), neg.add(-amount))

given Lattice[PosNeg] = Lattice.derived

test("PosNegCounter from scratch"):
    given ReplicaId = ReplicaId.predefined("replica-alice")

    var pn = PosNeg(Counter(), Counter())
    pn = pn `merge` pn.add(5)
    pn = pn `merge` pn.add(3)
    assertEquals(pn.value, 8)

    pn = pn `merge` pn.add(-4)
    assertEquals(pn.value, 4)

test("PosNegCounter concurrent from scratch"):
    given ReplicaId = ReplicaId.predefined("replica-alice")

    val deltaA = PosNeg(Counter(), Counter()).add(5)
    val deltaB =
        given ReplicaId = ReplicaId.predefined("replica-bob")
        PosNeg(Counter(), Counter()).add(3)

    val merged = PosNeg(Counter(), Counter()) `merge` deltaA `merge` deltaB
    assertEquals(merged.value, 8)

Causal Time and Ordering-Based Data Types

A common way to allow any data to be replicated is to order values by when they were created. The last write (largest timestamp) wins.

CausalTime is a causally consistent wall-time clock that provides a total order across replicas. It has three components:

  • time: wall-clock milliseconds

  • causal: a monotonic counter for concurrent updates in the same ms

  • random: a tie-breaker when all else is equal

LastWriterWins[A] pairs a CausalTime with a payload A. Its lattice picks the value with the later timestamp: 

test("LastWriterWins"):
    val lww1 = LastWriterWins.now("hello")
    val lww2 = LastWriterWins.now("world")

    // the later timestamp wins
    val mergedLWW = lww1 `merge` lww2
    assertEquals(mergedLWW.read, "world")

Common Replicated Data Types

GrowOnlyCounter

An increment-only counter. Internally it is a map from replica id to count. Each inc() or add(n) produces a delta. Needs an identity so each replica can write to its own map key. 

test("GrowOnlyCounter"):
    val localId  = rdts.base.ReplicaId.predefined("replica-alice")
    val remoteId = rdts.base.ReplicaId.predefined("replica-bob")

    var gCounter = GrowOnlyCounter.zero

    gCounter = gCounter `merge` gCounter.inc()(using localId)
    gCounter = gCounter `merge` gCounter.add(5)(using localId)

    assertEquals(gCounter.value, 6)

    gCounter = gCounter `merge` GrowOnlyCounter.zero.inc()(using remoteId)
    assertEquals(gCounter.value, 7)

PosNegCounter (PNCounter)

A counter for both increment and decrement. It is a product of two Counters (positive and negative), merged component-wise. Needs an identity. 

test("PosNegCounter"):
    val localId = rdts.base.ReplicaId.predefined("replica-alice")

    val pnCounter = PosNegCounter.zero

    val pn1 = pnCounter.add(5)(using localId)
    val pn2 = pnCounter.add(-3)(using localId)

    val mergedPN = pnCounter `merge` pn1 `merge` pn2
    assertEquals(mergedPN.value, 2)

GrowOnlySet (G-Set)

A set that only supports adding elements. Merge is set union. No identity needed: elements are self-contained. 

test("GrowOnlySet"):
    val gSet = GrowOnlySet.empty[String]

    val gs1 = gSet.add("apple")
    val gs2 = gSet.add("banana")

    val mergedGS = gSet `merge` gs1 `merge` gs2
    assertEquals(mergedGS.set, Set("apple", "banana"))

    val gs3       = gSet.add("apple")
    val mergedGS2 = mergedGS `merge` gs3
    assertEquals(mergedGS2.set, Set("apple", "banana"))

GrowOnlyList (G-List)

An append-only list that also supports insertion at an index. Elements are ordered by causal timestamps. Identity is used to allocate fresh timestamps for inserted elements. 

test("GrowOnlyList"):
    val gList = GrowOnlyList.empty[String]

    val gl1 = gList.insertAfter(GrowOnlyList.headDot, "a")
    val gl2 = gl1.insertAfter(GrowOnlyList.headDot, "b")

    val mergedGL = gList `merge` gl1 `merge` gl2
    assertEquals(mergedGL.toList, List("b", "a"))

ReplicatedSet

A set supporting add and remove (Observed-Remove Set). Removals do not override concurrent adds: the add wins. Needs an identity to create fresh dots for added elements. 

test("ReplicatedSet"):
    val localId = rdts.base.ReplicaId.predefined("replica-alice")

    var rSet = ReplicatedSet.empty[String]

    rSet = rSet `merge` rSet.add("x")(using localId)
    rSet = rSet `merge` rSet.add("y")(using localId)

    assertEquals(rSet.elements, Set("x", "y"))

    rSet = rSet `merge` rSet.remove("x")
    assertEquals(rSet.elements, Set("y"))

EnableWinsFlag (EWFlag)

A boolean flag where enable always wins over concurrent disable. Needs an identity to allocate fresh dots on enable. 

test("EnableWinsFlag"):
    val localId = rdts.base.ReplicaId.predefined("replica-alice")

    val flag = EnableWinsFlag.empty

    val enabled = flag.enable(using localId)()
    assert(enabled.read == true, "enable should set flag")

    val dEnabled   = flag.disable()
    val dDisabled  = flag.enable(using localId)()
    val mergedFlag = dEnabled `merge` dDisabled
    assert(mergedFlag.read == true, "enable should win over disable")

MultiVersionRegister (MV-Register)

A register that holds at most one value without concurrent writes, or all concurrently written values as a set. Needs an identity to create fresh dots for written values. 

test("MultiVersionRegister"):
    val localId  = rdts.base.ReplicaId.predefined("replica-alice")
    val remoteId = rdts.base.ReplicaId.predefined("replica-bob")

    val mvReg = MultiVersionRegister.empty[String]

    val afterFirst = mvReg `merge` mvReg.write("first")(using localId)
    assertEquals(afterFirst.read, Set("first"))

    val afterConcurrent = afterFirst `merge` mvReg.write("second")(using remoteId)
    assertEquals(afterConcurrent.read, Set("first", "second"))

    val afterMerge = afterConcurrent `merge` afterConcurrent.write("third")(using localId)
    assertEquals(afterMerge.read, Set("third"))

ObserveRemoveMap (OR-Map)

A map with key-value semantics. Removals are tracked via dots and do not override concurrent updates. Needs an identity for update operations. 

test("ObserveRemoveMap"):
    given Lattice[String] = Lattice.assertEquals
    val localId           = rdts.base.ReplicaId.predefined("replica-alice")

    val orMap = ObserveRemoveMap.empty[String, String]

    val afterM1 = orMap `merge` orMap.update("key1", "value1")(using localId)
    assert(afterM1.get("key1") == Some("value1"), "should have key1")

    val mRemove          = afterM1.remove("key1")
    val mAdd             = afterM1.update("key1", "value2")(using localId)
    val afterConcurrent2 = mRemove `merge` mAdd
    assert(afterConcurrent2.get("key1") == Some("value2"), "add should win over remove")

ReplicatedList (RGA)

A replicated list with insert, update, and delete. Uses a directed graph of dots for causal ordering. Needs an identity for all modifying operations. 

test("ReplicatedList"):
    val localId = rdts.base.ReplicaId.predefined("replica-alice")

    var rList = ReplicatedList.empty[String]

    rList = rList `merge` rList.append("a")(using localId)
    rList = rList `merge` rList.append("b")(using localId)
    assertEquals(rList.toList, List("a", "b"))

    rList = rList `merge` rList.insertAt(1, "c")(using localId)
    assertEquals(rList.toList, List("a", "c", "b"))

    rList = rList `merge` rList.delete(1)
    assertEquals(rList.toList, List("a", "b"))

Epoch Wrapper

Epoch[E] adds a monotonic counter to any datatype. A higher epoch wins over a lower epoch. Useful for wrapping types with weak merge semantics. 

test("Epoch"):
    val localId = rdts.base.ReplicaId.predefined("replica-alice")

    val epoch = Epoch.empty[GrowOnlyCounter]

    val ep1 = epoch.write(epoch.value.inc()(using localId))
    val ep2 = epoch.write(epoch.value.inc()(using localId))

    val mergedEpoch = epoch `merge` ep1 `merge` ep2
    assertEquals(mergedEpoch.read.value, 1)

The End

Don’t worry about the closing brace :-) 

}