cassandra

# package main
use github.com/synodlang/examples/cassandra
use github.com/synodlang/examples/gossip
use github.com/synodlang/examples/ring
use std/lesson
use std/sim

define main
  sim.Seed 3
  set nodes to (sim.Cluster of 6 running cassandra)
  each i, n in nodes
    if (i more than 0)
      gossip.Join n to (item 0 of nodes)
  set replicas to (ring.Owners of "user:42" among nodes count 3)
  set coordinator to (item 0 of replicas)
  set victim to (item 1 of replicas)
  after 5s
    deliver (message cassandra.Write with Key "user:42", Value "Ada") to coordinator
  after 6s
    sim.Crash victim
  after 12s
    deliver (message cassandra.Write with Key "user:42", Value "Grace") to coordinator
    lesson.Ask "One replica is down. Does the write succeed, and how will that replica catch up?" choosing {"It fails until the replica returns", "It succeeds; a hint replays it when the replica returns", "It succeeds, but the replica never gets it"}
  after 14s
    deliver (message cassandra.Read with Key "user:42") to (item 2 of replicas)
  after 18s
    sim.Restart victim
    lesson.Note "Data lives in memory here, so the restarted replica is refilled only by hints."
  sim.StopAt 30s

# package cassandra
use github.com/synodlang/examples/gossip
use github.com/synodlang/examples/ring
use std/lists
use std/maps
use std/net
use std/sim
use std/text
use std/time

Write has
  set Key to ""
  set Value to ""

Read has
  set Key to ""

Written has
  set Key to ""
  set Acks to 0

ReadDone has
  set Key to ""
  set Value to ""

Mutation has
  set key to ""
  set value to ""
  set stamp to 0s
  set request to ""

MutationAck has
  set request to ""

Fetch has
  set key to ""
  set request to ""

FetchReply has
  set request to ""
  set value to ""
  set stamp to 0s

Hint has
  set target to sim.Nobody
  set mutation to (message Mutation with key "")

Node has
  set Data to {}
  set PendingHints to 0
  set versions to {}
  set hints to {}
  set nextRequest to 0
  set acks to {}
  set keys to {}
  set replies to {}

define Factor
  result 3

define Quorum
  result 2

# Membership comes from gossip, never from a global list
define [n: Node] Members
  result (maps.Keys n.gossip.Known)

define [n: Node] ReplicasOf [k: text]
  result (ring.Owners of k among n.Members count Factor)

define [n: Node] NewRequest [k: text]
  set n.nextRequest to (n.nextRequest plus 1)
  set id to (text.From n.nextRequest)
  set item id of n.keys to k
  result id

define [n: Node] start
  time.Every 1s
    set waiting to {}
    each h in n.hints
      if (h.target is in n.gossip.Down)
        set waiting to (lists.Append h to waiting)
      else
        net.Send h.mutation from n to h.target
    set n.hints to waiting
    set n.PendingHints to (lists.Size n.hints)

# Coordinator: write to every replica; keep a hint for replicas that look down
when [n: Node] (sim.Delivered Write)
  set id to (n.NewRequest Write.Key)
  set item id of n.acks to 0
  set m to (message Mutation with key Write.Key, value Write.Value, stamp time.Now, request id)
  each r in (n.ReplicasOf Write.Key)
    if (r is in n.gossip.Down)
      set n.hints to (lists.Append (message Hint with target r, mutation m) to n.hints)
      set n.PendingHints to (lists.Size n.hints)
    else
      net.Send m from n to r

define [n: Node] Store [m: Mutation]
  set item m.key of n.Data to m.value
  set item m.key of n.versions to m.stamp

# Replica: keep the newest value per key (last write wins)
define [n: Node] Apply [m: Mutation]
  set k to m.key
  if (k is in n.versions)
    if (m.stamp more than (item k of n.versions))
      n.Store m
  else
    n.Store m

when [n: Node] (net.Received Mutation from coordinator)
  n.Apply Mutation
  net.Send (message MutationAck with request Mutation.request) from n to coordinator

