Event Store algo

es2.cfg

\* Add statements after this line.
CONSTANTS
    NIL = NIL
    MaxCrashes = 0
    NumProcs = 3
SPECIFICATION
Spec
\* INVARIANT definition
INVARIANT
Invariant

es2.tla

---- MODULE es2 ----
EXTENDS Naturals, TLC, Sequences, Integers, FiniteSets

CONSTANTS NIL, NumProcs, MaxCrashes

\* Return a set of sequences representing all permutations of a set S.
SeqPermutations(S) == 
   {f \in [1..Cardinality(S) -> S] : \A w \in S : \E v \in DOMAIN f : f[v]=w}   

(* --fair algorithm eventstore

variables

\*   All possible permutations of events.
\*  events \in SeqPermutations({
\*    [id |-> "id1", nodes |-> {0}],
\*    [id |-> "id2", nodes |-> {3, 4}],
\*    [id |-> "id3", nodes |-> {0, 1}],
\*    [id |-> "id4", nodes |-> {-1, 1}],
\*    [id |-> "id6", nodes |-> {-1, 0, 1}],
\*    [id |-> "id5", nodes |-> {1, 3}],
\*    [id |-> "id7", nodes |-> {3, 4}]
\*  }),

 events = <<
     [id |-> "id1", nodes |-> {0}],
   [id |-> "id2", nodes |-> {3, 4}],
   [id |-> "id3", nodes |-> {0, 1}],
   [id |-> "id4", nodes |-> {-1, 1}],
   [id |-> "id6", nodes |-> {-1, 0, 1}],
   [id |-> "id5", nodes |-> {1, 3}],
   [id |-> "id7", nodes |-> {3, 4}]
   >>,

\* Copy events to allEvents as events is mutable.
allEvents = events,
 
 \* Function representing a database. See macro `insert` for the record structure.
db = [x \in {} |-> {}];
 
define
    
    \* For a given set of nodes, return all nodes that are connected by parent node.
    ConnectedDBRecords(nodes) ==
        LET
            RECURSIVE Connected(_, _)
            Connected(node, seen) ==
                IF node \in DOMAIN db /\ node \notin seen
                THEN {db[node]} \union Connected(db[node].parent, seen \union {node})
                ELSE {}
        IN
            UNION {Connected(node, {}): node \in nodes}
    
    Min(S) == CHOOSE s \in S : \A t \in S : s <= t

    \* Root is either a min node of existing root nodes, or a min node of the event. 
    DefineRoot(event, records) ==
        IF records = {}
        THEN Min(event.nodes)
        ELSE Min({r.node: r \in {r \in records: r.parent = NIL}})

    \* Returns nodes that used to be roots.
    StaleRoots(node) ==
        LET
            RECURSIVE PrevRootsOfInternal(_, _)
            PrevRootsOfInternal(mfp, seen) ==
                IF db[mfp].staleRoots = {} \/ mfp \in seen
                THEN {}
                ELSE
                    db[mfp].staleRoots
                    \union
                    (UNION {PrevRootsOfInternal(p, seen \cup {mfp}): p \in db[mfp].staleRoots})
        IN
            PrevRootsOfInternal(node, {})

    \* Returns all events that are (transitively) associated with the given node.
     EventsOf(node) == UNION {db[r].events: r \in {node} \union StaleRoots(node)}
end define;
 
macro insert(node, parent, wasRoot) begin
    db := db @@ (node :> [node |-> node, events |-> {}, parent |-> parent, staleRoots |-> {}, wasRoot |-> wasRoot ]);
end macro;
 
fair process EventsConsumer \in 1..NumProcs
variable
    connectedRecords = NIL,
    e = NIL,
    root = NIL,
    staleRoots = NIL,
    newNodes = NIL,
    currentNode = NIL,
    crashes = 0;
begin

START:
while events /= <<>> \/ e # NIL do
    if e = NIL then
        e := Head(events);
        events := Tail(events);
    end if;
    connectedRecords := ConnectedDBRecords(e.nodes);
    root := DefineRoot(e, connectedRecords);
    staleRoots := {r.node: r \in {r \in connectedRecords: r.parent = NIL /\ r.node # root}};
    newNodes := (e.nodes \ {r.node: r \in connectedRecords}) \ {root};

UPDATE_ROOT:
    if root \notin {r.node: r \in connectedRecords} then
        \* Insert root if not exists in DB, retry otherwise.
        if root \notin DOMAIN db \/ db[root].parent = NIL then
          insert(root, NIL, TRUE)
        else
          goto START;
        end if;
    end if;

ADD_STALE_ROOTS_TO_ROOT:
    db[root].staleRoots := db[root].staleRoots \union staleRoots;

LOOP_UPDATE_STALE_ROOT:
    while staleRoots # {} do
        currentNode := CHOOSE x \in staleRoots: TRUE;
        staleRoots := staleRoots \ {currentNode};
    UPDATE_STALE_ROOT:
        either
            \* Update stal root to point to the new one if it wasn't updated concurrenly,
            \* retry otherwise.
            if db[currentNode].parent = NIL \/ db[currentNode].parent = root then
                db[currentNode].parent := root;
            else
                goto START;
            end if;
        or
            \* Simulate system crash by retrying from START.
            await crashes < MaxCrashes;
            crashes := crashes + 1;
            goto START;
        end either;
    end while;

LOOP_INSERT_NEW_NODE:
    while newNodes # {} do
        currentNode := CHOOSE x \in newNodes: TRUE;
        newNodes := newNodes \ {currentNode};
    INSERT_NEW_NODE:
        either
            \* Insert record currentNode -> root if it doesn't exist,
            \* retry otherwise.
            if currentNode \notin DOMAIN db then
                insert(currentNode, root, FALSE)
            else
                goto START;
            end if;
        or
            \* Simulate system crash by retrying from START.
            await crashes < MaxCrashes;
            crashes := crashes + 1;
            goto START;
        end either;
    end while;

WRITE_EVENT:
    db[root].events := db[root].events \union {e.id};
    e := NIL;
end while;
end process;

end algorithm *)

\* BEGIN TRANSLATION - the hash of the PCal code: PCal-a1550d78bc094a78e0beee63bfbabaa6 (chksum(pcal) = "9810ecb8" /\ chksum(tla) = "442fdddc") (chksum(pcal) = "9810ecb8" /\ chksum(tla) = "442fdddc") (chksum(pcal) = "9810ecb8" /\ chksum(tla) = "442fdddc") (chksum(pcal) = "9810ecb8" /\ chksum(tla) = "442fdddc") (chksum(pcal) = "9810ecb8" /\ chksum(tla) = "442fdddc") (chksum(pcal) = "9810ecb8" /\ chksum(tla) = "442fdddc") (chksum(pcal) = "2f633010" /\ chksum(tla) = "d6a008") (chksum(pcal) = "2f633010" /\ chksum(tla) = "d6a008") (chksum(pcal) = "6f8b5c49" /\ chksum(tla) = "ec4533f") (chksum(pcal) = "6f8b5c49" /\ chksum(tla) = "ec4533f") (chksum(pcal) = "e1d69860" /\ chksum(tla) = "32d024ea") (chksum(pcal) = "e1d69860" /\ chksum(tla) = "32d024ea") (chksum(pcal) = "3ccd37f6" /\ chksum(tla) = "c48bd49c") (chksum(pcal) = "f0f2df09" /\ chksum(tla) = "969f9a91") (chksum(pcal) = "b834216f" /\ chksum(tla) = "4c23f733") (chksum(pcal) = "2d9599a9" /\ chksum(tla) = "82567012") (chksum(pcal) = "7f353d62" /\ chksum(tla) = "462e1df8") (chksum(pcal) = "12449f7d" /\ chksum(tla) = "78bf56c6") (chksum(pcal) = "284bc789" /\ chksum(tla) = "e8b99687") (chksum(pcal) = "284bc789" /\ chksum(tla) = "e8b99687") (chksum(pcal) = "284bc789" /\ chksum(tla) = "e8b99687") (chksum(pcal) = "9b6a744b" /\ chksum(tla) = "cec3c966") (chksum(pcal) = "4ede9aad" /\ chksum(tla) = "e80133d8") (chksum(pcal) = "dd672b9b" /\ chksum(tla) = "33684bab") (chksum(pcal) = "df6a57a2" /\ chksum(tla) = "c3bf3ded") (chksum(pcal) = "88040025" /\ chksum(tla) = "e2837d1d") (chksum(pcal) = "d793d8c0" /\ chksum(tla) = "f9773926") (chksum(pcal) = "d793d8c0" /\ chksum(tla) = "bd7e9abc") (chksum(pcal) = "3acb8b" /\ chksum(tla) = "791d3866") (chksum(pcal) = "586882e6" /\ chksum(tla) = "76e570e8") (chksum(pcal) = "586882e6" /\ chksum(tla) = "76e570e8") (chksum(pcal) = "586882e6" /\ chksum(tla) = "76e570e8") (chksum(pcal) = "586882e6" /\ chksum(tla) = "76e570e8") (chksum(pcal) = "586882e6" /\ chksum(tla) = "76e570e8") (chksum(pcal) = "586882e6" /\ chksum(tla) = "76e570e8") (chksum(pcal) = "586882e6" /\ chksum(tla) = "76e570e8") (chksum(pcal) = "a076b48c" /\ chksum(tla) = "44686d9a") (chksum(pcal) = "834db2e7" /\ chksum(tla) = "daa4c830") (chksum(pcal) = "d075ebbe" /\ chksum(tla) = "23aa35c6") (chksum(pcal) = "d075ebbe" /\ chksum(tla) = "23aa35c6") (chksum(pcal) = "d075ebbe" /\ chksum(tla) = "23aa35c6") (chksum(pcal) = "93280ef0" /\ chksum(tla) = "732cc3b9") (chksum(pcal) = "d075ebbe" /\ chksum(tla) = "171632dd") (chksum(pcal) = "faa23970" /\ chksum(tla) = "da89af3a") (chksum(pcal) = "735fdec7" /\ chksum(tla) = "35a3d5cd") (chksum(pcal) = "3b66ab46" /\ chksum(tla) = "2ae36eb") (chksum(pcal) = "735fdec7" /\ chksum(tla) = "75808d7") (chksum(pcal) = "3a039bb8" /\ chksum(tla) = "13989a11") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "4f940cac" /\ chksum(tla) = "7958653f") (chksum(pcal) = "3a039bb8" /\ chksum(tla) = "79c810c1") (chksum(pcal) = "d1074a44" /\ chksum(tla) = "f6ad153") (chksum(pcal) = "f07032a0" /\ chksum(tla) = "ef2b671") (chksum(pcal) = "feb13180" /\ chksum(tla) = "8be79904") (chksum(pcal) = "f387dd10" /\ chksum(tla) = "d61a3079") (chksum(pcal) = "f387dd10" /\ chksum(tla) = "c54d90ea") (chksum(pcal) = "f387dd10" /\ chksum(tla) = "f0b5715f") (chksum(pcal) = "cda0d1e1" /\ chksum(tla) = "3a8974eb") (chksum(pcal) = "6889986e" /\ chksum(tla) = "2a3b72cb") (chksum(pcal) = "f2c02283" /\ chksum(tla) = "801b48bb") (chksum(pcal) = "b2f95dda" /\ chksum(tla) = "b843d73b") (chksum(pcal) = "cc993116" /\ chksum(tla) = "df04790b") (chksum(pcal) = "e256c884" /\ chksum(tla) = "f6f93647") (chksum(pcal) = "11bf64aa" /\ chksum(tla) = "82fdda80") (chksum(pcal) = "11bf64aa" /\ chksum(tla) = "82fdda80") (chksum(pcal) = "11bf64aa" /\ chksum(tla) = "82fdda80") (chksum(pcal) = "11bf64aa" /\ chksum(tla) = "82fdda80") (chksum(pcal) = "11bf64aa" /\ chksum(tla) = "82fdda80") (chksum(pcal) = "11bf64aa" /\ chksum(tla) = "82fdda80") (chksum(pcal) = "11bf64aa" /\ chksum(tla) = "82fdda80") (chksum(pcal) = "3aa6a1e2" /\ chksum(tla) = "6cfa01c0") (chksum(pcal) = "e6a05079" /\ chksum(tla) = "52de0704") (chksum(pcal) = "73ac4ea1" /\ chksum(tla) = "a9fe76fc") (chksum(pcal) = "6c5d1399" /\ chksum(tla) = "4b370ebc") (chksum(pcal) = "897c975d" /\ chksum(tla) = "1d9c2cbc") (chksum(pcal) = "897c975d" /\ chksum(tla) = "1d9c2cbc") (chksum(pcal) = "897c975d" /\ chksum(tla) = "1d9c2cbc")
VARIABLES events, allEvents, db, pc

(* define statement *)
ConnectedDBRecords(nodes) ==
    LET
        RECURSIVE Connected(_, _)
        Connected(node, seen) ==
            IF node \in DOMAIN db /\ node \notin seen
            THEN {db[node]} \union Connected(db[node].parent, seen \union {node})
            ELSE {}
    IN
        UNION {Connected(node, {}): node \in nodes}

Min(S) == CHOOSE s \in S : \A t \in S : s <= t


DefineRoot(event, records) ==
    IF records = {}
    THEN Min(event.nodes)
    ELSE Min({r.node: r \in {r \in records: r.parent = NIL}})


StaleRoots(node) ==
    LET
        RECURSIVE PrevRootsOfInternal(_, _)
        PrevRootsOfInternal(mfp, seen) ==
            IF db[mfp].staleRoots = {} \/ mfp \in seen
            THEN {}
            ELSE
                db[mfp].staleRoots
                \union
                (UNION {PrevRootsOfInternal(p, seen \cup {mfp}): p \in db[mfp].staleRoots})
    IN
        PrevRootsOfInternal(node, {})


 EventsOf(node) == UNION {db[r].events: r \in {node} \union StaleRoots(node)}

VARIABLES connectedRecords, e, root, staleRoots, newNodes, currentNode, 
          crashes

vars == << events, allEvents, db, pc, connectedRecords, e, root, staleRoots, 
           newNodes, currentNode, crashes >>

ProcSet == (1..NumProcs)

Init == (* Global variables *)
        /\ events =        <<
                      [id |-> "id1", nodes |-> {0}],
                    [id |-> "id2", nodes |-> {3, 4}],
                    [id |-> "id3", nodes |-> {0, 1}],
                    [id |-> "id4", nodes |-> {-1, 1}],
                    [id |-> "id6", nodes |-> {-1, 0, 1}],
                    [id |-> "id5", nodes |-> {1, 3}],
                    [id |-> "id7", nodes |-> {3, 4}]
                    >>
        /\ allEvents = events
        /\ db = [x \in {} |-> {}]
        (* Process EventsConsumer *)
        /\ connectedRecords = [self \in 1..NumProcs |-> NIL]
        /\ e = [self \in 1..NumProcs |-> NIL]
        /\ root = [self \in 1..NumProcs |-> NIL]
        /\ staleRoots = [self \in 1..NumProcs |-> NIL]
        /\ newNodes = [self \in 1..NumProcs |-> NIL]
        /\ currentNode = [self \in 1..NumProcs |-> NIL]
        /\ crashes = [self \in 1..NumProcs |-> 0]
        /\ pc = [self \in ProcSet |-> "START"]

START(self) == /\ pc[self] = "START"
               /\ IF events /= <<>> \/ e[self] # NIL
                     THEN /\ IF e[self] = NIL
                                THEN /\ e' = [e EXCEPT ![self] = Head(events)]
                                     /\ events' = Tail(events)
                                ELSE /\ TRUE
                                     /\ UNCHANGED << events, e >>
                          /\ connectedRecords' = [connectedRecords EXCEPT ![self] = ConnectedDBRecords(e'[self].nodes)]
                          /\ root' = [root EXCEPT ![self] = DefineRoot(e'[self], connectedRecords'[self])]
                          /\ staleRoots' = [staleRoots EXCEPT ![self] = {r.node: r \in {r \in connectedRecords'[self]: r.parent = NIL /\ r.node # root'[self]}}]
                          /\ newNodes' = [newNodes EXCEPT ![self] = (e'[self].nodes \ {r.node: r \in connectedRecords'[self]}) \ {root'[self]}]
                          /\ pc' = [pc EXCEPT ![self] = "UPDATE_ROOT"]
                     ELSE /\ pc' = [pc EXCEPT ![self] = "Done"]
                          /\ UNCHANGED << events, connectedRecords, e, root, 
                                          staleRoots, newNodes >>
               /\ UNCHANGED << allEvents, db, currentNode, crashes >>

