A simple Reliable Causal Broadcast implementation using F# and Akka.NET

Program.fs

module Program =
        
    type InMemoryDb(replica: ReplicaId) =
        let snapshot = ref null
        let mutable events : Map<uint64,obj> = Map.empty
        interface Db with
            member _.SaveSnapshot state = async { snapshot := (box state) } 
            member _.LoadSnapshot<'s>() = async {
                match !snapshot with
                | :? 's as state -> return Some state
                | _ -> return None
            }
            member _.LoadEvents<'e>(from) = asyncSeq {
                for (seqNr, e) in Map.toSeq events do
                    if seqNr >= from then
                        let casted : Event<'e> = downcast e
                        yield casted
            }
            member _.SaveEvents evts = async {
                for event in evts do
                    events <- Map.add event.SeqNr (box event) events
            }
    
    let main () =
        let sys = System.create "sys" <| Configuration.parse "akka.loglevel = DEBUG"
        let a = spawn sys "A" <| props (ORSet.props (InMemoryDb "A") "A")
        let b = spawn sys "B" <| props (ORSet.props (InMemoryDb "B") "B")
        async {
            a <! Connect("B", b)
            b <! Connect("A", a)
            
            let! state = ORSet.add 1L a
            printfn "State on node A (first): %A" state
            
            do! Async.Sleep 1000
            
            let! state = ORSet.add 2L a
            printfn "State on node A (second): %A" state
            
            do! Async.Sleep 5000
            
            let! state = ORSet.query b
            printfn "State on node B (after sync): %A" state
            let! state = ORSet.remove 2L b
            printfn "State on node B (after update): %A" state
            
            do! Async.Sleep 5000
            
            let! state1 = ORSet.query a
            let! state2 = ORSet.query b
            
            assert (state1 = state2)
            printfn "SUCCESS"
        } |> Async.RunSynchronously
        0

ReliableCausalBroadcast.fs

/// Reliable causal broadcast implementation
module DemoFs.RCB

open Akka.Actor
open Akkling
open System
open FSharp.Control

type ReplicaId = String
type Ord = 
    | Lt = -1  // lower
    | Eq = 0   // equal
    | Gt = 1   // greater
    | Cc = 2   // concurrent

[<RequireQualifiedAccess>]
module Helpers =

    /// Helper method for insert-or-update semantic for Map
    let upsert k v fn map =
        match Map.tryFind k map with
        | None -> Map.add k v map
        | Some v -> Map.add k (fn v) map
   
type VTime = Map<ReplicaId, int64>

[<RequireQualifiedAccess>]     
module Version =  

    let zero: VTime = Map.empty
    
    let inc r (vv: VTime): VTime = vv |> Helpers.upsert r 1L ((+)1L)

    let set r ts (vv: VTime): VTime = Map.add r ts vv

    let merge (vv1: VTime) (vv2: VTime) =
        vv2 |> Map.fold (fun acc k v2 -> Helpers.upsert k v2 (max v2) acc) vv1

    let compare (a: VTime) (b: VTime): Ord = 
        let valOrDefault k map =
            match Map.tryFind k map with
            | Some v -> v
            | None   -> 0L
        let akeys = a |> Map.toSeq |> Seq.map fst |> Set.ofSeq
        let bkeys = b |> Map.toSeq |> Seq.map fst |> Set.ofSeq
        (akeys + bkeys)
        |> Seq.fold (fun prev k ->
            let va = valOrDefault k a
            let vb = valOrDefault k b
            match prev with
            | Ord.Eq when va > vb -> Ord.Gt
            | Ord.Eq when va < vb -> Ord.Lt
            | Ord.Lt when va > vb -> Ord.Cc
            | Ord.Gt when va < vb -> Ord.Cc
            | _ -> prev ) Ord.Eq