# Coordinator: the write succeeds at QUORUM acknowledgements
when [n: Node] (net.Received MutationAck from replica)
  set id to MutationAck.request
  set item id of n.acks to ((item id of n.acks) plus 1)
  if ((item id of n.acks) is Quorum)
    emit (message Written with Key (item id of n.keys), Acks Quorum) by n

# Coordinator: read from the live replicas
when [n: Node] (sim.Delivered Read)
  set id to (n.NewRequest Read.Key)
  set item id of n.replies to {}
  each r in (n.ReplicasOf Read.Key)
    if (r is not in n.gossip.Down)
      net.Send (message Fetch with key Read.Key, request id) from n to r

when [n: Node] (net.Received Fetch from coordinator)
  set k to Fetch.key
  if (k is in n.Data)
    net.Send (message FetchReply with request Fetch.request, value (item k of n.Data), stamp (item k of n.versions)) from n to coordinator
  else
    net.Send (message FetchReply with request Fetch.request) from n to coordinator

# Coordinator: at QUORUM replies, answer with the newest value
when [n: Node] (net.Received FetchReply from replica)
  set id to FetchReply.request
  set item id of n.replies to (lists.Append FetchReply to (item id of n.replies))
  if ((lists.Size (item id of n.replies)) is Quorum)
    set best to (item 0 of (item id of n.replies))
    each r in (item id of n.replies)
      if (r.stamp more than best.stamp)
        set best to r
    emit (message ReadDone with Key (item id of n.keys), Value best.value) by n

# package gossip
use std/disk
use std/lists
use std/net
use std/random
use std/time

digest has
  set known to {}

Node has
  set Generation to 0
  set Heartbeat to 0
  set Known to {}
  set Down to {}
  set peers to {}
  set seenAt to {}

define Join [n: Node] to [seed: node]
  set n.peers to {seed}

# Versions combine generation and heartbeat, so a restarted node is believed again
define [n: Node] Version
  result ((n.Generation times 1000000) plus n.Heartbeat)

define [n: Node] Learn [k: node] at [v: number]
  set item k of n.Known to v
  set item k of n.seenAt to time.Now
  if ((k is not n) and (k is not in n.peers))
    set n.peers to (lists.Append k to n.peers)

define [n: Node] CheckPeers
  set suspects to {}
  each k, t in n.seenAt
    if ((time.Now minus t) more than 4s)
      set suspects to (lists.Append k to suspects)
  set n.Down to suspects

define [n: Node] start
  set n.Generation to ((disk.Read "generation" on n or 0) plus 1)
  disk.Write n.Generation at "generation" on n
  time.Every 1s
    set n.Heartbeat to (n.Heartbeat plus 1)
    n.Learn n at n.Version
    n.CheckPeers
    if ((lists.Size n.peers) more than 0)
      net.Send (message digest with known n.Known) from n to (random.Pick n.peers)

when [n: Node] (net.Received digest from sender)
  each k, v in digest.known
    if (k is in n.Known)
      if (v more than (item k of n.Known))
        n.Learn k at v
    else
      n.Learn k at v
  if (sender is not in n.peers)
    set n.peers to (lists.Append sender to n.peers)

# package ring
use std/lists
use std/maps
use std/math
use std/sim
use std/text

Placed has
  set Key to ""
  set Owner to sim.Nobody

define Vnodes
  result 8

# Each node owns several points on a ring of [0, 1), derived from its name
define TokensOf [n: node]
  set tokens to {}
  each i in (lists.Range 1 to Vnodes)
    set tokens to (lists.Append (math.Hash (text.Join {(sim.Name n), "#", (text.From i)})) to tokens)
  result tokens

# Clockwise distance from a key's point to the nearest point of a node
define Distance from [h: number] to [n: node]
  set best to 1
  each t in (TokensOf n)
    set d to (math.Remainder of (t minus h) by 1)
    if (d less than best)
      set best to d
  result best

# The first r nodes clockwise from the key own it
define Owners of [k: text] among [ns: list of node] count [r: number]
  set h to (math.Hash k)
  set distances to {}
  each n in ns
    set item n of distances to (Distance from h to n)
  result (lists.Take r from (maps.KeysByValue distances))