UPDATE_ROOT(self) == /\ pc[self] = "UPDATE_ROOT"
                     /\ IF root[self] \notin {r.node: r \in connectedRecords[self]}
                           THEN /\ IF root[self] \notin DOMAIN db \/ db[root[self]].parent = NIL
                                      THEN /\ db' = db @@ (root[self] :> [node |-> root[self], events |-> {}, parent |-> NIL, staleRoots |-> {}, wasRoot |-> TRUE ])
                                           /\ pc' = [pc EXCEPT ![self] = "ADD_STALE_ROOTS_TO_ROOT"]
                                      ELSE /\ pc' = [pc EXCEPT ![self] = "START"]
                                           /\ db' = db
                           ELSE /\ pc' = [pc EXCEPT ![self] = "ADD_STALE_ROOTS_TO_ROOT"]
                                /\ db' = db
                     /\ UNCHANGED << events, allEvents, connectedRecords, e, 
                                     root, staleRoots, newNodes, currentNode, 
                                     crashes >>

ADD_STALE_ROOTS_TO_ROOT(self) == /\ pc[self] = "ADD_STALE_ROOTS_TO_ROOT"
                                 /\ db' = [db EXCEPT ![root[self]].staleRoots = db[root[self]].staleRoots \union staleRoots[self]]
                                 /\ pc' = [pc EXCEPT ![self] = "LOOP_UPDATE_STALE_ROOT"]
                                 /\ UNCHANGED << events, allEvents, 
                                                 connectedRecords, e, root, 
                                                 staleRoots, newNodes, 
                                                 currentNode, crashes >>