/// A contaienr for user-defined events of type 'e. It contains metadata necessary
/// to partially order and distribute events in peer-to-peer fashion. 
type Event<'e> =
    { /// The replica, on which this event originally was created.
      Origin: ReplicaId
      /// The sequence number given by the origin replica at the moment of event creation.
      /// This allows us to keep track of replication progress with remote replicas even
      /// when we didn't received their events directly, but via intermediate replica. 
      OriginSeqNr: uint64
      /// The sequence number given by the local replica. For events created by current replica
      /// it's the same as `OriginSeqNr`. For replicated events it's usually higher.
      LocalSeqNr: uint64
      /// Vector clock which describes happened-before relationships between events from
      /// different replicas, enabling to establish partial order among them.
      Version: VTime
      /// An user-defined event data.
      Data: 'e }
    override this.ToString() = sprintf "(%s, %i, %i, %A, %O)" this.Origin this.OriginSeqNr this.LocalSeqNr this.Version this.Data 

type Endpoint<'s,'c,'e> = IActorRef<Protocol<'s,'c,'e>>

and Protocol<'s,'c,'e> =
    /// Attaches another replica to continuously synchronize with current one.
    | Connect of replicaId:ReplicaId * Endpoint<'s,'c,'e>
    /// Request to read up to `maxCount` events from a given replica starting from `seqNr`. Additionally a `filter`
    /// is provided to deduplicate possible events on the sender side (it will be then used second time on receiver side).
    /// This message is expected to be replied with `Recovered`, which contains all events satisfying seqNr/filter criteria. 
    | Replicate of seqNr:uint64 * maxCount:int * filter:VTime * replyTo:Endpoint<'s,'c,'e>
    | ReplicateTimeout of ReplicaId
    /// Response to `Recover` - must always be send. Empty content notifies about end of event stream. `toSeqNr` informs
    /// up to which sequence number this message advanced.
    | Replicated of from:ReplicaId * toSeqNr:uint64 * events:Event<'e>[]
    /// Request for a state. It should be replied with state being application of `Crdt.Query` over `ReplicationState.Crdt`.
    | Query
    /// Persists an event into current replica. Replied with updated, materialized state after success. 
    | Command of 'c
    /// Message send at the beginning of recovery phase with the latest persisted snapshot of the state (if there was any) 
    | Loaded of ReplicationState<'s>
    /// Periodic trigger to persist current state snapshot (only performed if state has changed since last snapshot, tracked by IsDirty flag). 
    | Snapshot
        
and ReplicationState<'s> =
    { /// Unique identifier of a given replica/node.
      Id: ReplicaId
      /// Checks if replication state has been modified after being persisted.
      IsDirty: bool
      /// Counter used to assign unique sequence number for the events to be stored locally.
      SeqNr: uint64
      /// Version vector describing the last observed event.
      Version: VTime
      /// Sequence numbers of remote replicas. When synchronizing (via `Recover` message) with remote replicas,
      /// we start doing so from the last known sequence numbers we received.
      Observed: Map<ReplicaId, uint64>
      /// CRDT object that is replicated.
      Crdt: 's }
    
[<RequireQualifiedAccess>]
module ReplicationState =
    let inline create (id: ReplicaId) state = { Id = id; IsDirty = false; SeqNr = 0UL; Version = Map.empty; Observed = Map.empty; Crdt = state }
    
    /// Checks if current event has NOT been observed by a replica identified by state. Unseen events are those, which
    /// have SeqNr higher than the highest observed sequence number on a given node AND their version vectors were not
    /// observed (meaning they are either greater or concurrent to current node version). 
    let unseen nodeId (state: ReplicationState<'s>) (e: Event<'e>) =
        match Map.tryFind nodeId state.Observed with
        | Some ver when e.OriginSeqNr <= ver -> false
        | _ -> (Version.compare e.Version state.Version) > Ord.Eq
        
[<Interface>]           
type Db =
    abstract SaveSnapshot: 's -> Async<unit>
    abstract LoadSnapshot: unit -> Async<'s option>
    abstract LoadEvents: startSeqNr:uint64 -> AsyncSeq<Event<'e>>
    abstract SaveEvents: events:Event<'e> seq -> Async<unit>

/// Use database `cursor` to read up to `count` elements and send them to the `target` as Recovered message.
/// Send only entries that have keys starting with a given `prefix` (eg. events belonging to specific nodeId).
/// Use `filter` to skip events that have been seen by the `target`. 
let replay (nodeId: ReplicaId) (filter: VTime) (target: Endpoint<'s,'c,'e>) (events: AsyncSeq<Event<'e>>) (count:int) = async {
    let buf = ResizeArray()
    let mutable cont = count > 0
    let mutable i = 0
    let mutable lastSeqNr = 0UL
    use cursor = events.GetEnumerator()
    while cont do
        match! cursor.MoveNext() with
        | Some e ->
            if Version.compare e.Version filter > Ord.Eq then
                buf.Add(e)
                i <- i + 1
            cont <- i < count
            lastSeqNr <- Math.Max(lastSeqNr, e.LocalSeqNr)
        | _ -> cont <- false
    let events = buf.ToArray()
    target <! Replicated(nodeId, lastSeqNr, events)
}
    
let recoverTimeout = TimeSpan.FromSeconds 5.
                   
type ReplicationStatus<'s,'c,'e> =
    { /// Access point for the remote replica.
      Endpoint: Endpoint<'s,'c,'e>
      /// Cancellation token for pending `RecoverTimeout`.
      Timeout: ICancelable }
    
[<Interface>]
type Crdt<'crdt,'state,'cmd,'event> =
    /// Get a default (zero) value of the CRDT.
    abstract Default: 'crdt
    /// Given a CRDT state return an actual value that user has interest in. Eg. ORSet still has to carry
    /// metadata timestamps, however from user perspective materialized value of ORSet is just ordrinary Set<'a>. 
    abstract Query: 'crdt -> 'state
    /// Equivalent of command handler in eventsourcing analogy.
    abstract Prepare: state:'crdt * command:'cmd -> 'event
    /// Equivalent of event handler in eventsourcing analogy.
    abstract Effect: state:'crdt * event:Event<'event> -> 'crdt
                   
let replicator (crdt: Crdt<'crdt,'state,'cmd,'event>) (db: Db) (id: ReplicaId) (ctx: Actor<Protocol<_,_,_>>) =    
    /// Cancel last pending `RecoverTimeout` task, and schedule it again.
    let refreshTimeouts nodeId progresses (ctx: Actor<_>) =
        let p = Map.find nodeId progresses
        p.Timeout.Cancel()
        let timeout = ctx.Schedule recoverTimeout ctx.Self (ReplicateTimeout nodeId)
        Map.add nodeId { p with Timeout = timeout } progresses
    
    let rec active (db: Db) (state: ReplicationState<'crdt>) (replicatingNodes: Map<ReplicaId, ReplicationStatus<'crdt,'cmd,'event>>) (ctx: Actor<_>) = actor {
        match! ctx.Receive() with
        | Query ->
            ctx.Sender() <! crdt.Query state.Crdt
            return! active db state replicatingNodes ctx
            
        | Replicate(from, count, filter, sender) ->
            logDebugf ctx "received recover request from %s: seqNr=%i, vt=%O" sender.Path.Name from filter
            let cursor = db.LoadEvents(from)
            replay state.Id filter sender cursor count |> Async.Start
            return! active db state replicatingNodes ctx 
            
        | Replicated(nodeId, lastSeqNr, [||]) ->
            // if we received empty event list, this node is up to date with `nodeId`
            // just schedule timeout, so when it happens we ask to Recover again
            logDebugf ctx "%s reached end of updates" nodeId
            let prog = refreshTimeouts nodeId replicatingNodes ctx
            let observedSeqNr = Map.tryFind nodeId state.Observed |> Option.defaultValue 0UL
            if lastSeqNr > observedSeqNr then
                let nstate = { state with Observed = Map.add nodeId lastSeqNr state.Observed }
                do! db.SaveSnapshot(nstate.Id, nstate)
                return! active db nstate prog ctx
            else
                return! active db state prog ctx
            
        | Replicated(nodeId, lastSeqNr, events) ->
            let mutable nstate = state
            let mutable remoteSeqNr = Map.tryFind nodeId nstate.Observed |> Option.defaultValue 0UL
            let toSave = ResizeArray()
            // for all events not seen by the current node, rewrite them to use local sequence nr, update the state
            // and save them in the database
            for e in events |> Array.filter (ReplicationState.unseen nodeId state) do
                logDebugf ctx "replicating event %O from replica %s" e nodeId
                let seqNr = nstate.SeqNr + 1UL
                let version = Version.merge nstate.Version e.Version // update current node version vector
                remoteSeqNr <- Math.Max(remoteSeqNr, e.LocalSeqNr) // increment observed remote sequence nr
                let nevent = { e with LocalSeqNr = seqNr }
                nstate <- { nstate with
                              Crdt = crdt.Effect(nstate.Crdt, nevent)
                              SeqNr = seqNr
                              Version = version
                              Observed = Map.add nodeId remoteSeqNr nstate.Observed }
                toSave.Add nevent
            do! db.SaveEvents toSave // save all unseen events together with updated state
            //do! db.SaveSnapshot nstate // in practice snapshot should be applied on condition (ideally in the same transaction)
            let target = Map.find nodeId replicatingNodes
            target.Endpoint <! Replicate(lastSeqNr+1UL, 100, nstate.Version, ctx.Self) // continue syncing 
            let prog = refreshTimeouts nodeId replicatingNodes ctx
            return! active db { nstate with IsDirty = true } prog ctx
            
        | ReplicateTimeout nodeId ->
            // if we didn't received Recovered in time or the last one was empty, upon timeout just retry the request
            logDebugf ctx "%s didn't replied to read request in time" nodeId
            let seqNr = Map.tryFind nodeId state.Observed |> Option.defaultValue 0UL
            let p = Map.find nodeId replicatingNodes
            p.Endpoint <! Replicate(seqNr+1UL, 100, state.Version, ctx.Self) 
            let timeout = ctx.Schedule recoverTimeout ctx.Self (ReplicateTimeout nodeId)
            let prog = Map.add nodeId { p with Timeout = timeout } replicatingNodes
            return! active db state prog ctx
            
        | Command(cmd) ->
            let sender = ctx.Sender()
            let seqNr = state.SeqNr + 1UL
            let version = Version.inc state.Id state.Version
            let data = crdt.Prepare(state.Crdt, cmd) // handle the command, produce event
            let event = { Origin = state.Id; OriginSeqNr = seqNr; LocalSeqNr = seqNr; Version = version; Data = data }
            let ncrdt = crdt.Effect(state.Crdt, event) // update the state with produced event
            let nstate = { state with Version = version; SeqNr = seqNr; Crdt = ncrdt }
            // store new event atomically with updated state
            do! db.SaveEvents [event]
            logDebugf ctx "stored event %O in a database" event
            sender <! crdt.Query ncrdt // send updated materialized CRDT state back to the sender
            return! active db { nstate with IsDirty = true } replicatingNodes ctx
                        
        | Connect(nodeId, endpoint) ->
            // connect with the remote replica, and start synchronizing with it
            let seqNr = Map.tryFind nodeId state.Observed |> Option.defaultValue 0UL
            endpoint <! Replicate(seqNr+1UL, 100, state.Version, ctx.Self)
            logDebugf ctx "connected with replica %s. Sending read request starting from %i" nodeId (seqNr+1UL)
            let timeout = ctx.Schedule recoverTimeout ctx.Self (ReplicateTimeout nodeId)
            return! active db state (Map.add nodeId { Endpoint = endpoint; Timeout = timeout } replicatingNodes) ctx
            
        | Snapshot when state.IsDirty ->
            logDebugf ctx "Snapshot triggered"
            let nstate = { state with IsDirty = false }
            do! db.SaveSnapshot nstate
            return! active db nstate replicatingNodes ctx
            
        | _ -> return Unhandled
    }
    
    let rec recovering (db: Db) (ctx: Actor<_>) = actor {
        match! ctx.Receive() with
        | Loaded state ->
            logDebugf ctx "Recovery phase done with state: %O" state
            ctx.UnstashAll()
            let interval = TimeSpan.FromSeconds 5.
            ctx.ScheduleRepeatedly interval interval ctx.Self Snapshot |> ignore
            return! active db state Map.empty ctx
            
        | _ ->
            // stash all other operations until recovery is complete
            ctx.Stash()
            return! recovering db ctx
    }
    
    async {
        // load state from DB snapshot or create a new empty one
        let! snapshot = db.LoadSnapshot()
        let mutable state = snapshot |> Option.defaultValue (ReplicationState.create id crdt.Default)
        // apply all events that happened since snapshot has been made
        for event in db.LoadEvents (state.SeqNr + 1UL) do
            state <- { state with
                          Crdt = crdt.Effect(state.Crdt, event)
                          SeqNr = event.LocalSeqNr
                          Version = Version.merge event.Version state.Version
                          Observed = Map.add event.Origin event.OriginSeqNr state.Observed }
        
        ctx.Self <! Loaded state
    } |> Async.Start
    
    recovering db ctx

[<RequireQualifiedAccess>]
module Counter =
    
    let private crdt =
        { new Crdt<int64,int64,int64,int64> with
            member _.Default = 0L
            member _.Query crdt = crdt
            member _.Prepare(_, op) = op
            member _.Effect(counter, e) = counter + e.Data }
    
    /// Used to create replication endpoint handling operation-based Counter protocol.
    let props db replica ctx = replicator crdt db replica ctx
    
    /// Increment counter maintainer by given `ref` endpoint by a given delta (can be negative).
    let inc (by: int64) (ref: Endpoint<int64,int64,int64>) : Async<int64> = ref <? Command by
    
    /// Retrieve the current state of the counter maintained by the given `ref` endpoint. 
    let query (ref: Endpoint<int64,int64,int64>) : Async<int64> = ref <? Query
    
[<RequireQualifiedAccess>]    
module ORSet =
    
    type ORSet<'a> when 'a: comparison = Set<'a * VTime>
    
    type Command<'a> =
        | Add of 'a
        | Remove of 'a
    
    type Operation<'a> =
        | Added of 'a
        | Removed of Set<VTime>
    
    type Endpoint<'a> when 'a: comparison = Endpoint<ORSet<'a>, Command<'a>, Operation<'a>>
    
    let private crdt : Crdt<ORSet<'a>, Set<'a>, Command<'a>, Operation<'a>> =
        { new Crdt<_,_,_,_> with
            member _.Default = Set.empty
            member _.Query(orset) = orset |> Set.map fst   
            member _.Prepare(orset, cmd) =
                match cmd with
                | Add item -> Added(item)
                | Remove item ->
                    let timestamps =
                        orset
                        |> Set.filter (fun (i, _) -> i = item)
                        |> Set.map snd
                    Removed timestamps
            member _.Effect(orset, e) =
                match e.Data with
                | Added(item) -> Set.add (item, e.Version) orset
                | Removed(versions) -> orset |> Set.filter (fun (_, ts) -> not (Set.contains ts versions)) }
    
    /// Used to create replication endpoint handling operation-based ORSet protocol.
    let props db replicaId ctx = replicator crdt db replicaId ctx
    
    /// Add new `item` into an ORSet maintained by the given `ref` endpoint. In case of add/remove conflicts add wins. 
    let add (item: 'a) (ref: Endpoint<'a>) : Async<Set<'a>> = ref <? Command (Add item)
    
    /// Remove an `item` from the ORSet maintained by the given `ref` endpoint. In case of add/remove conflicts add wins.
    let remove (item: 'a) (ref: Endpoint<'a>) : Async<Set<'a>> = ref <? Command (Remove item)
    
    /// Retrieve the current state of the ORSet maintained by the given `ref` endpoint. 
    let query (ref: Endpoint<'a>) : Async<Set<'a>> = ref <? Query
    
[<RequireQualifiedAccess>]    
module LWWRegister =
        
    [<Struct>]
    type LWWRegister<'a> =
        { Timestamp: struct(DateTime * ReplicaId)
          Value: 'a voption }
        
    type Operation<'a> = DateTime * 'a voption
    
    type Endpoint<'a> = Endpoint<LWWRegister<'a>, 'a voption, Operation<'a>>
    
    let private crdt : Crdt<LWWRegister<'a>, 'a voption, 'a voption, Operation<'a>> =
        { new Crdt<_,_,_,_> with
            member _.Default = { Timestamp = struct(DateTime.MinValue, ""); Value = ValueNone }
            member _.Query crdt = crdt.Value
            member _.Prepare(_, value) = (DateTime.UtcNow, value)
            member _.Effect(existing, e) =
                let (at, value) = e.Data
                let timestamp = struct(at, e.Origin)
                if existing.Timestamp < timestamp then
                    { existing with Timestamp = timestamp; Value = value }
                else existing }
    
    let props db replica ctx = replicator crdt db replica ctx
    let updte (value: 'a voption) (ref: Endpoint<'a>) : Async<'a voption> = ref <? Command value
    let query (ref: Endpoint<'a>) : Async<'a voption> = ref <? Query

[<RequireQualifiedAccess>]    
module MVRegister =
        
    type MVRegister<'a> = (VTime * 'a voption) list
    
    type Endpoint<'a> = Endpoint<MVRegister<'a>, 'a voption, 'a voption>
    
    let private crdt : Crdt<MVRegister<'a>, 'a list, 'a voption, 'a voption> =
        { new Crdt<_,_,_,_> with
            member _.Default = []
            member _.Query crdt =
                crdt
                |> List.choose (function (_, ValueSome v) -> Some v | _ -> None)
            member _.Prepare(_, value) = value
            member _.Effect(existing, e) =
                let concurrent =
                    existing
                    |> List.filter (fun (vt, _) -> Version.compare vt e.Version = Ord.Cc)
                (e.Version, e.Data)::concurrent }
    
    let props db replica ctx = replicator crdt db replica ctx
    let updte (value: 'a voption) (ref: Endpoint<'a>) : Async<'a voption> = ref <? Command value
    let query (ref: Endpoint<'a>) : Async<'a voption> = ref <? Query
    
[<RequireQualifiedAccess>]
module Rga =
    
    /// Virtual index - while physical index of an element in RGA changes as new elements are appended or removed,
    /// a virtual index always stays the same. It allows tracking the item position over time.
    type VPtr = (int * ReplicaId)
        
    type Vertex<'a> = (VPtr * 'a option)
    type Rga<'a> =
        { Sequencer: VPtr
          Vertices: Vertex<'a>[] }
        
    type Command<'a> =
        | Insert of index:int * value:'a
        | RemoveAt of index:int
        
    type Operation<'a> =
        | Inserted of after:VPtr * at:VPtr * value:'a
        | Removed of at:VPtr
        
    /// Checks if given vertex has been tombstoned.
    let inline isTombstone (_, data) = Option.isNone data
        
    /// Maps user-given index (which ignores tombstones) into physical index inside of `vertices` array.
    let private indexWithTombstones index vertices =
        let rec loop offset remaining (vertices: Vertex<'a>[]) =
            if remaining = 0 then offset
            elif isTombstone vertices.[offset] then loop (offset+1) remaining vertices // skip over tombstones
            else loop (offset+1) (remaining-1) vertices
        loop 1 index vertices // skip head as it's always tombstoned (it serves as reference point)
    
    /// Maps user-given VIndex into physical index inside of `vertices` array.
    let private indexOfVPtr ptr vertices =
        let rec loop offset ptr (vertices: Vertex<'a>[]) =
            if ptr = fst vertices.[offset] then offset
            else loop (offset+1) ptr vertices
        loop 0 ptr vertices
        
    /// Recursively checks if the next vertex on the right of a given `offset`
    /// has position higher than `pos` at if so, shift offset to the right.
    /// 
    /// By design, when doing concurrent inserts, we skip over elements on the right
    /// if their Position is higher than Position of inserted element. 
    let rec private shift offset ptr (vertices: Vertex<'a>[]) =
        if offset >= vertices.Length then offset // append at the end
        else
            let (next, _) = vertices.[offset]
            if next < ptr then offset
            else shift (offset+1) ptr vertices // move insertion point to the right
        
    /// Increments given sequence number.
    let inline private nextSeqNr ((i, id): VPtr) : VPtr = (i+1, id)
    
    let private createInserted i value rga = 
      let index = indexWithTombstones i rga.Vertices // start from 1 to skip header vertex
      let prev = fst rga.Vertices.[index-1] // get VPtr of previous element or RGA's head
      let at = nextSeqNr rga.Sequencer
      Inserted(prev, at, value)
      
    let private createRemoved i rga =
      let index = indexWithTombstones i rga.Vertices // start from 1 to skip header vertex
      let at = fst rga.Vertices.[index] // get VPtr of a previous element
      Removed at
    
    let private applyInserted (predecessor: VPtr) (ptr: VPtr) value rga =
      // find index where predecessor vertex can be found
      let predecessorIdx = indexOfVPtr predecessor rga.Vertices
      // adjust index where new vertex is to be inserted
      let insertIdx = shift (predecessorIdx+1) ptr rga.Vertices
      // update RGA to store the highest observed sequence number
      let (seqNr, replicaId) = rga.Sequencer
      let nextSeqNr = (max (fst ptr) seqNr, replicaId)
      let newVertices = Array.insert insertIdx (ptr, Some value) rga.Vertices
      { Sequencer = nextSeqNr; Vertices = newVertices }
      
    let private applyRemoved ptr rga =
      // find index where removed vertex can be found and clear its content to tombstone it
      let index = indexOfVPtr ptr rga.Vertices
      let (at, _) = rga.Vertices.[index]
      { rga with Vertices = Array.replace index (at, None) rga.Vertices }
        
    let private crdt (replicaId: ReplicaId) : Crdt<Rga<'a>, 'a[], Command<'a>, Operation<'a>> =
        { new Crdt<_,_,_,_> with
            member _.Default = { Sequencer = (0,replicaId); Vertices = [| ((0,""), None) |] }            
            member _.Query rga = rga.Vertices |> Array.choose snd            
            member _.Prepare(rga, cmd) =
                match cmd with
                | Insert(i, value) -> createInserted i value rga
                | RemoveAt(i) -> createRemoved i rga
            member _.Effect(rga, e) =
                match e.Data with
                | Inserted(predecessor, ptr, value) -> applyInserted predecessor ptr value rga
                | Removed(ptr) -> applyRemoved ptr rga
        }
        
    type Endpoint<'a> = Endpoint<Rga<'a>, Command<'a>, Operation<'a>>
        
    /// Used to create replication endpoint handling operation-based RGA protocol.
    let props db replicaId ctx = replicator (crdt replicaId) db replicaId ctx
     
    /// Inserts an `item` at given index. To insert at head use 0 index,
    /// to push back to a tail of sequence insert at array length. 
    let insert (index: int) (item: 'a) (ref: Endpoint<'a>) : Async<'a[]> = ref <? Command (Insert(index, item))
    
    /// Removes item stored at a provided `index`.
    let removeAt (index: int) (ref: Endpoint<'a>) : Async<'a[]> = ref <? Command (RemoveAt index)
    
    /// Retrieve an array of elements maintained by the given `ref` endpoint. 
    let query (ref: Endpoint<'a>) : Async<'a[]> = ref <? Query

/// Block-wise RGA. It exposes operations for adding/removing multiple elements at once.
[<RequireQualifiedAccess>]
module BWRga =
    
    type Position = (int * ReplicaId)
    [<Struct>]
    type PositionOffset =
        { Position: Position; Offset: int }
        override this.ToString () = sprintf "(%i%s:%i)" (fst this.Position) (snd this.Position) this.Offset
    
    [<Struct>]
    type Content<'a> =
        | Content of content:'a[]
        | Tombstone of skipped:int
        member this.Slice(offset: int) =
            match this with
            | Content data ->
                let left = Array.take offset data
                let right = Array.skip offset data
                (Content left, Content right)
            | Tombstone length ->
                (Tombstone offset, Tombstone (length - offset))
        
    type Block<'a> =
        { Ptr: PositionOffset
          //TODO: in this approach Block contains both user data and CRDT metadata, it's possible
          // however to split these appart and all slicing manipulations can be performed on blocks
          // alone. In this case Query method could return an user data right away with no extra
          // modifications, while the user-content could be stored in optimized structure such as Rope,
          // instead of deeply cloned arrays used here.
          Data: Content<'a> }
        member this.Length =
            match this.Data with
            | Content data -> data.Length
            | Tombstone skipped -> skipped
            
        override this.ToString() =
            sprintf "%O -> %A" this.Ptr this.Data
            
    [<RequireQualifiedAccess>]
    module Block =
        let tombstone (block: Block<'a>) = { block with Data = Tombstone block.Length }
        
        let isTombstone (block: Block<'a>) = match block.Data with Tombstone _ -> true | _ -> false
        
        let split (offset) (block: Block<'a>) =
            if offset = block.Length then (block, None)
            else
                let ptr = block.Ptr
                let (a, b) = block.Data.Slice offset
                let left = { block with Data = a }
                let right = { Ptr = { ptr with Offset = ptr.Offset + offset }; Data = b }
                (left, Some right)
        
    type Rga<'a> =
        { Sequencer: Position
          Blocks: Block<'a>[] }
        
    type Command<'a> =
        | Insert of index:int * 'a[]
        | RemoveAt of index:int * count:int
        
    type Operation<'a> =
        | Inserted of after:PositionOffset * at:Position * value:'a[]
        | Removed of slices:(PositionOffset*int) list
                
    /// Given user-aware index, return an index of a block and position inside of that block,
    /// which matches provided index.
    let private findByIndex idx blocks =
        let rec loop currentIndex consumed (idx: int) (blocks: Block<'a>[]) =
            if idx = consumed then (currentIndex, 0)
            else
                let block = blocks.[currentIndex]
                if Block.isTombstone block then
                    loop (currentIndex+1) consumed idx blocks
                else
                    let remaining = idx - consumed
                    if remaining <= block.Length then
                        // we found the position somewhere in the block
                        (currentIndex, remaining)
                    else
                        // move to the next block with i shortened by current block length
                        loop (currentIndex + 1) (consumed + block.Length) idx blocks
        loop 0 0 idx blocks
                
    let private findByPositionOffset ptr blocks =
        let rec loop idx ptr (blocks: Block<'a>[]) =
            let block = blocks.[idx]
            if block.Ptr.Position = ptr.Position then 
                if block.Ptr.Offset + block.Length >= ptr.Offset  then (idx, ptr.Offset-block.Ptr.Offset)
                else loop (idx+1) ptr blocks
            else loop (idx+1) ptr blocks
        loop 0 ptr blocks
                
    /// Recursively check if the next vertex on the right of a given `offset`
    /// has position higher than `pos` at if so, shift offset to the right.  
    let rec private shift offset pos (blocks: Block<'a>[]) =
        if offset >= blocks.Length then offset // append at the tail
        else
            let next = blocks.[offset].Ptr.Position
            if next < pos then offset
            else shift (offset+1) pos blocks // move insertion point to the right
            
    /// Increments given sequence number.
    let inline private nextSeqNr ((i, id): Position) : Position = (i+1, id)
                
    let private sliceBlocks start count blocks =
        let rec loop acc idx offset remaining (blocks: Block<'a>[]) =
            let block = blocks.[idx]
            let ptr = block.Ptr
            let ptr = { ptr with Offset = ptr.Offset + offset }
            let len = block.Length - offset
            if len > remaining then (ptr, remaining)::acc
            elif len = 0 then loop acc (idx+1) 0 remaining blocks // skip over empty blocks
            else loop ((ptr, len)::acc) (idx+1) 0 (remaining-len) blocks
        let (first, offset) = findByIndex start blocks
        loop [] first offset count blocks |> List.rev
        
    let private filterBlocks slices blocks =
        let rec loop (acc: ResizeArray<Block<'a>>) idx slices (blocks: Block<'a>[]) =
            match slices with
            | [] ->
                for i=idx to blocks.Length-1 do
                    acc.Add blocks.[i] // copy over remaining blocks
                acc.ToArray()
            | (ptr, length)::tail ->
                let block = blocks.[idx]
                if block.Ptr.Position = ptr.Position then // we found valid block
                    let currLen = block.Length
                    if block.Ptr.Offset = ptr.Offset then // the beginning of deleted block was found
                        if currLen = length then // deleted block exactly matches bounds
                            acc.Add (Block.tombstone block)
                            loop acc (idx+1) tail blocks
                        elif currLen < length then // deleted block is longer, delete current one and keep remainder
                            acc.Add (Block.tombstone block)
                            let ptr = { ptr with Offset = ptr.Offset + currLen }
                            loop acc (idx+1) ((ptr, length-currLen)::tail) blocks
                        else // deleted block is shorter, we need to split current block and tombstone left side
                            let (left, Some right) = Block.split length block
                            acc.Add (Block.tombstone left)
                            acc.Add right
                            loop acc (idx+1) tail blocks
                    elif block.Ptr.Offset < ptr.Offset && block.Ptr.Offset + currLen > ptr.Offset then // the deleted block starts inside of a current one
                        let splitPoint = ptr.Offset - block.Ptr.Offset
                        let (left, Some right) = Block.split splitPoint block
                        acc.Add left
                        if length > right.Length then // remainer is longer than right, we need to subtract it and keep around
                            let remainer = length - right.Length
                            acc.Add (Block.tombstone right)
                            let pos = { ptr with Offset = right.Ptr.Offset + right.Length }
                            loop acc (idx+1) ((pos, remainer)::tail) blocks
                        else 
                            let (del, right) = Block.split length right
                            acc.Add (Block.tombstone del)
                            right |> Option.iter acc.Add
                            loop acc (idx+1) tail blocks
                    else    // position ID is correct but offset doesn't fit, we need to move on
                        acc.Add block
                        loop acc (idx+1) slices blocks
                else
                    acc.Add block
                    loop acc (idx+1) slices blocks
        loop (ResizeArray()) 1 slices blocks

    let private crdt (replicaId: ReplicaId) : Crdt<Rga<'a>, 'a[], Command<'a>, Operation<'a>> =
        { new Crdt<_,_,_,_> with
            member _.Default =
                let head = { Ptr = { Position = (0,""); Offset = 0 }; Data = Tombstone 0 }
                { Sequencer = (0,replicaId); Blocks = [| head |] }            
            member _.Query rga = rga.Blocks |> Array.collect (fun block -> match block.Data with Content data -> data | _ -> [||])        
            member _.Prepare(rga, cmd) =
                match cmd with
                | Insert(idx, slice) ->
                    let (index, offset) = findByIndex idx rga.Blocks
                    let ptr = rga.Blocks.[index].Ptr 
                    let at = nextSeqNr rga.Sequencer
                    Inserted({ ptr with Offset = ptr.Offset+offset }, at, slice)
                | RemoveAt(idx, count) -> 
                    let slices = sliceBlocks idx count rga.Blocks
                    Removed slices
                    
            member _.Effect(rga, e) =
                match e.Data with
                | Inserted(after, at, slice) ->
                    let (index, split) = findByPositionOffset after rga.Blocks
                    let indexAdjusted = shift (index+1) at rga.Blocks
                    let block = rga.Blocks.[index]
                    let newBlock = { Ptr = { Position = at; Offset = 0}; Data = Content slice }
                    let (left, right) = Block.split split block
                    let (seqNr, replicaId) = rga.Sequencer
                    let nextSeqNr = (max (fst at) seqNr, replicaId) 
                    let blocks =
                        rga.Blocks
                        |> Array.replace index left 
                        |> Array.insert indexAdjusted newBlock
                    match right with
                    | Some right ->
                        let blocks = blocks |> Array.insert (indexAdjusted+1) right
                        { Sequencer = nextSeqNr; Blocks = blocks }
                    | None ->
                        { Sequencer = nextSeqNr; Blocks = blocks }
                | Removed(slices) ->
                    let blocks = filterBlocks slices rga.Blocks                    
                    { rga with Blocks = blocks }
        }
        
    type Endpoint<'a> = Endpoint<Rga<'a>, Command<'a>, Operation<'a>>
        
    /// Used to create replication endpoint handling operation-based RGA protocol.
    let props db replicaId ctx = replicator (crdt replicaId) db replicaId ctx
     
    let insertRange (index: int) (slice: 'a[]) (ref: Endpoint<'a>) : Async<'a[]> = ref <? Command (Insert(index, slice))
    
    let removeRange (index: int) (count: int) (ref: Endpoint<'a>) : Async<'a[]> = ref <? Command (RemoveAt(index, count))
    
    /// Retrieve the current state of the RGA maintained by the given `ref` endpoint. 
    let query (ref: Endpoint<'a>) : Async<'a[]> = ref <? Query 
    
[<RequireQualifiedAccess>]
module LSeq =
    
    [<Struct;CustomComparison;CustomEquality>]
    type VPtr =
        { Sequence: byte[]; Id: ReplicaId }
        override this.ToString() =
            String.Join('.', this.Sequence) + ":" + string this.Id 
        member this.CompareTo(other) =
            // apply lexical comparison of sequence elements
            let len = min this.Sequence.Length other.Sequence.Length
            let mutable i = 0
            let mutable cmp = 0
            while cmp = 0 && i < len do
                cmp <- this.Sequence.[i].CompareTo other.Sequence.[i]
                i <- i + 1
            if cmp = 0 then
                // one of the sequences is subsequence of another one,
                // compare their lengths (cause maybe they're the same)
                // then compare replica ids
                cmp <- this.Sequence.Length - other.Sequence.Length
                if cmp = 0 then this.Id.CompareTo other.Id else cmp
            else cmp
        interface IComparable<VPtr> with member this.CompareTo other = this.CompareTo other
        interface IComparable with member this.CompareTo other = match other with :? VPtr as vptr -> this.CompareTo(vptr)
        interface IEquatable<VPtr> with member this.Equals other = this.CompareTo other = 0
    
    type Vertex<'a> = (VPtr * 'a)
    type LSeq<'a> = Vertex<'a>[]
            
    type Command<'a> =
        | Insert of index:int * value:'a
        | RemoveAt of index:int
        
    type Operation<'a> =
        | Inserted of at:VPtr * value:'a
        | Removed of at:VPtr
        
    /// Binary search for index of `vptr` in an ordered sequence, looking for a place to insert
    /// an element. If `vptr` is the lowest element, 0 will be returned. If it's the highest
    /// one: lseq.Length will be returned.
    let private binarySearch vptr (lseq: LSeq<_>) =
        let mutable i = 0
        let mutable j = lseq.Length
        while i < j do
            let half = (i + j) / 2
            if vptr >= fst lseq.[half] then i <- half + 1
            else j <- half
        i
    
    /// Generates a byte sequence that - ordered lexically - would fit between `lo` and `hi`.
    let private generateSeq (lo: byte[]) (hi: byte[]) =
        let rec loop (acc: ResizeArray<byte>) i (lo: byte[]) (hi: byte[])  =
            let min = if i >= lo.Length then 0uy else lo.[i]
            let max = if i >= hi.Length then 255uy else hi.[i]
            if min + 1uy < max then
                acc.Add (min + 1uy)
                acc.ToArray()
            else
                acc.Add min
                loop acc (i+1) lo hi
        loop (ResizeArray (min lo.Length hi.Length)) 0 lo hi
    
    let private crdt (replicaId: ReplicaId) : Crdt<LSeq<'a>, 'a[], Command<'a>, Operation<'a>> =
        { new Crdt<_,_,_,_> with
            member _.Default = [||]     
            member _.Query lseq = lseq |> Array.map snd
            member _.Prepare(lseq, cmd) =
                match cmd with
                | Insert(i, value) ->
                    let left = if Array.isEmpty lseq || i = 0 then [||] else (fst lseq.[i-1]).Sequence  
                    let right = if i = lseq.Length then [||] else (fst lseq.[i]).Sequence
                    let ptr = { Sequence = generateSeq left right; Id = replicaId }
                    Inserted(ptr, value)
                | RemoveAt(i) -> Removed(fst lseq.[i])
            member _.Effect(lseq, e) =
                match e.Data with
                | Inserted(ptr, value) ->
                    let idx = binarySearch ptr lseq
                    Array.insert idx (ptr, value) lseq
                | Removed(ptr) -> 
                    let idx = binarySearch ptr lseq
                    Array.removeAt idx lseq
        }
        
    type Endpoint<'a> = Endpoint<LSeq<'a>, Command<'a>, Operation<'a>>
        
    /// Used to create replication endpoint handling operation-based RGA protocol.
    let props db replicaId ctx = replicator (crdt replicaId) db replicaId ctx
     
    /// Inserts an `item` at given index. To insert at head use 0 index,
    /// to push back to a tail of sequence insert at array length. 
    let insert (index: int) (item: 'a) (ref: Endpoint<'a>) : Async<'a[]> = ref <? Command (Insert(index, item))
    
    /// Removes item stored at a provided `index`.
    let removeAt (index: int) (ref: Endpoint<'a>) : Async<'a[]> = ref <? Command (RemoveAt index)
    
    /// Retrieve an array of elements maintained by the given `ref` endpoint. 
    let query (ref: Endpoint<'a>) : Async<'a[]> = ref <? Query