LOOP_UPDATE_STALE_ROOT(self) == /\ pc[self] = "LOOP_UPDATE_STALE_ROOT"
                                /\ IF staleRoots[self] # {}
                                      THEN /\ currentNode' = [currentNode EXCEPT ![self] = CHOOSE x \in staleRoots[self]: TRUE]
                                           /\ staleRoots' = [staleRoots EXCEPT ![self] = staleRoots[self] \ {currentNode'[self]}]
                                           /\ pc' = [pc EXCEPT ![self] = "UPDATE_STALE_ROOT"]
                                      ELSE /\ pc' = [pc EXCEPT ![self] = "LOOP_INSERT_NEW_NODE"]
                                           /\ UNCHANGED << staleRoots, 
                                                           currentNode >>
                                /\ UNCHANGED << events, allEvents, db, 
                                                connectedRecords, e, root, 
                                                newNodes, crashes >>

UPDATE_STALE_ROOT(self) == /\ pc[self] = "UPDATE_STALE_ROOT"
                           /\ \/ /\ IF db[currentNode[self]].parent = NIL \/ db[currentNode[self]].parent = root[self]
                                       THEN /\ db' = [db EXCEPT ![currentNode[self]].parent = root[self]]
                                            /\ pc' = [pc EXCEPT ![self] = "LOOP_UPDATE_STALE_ROOT"]
                                       ELSE /\ pc' = [pc EXCEPT ![self] = "START"]
                                            /\ db' = db
                                 /\ UNCHANGED crashes
                              \/ /\ crashes[self] < MaxCrashes
                                 /\ crashes' = [crashes EXCEPT ![self] = crashes[self] + 1]
                                 /\ pc' = [pc EXCEPT ![self] = "START"]
                                 /\ db' = db
                           /\ UNCHANGED << events, allEvents, connectedRecords, 
                                           e, root, staleRoots, newNodes, 
                                           currentNode >>

LOOP_INSERT_NEW_NODE(self) == /\ pc[self] = "LOOP_INSERT_NEW_NODE"
                              /\ IF newNodes[self] # {}
                                    THEN /\ currentNode' = [currentNode EXCEPT ![self] = CHOOSE x \in newNodes[self]: TRUE]
                                         /\ newNodes' = [newNodes EXCEPT ![self] = newNodes[self] \ {currentNode'[self]}]
                                         /\ pc' = [pc EXCEPT ![self] = "INSERT_NEW_NODE"]
                                    ELSE /\ pc' = [pc EXCEPT ![self] = "WRITE_EVENT"]
                                         /\ UNCHANGED << newNodes, currentNode >>
                              /\ UNCHANGED << events, allEvents, db, 
                                              connectedRecords, e, root, 
                                              staleRoots, crashes >>

INSERT_NEW_NODE(self) == /\ pc[self] = "INSERT_NEW_NODE"
                         /\ \/ /\ IF currentNode[self] \notin DOMAIN db
                                     THEN /\ db' = db @@ (currentNode[self] :> [node |-> currentNode[self], events |-> {}, parent |-> root[self], staleRoots |-> {}, wasRoot |-> FALSE ])
                                          /\ pc' = [pc EXCEPT ![self] = "LOOP_INSERT_NEW_NODE"]
                                     ELSE /\ pc' = [pc EXCEPT ![self] = "START"]
                                          /\ db' = db
                               /\ UNCHANGED crashes
                            \/ /\ crashes[self] < MaxCrashes
                               /\ crashes' = [crashes EXCEPT ![self] = crashes[self] + 1]
                               /\ pc' = [pc EXCEPT ![self] = "START"]
                               /\ db' = db
                         /\ UNCHANGED << events, allEvents, connectedRecords, 
                                         e, root, staleRoots, newNodes, 
                                         currentNode >>

WRITE_EVENT(self) == /\ pc[self] = "WRITE_EVENT"
                     /\ db' = [db EXCEPT ![root[self]].events = db[root[self]].events \union {e[self].id}]
                     /\ e' = [e EXCEPT ![self] = NIL]
                     /\ pc' = [pc EXCEPT ![self] = "START"]
                     /\ UNCHANGED << events, allEvents, connectedRecords, root, 
                                     staleRoots, newNodes, currentNode, 
                                     crashes >>

EventsConsumer(self) == START(self) \/ UPDATE_ROOT(self)
                           \/ ADD_STALE_ROOTS_TO_ROOT(self)
                           \/ LOOP_UPDATE_STALE_ROOT(self)
                           \/ UPDATE_STALE_ROOT(self)
                           \/ LOOP_INSERT_NEW_NODE(self)
                           \/ INSERT_NEW_NODE(self) \/ WRITE_EVENT(self)

(* Allow infinite stuttering to prevent deadlock on termination. *)
Terminating == /\ \A self \in ProcSet: pc[self] = "Done"
               /\ UNCHANGED vars

Next == (\E self \in 1..NumProcs: EventsConsumer(self))
           \/ Terminating

Spec == /\ Init /\ [][Next]_vars
        /\ WF_vars(Next)
        /\ \A self \in 1..NumProcs : WF_vars(EventsConsumer(self))

Termination == <>(\A self \in ProcSet: pc[self] = "Done")

\* END TRANSLATION - the hash of the generated TLA code (remove to silence divergence warnings): TLA-cbc3c735ad20c68f2f54a0f9035e67ef


\* Invariant checks that:
\* - there is exactly one root (representative)
\* - stale roots (representatives) are correctly linked
\* - all events are reachable from the root
Invariant == 
    (\A self \in ProcSet: pc[self] = "Done") =>
    LET
         foundRoots == {r \in DOMAIN db: db[r].parent = NIL}
         foundRoot == CHOOSE r \in foundRoots: TRUE
    IN
        /\ Cardinality(foundRoots) = 1
        /\ StaleRoots(foundRoot) = {node \in DOMAIN db: db[node].wasRoot /\ node # foundRoot}
        /\ EventsOf(foundRoot) = {allEvents[i].id: i \in 1..Len(allEvents)}

====