mermaidlive

README

Mermaid Diagram Live Update Demo

Running

go run ./cmd/mermaidlive

↓

http://localhost:8080/ui/

• to change the default countdown delay, provide the option, e.g. -delay 150ms

Embedded Resources

to only generate UI resources from ui-src, run:

go run ./cmd/mermaidlive -transpile

to build a binary with embedded UI:

go build --tags=embed ./cmd/mermaidlive

Ideas Sketched in the Spike

• developer experience of live reload back-end → ResourcesRefreshed → front-end
• UI served from a binary-embedded filesystem
• pub-sub long-polling connected clients and live viewers
• asynchronously running state machine observable via published events
• live re-rendering of a mermaid chart via events
• initial rendering of a chart via the LastSeenState event published upon connecting to the stream
• deployment on fly.io
• sharing Gherkin features between unit, API and Browser tests, and sharing step implementations between scenarios
• asynchronously connected client tests: two API and Browser clients
• long poll re-connects and showing the connection status to the user
• a persistent distributed G-Counter (grow-only counter) CRDT for started connections that is eventually-consistent, once service replicas see each other
  • service discovery via fly.io internal DNS polling
  • replication via a fully-connected ZeroMQ ineternal network mesh
  • a simple persistence of the CRDT counter in a continuously re-written JSON-structured file located on machine-bound fly.io volumes
  • not using a separately deployed database for the CRDT

Architecture

flowchart LR
    Server --serves---> UI
    StateMachine --runs on --> Server
    UI --posts commands to --> Server
    Server --forwards commands to -->StateMachine
    Server --publishes events to -->PubSub
    StateMachine --publishes events to -->PubSub
    Server --subscribes each connected client to -->PubSub
    Server --streams events to --> UI

Testing

• the specification contains shared steps
• state machine-level "unit" test steps
  • exececise the async state machine
• API-level test steps
  • start the server at port 8081
  • exercise the specification, including scenarios tagged with @api
• Browser-level: test steps

Unit

go test -v ./...

API-based

• the test starts a temporary server instance and runs the tests against it

./scripts/test-api.sh

Browser-based

• the test uses Playwright via the cucumber-playwright template
• prerequisites:
  • install dependencies: npm i
  • initialize Playwright: npx playwright install
• start the server first, e.g.:

./scripts/run-dev.sh -delay 150ms

• run the tests:

./scripts/test-ui.sh

Approach

• Mermaid API
• JSON Streaming
• Identifiable concurrent processes are modeled with phony (Go)
• Distributing shared state from the server to client connections via Pub/Sub

Deployment

• currently deployed on fly.io → mermaidlive.fly.dev
• docker compose

Experiments

Observing Replication in Docker

to start 3 initial replicas:

docker compose up -d

→ visit and reload the UI: http://localhost:8080/

stop 2 replicas:

docker stop mermaidlive-mermaidlive-2 mermaidlive-mermaidlive-3

↓

log:

mermaidlive-1 | Peers changed [172.19.0.4 172.19.0.5] -> []

reload the UI a couple of times and re-start the replicas:

docker start mermaidlive-mermaidlive-2 mermaidlive-mermaidlive-3

↓

log:

mermaidlive-1 | Peers changed [] -> [172.19.0.4 172.19.0.5]
mermaidlive-1 | Connecting to tcp://[172.19.0.4]:5000
mermaidlive-1 | Connecting to tcp://[172.19.0.5]:5000
mermaidlive-2 | New visitor count: 25
mermaidlive-3 | New visitor count: 25
mermaidlive-2 | Peers changed [172.19.0.2] -> [172.19.0.2 172.19.0.5]
mermaidlive-2 | Connecting to tcp://[172.19.0.5]:5000
mermaidlive-3 | Peers changed [172.19.0.2] -> [172.19.0.2 172.19.0.4]
mermaidlive-3 | Connecting to tcp://[172.19.0.4]:5000

observe, how the replicated visitor count is propagated to the replicas.

Scale replicas for further experiments, e.g.:

docker compose scale mermaidlive=5

copying the replica file to the local filesystem reveals the structure of the G-Counter:

docker compose cp mermaidlive:/appdata/my.gcounter .

↓

{
  "peers": {
    "3d226c43af66": 7,
    "c2206af9aba2": 9
  }
}

which results in the observable counter value of 16.

Documentation

Index

• Constants
• Variables
• func GetCounterDirectory() string
• func GetCounterIdentity() string
• func GetFS() http.FileSystem
• func Refresh()
• func StartWatching(eventPublisher *pubsub.PubSub[string, Event]) *fsnotify.Watcher
• type AsyncFSM
  • func NewAsyncFSM(events *pubsub.PubSub[string, Event]) *AsyncFSM
  • func NewCustomAsyncFSM(events *pubsub.PubSub[string, Event], delay time.Duration) *AsyncFSM
  • func (fsm *AsyncFSM) AbortWork()
  • func (fsm *AsyncFSM) CurrentState() string
  • func (fsm *AsyncFSM) IsWaiting() bool
  • func (fsm *AsyncFSM) StartWork()
• type Cluster
  • func NewCluster(events *pubsub.PubSub[string, Event], ...) *Cluster
  • func (ps *Cluster) Start()
• type CounterListener
  • func NewCounterListener(events *pubsub.PubSub[string, Event]) *CounterListener
  • func (n *CounterListener) OnNewCount(ev percounter.CountEvent)
• type Event
  • func GetReplicasEvent(count int) Event
  • func NewEventWithParam(name string, p any) Event
  • func NewEventWithReason(name, reason string) Event
  • func NewSimpleEvent(name string) Event
• type FlyPeerLocator
  • func NewFlyPeerLocator(flyDiscoveryDomainName string) *FlyPeerLocator
  • func (l *FlyPeerLocator) GetMyIP() string
  • func (l *FlyPeerLocator) GetPeers() ([]string, int, error)
• type LoadBalancer
• type PeerLocator
  • func ChoosePeerLocator() PeerLocator
• type PersistentClusterObserver
  • func NewPersistentClusterObserver(identity string, myIP string, events *pubsub.PubSub[string, Event]) *PersistentClusterObserver
  • func (o *PersistentClusterObserver) AfterMessageReceived(peer string, msg []byte)
  • func (o *PersistentClusterObserver) AfterMessageSent(peer string, msg []byte)
• type ResponseForwarding
• type Server
  • func NewServerWithOptions(port string, events *pubsub.PubSub[string, Event], fs http.FileSystem, ...) *Server
  • func (s *Server) Run(port string)
  • func (s *Server) WaitToDrainConnections()
• type ServerStatus
• type TraefikPeerLocator
  • func NewTraefikPeerLocator(traetraefikServicesUrl string) *TraefikPeerLocator
  • func (l *TraefikPeerLocator) GetMyIP() string
  • func (l *TraefikPeerLocator) GetPeers() ([]string, int, error)
• type TraefikReplicas
• type TraefikServerResponse
• type VisitorTracker
  • func NewVisitorTracker(events *pubsub.PubSub[string, Event]) *VisitorTracker
  • func (v *VisitorTracker) Joined()
  • func (v *VisitorTracker) Left()

Constants

const ClosedConnectionsCounter = "closed-connections"

const ClusterMessageEvent = "ClusterMessage"

const ClusterMessageTopic = "cluster-events"

const InternalTopic = "internal-events"

const NewConnectionsCounter = "newconnections"

const SourceReplicaIdKey = "Source-Replica-Id"

const StartedConnectionsCounter = "started-connections"

const Topic = "events"

const TotalClusterVisitorsActiveEvent = "TotalClusterVisitorsActive"

const TotalVisitorsEvent = "TotalVisitors"

const VisitorJoinedEvent = "VisitorJoined"

const VisitorLeftEvent = "VisitorLeft"

const VisitorsActiveEvent = "VisitorsActive"

Variables

var ClusterObservabilityEnabled = false

var DoEmbed = false

Functions

func GetCounterDirectory

func GetCounterDirectory() string

func GetCounterIdentity

func GetCounterIdentity() string

func GetFS

func GetFS() http.FileSystem

func Refresh

func Refresh()

func StartWatching

func StartWatching(eventPublisher *pubsub.PubSub[string, Event]) *fsnotify.Watcher

Types

type AsyncFSM

type AsyncFSM struct {
	phony.Inbox
	// contains filtered or unexported fields
}

func NewAsyncFSM

func NewAsyncFSM(events *pubsub.PubSub[string, Event]) *AsyncFSM

func NewCustomAsyncFSM

func NewCustomAsyncFSM(events *pubsub.PubSub[string, Event], delay time.Duration) *AsyncFSM

func (*AsyncFSM) AbortWork

func (fsm *AsyncFSM) AbortWork()

func (*AsyncFSM) CurrentState

func (fsm *AsyncFSM) CurrentState() string

func (*AsyncFSM) IsWaiting

func (fsm *AsyncFSM) IsWaiting() bool

sync queries - not to be used from within actor behaviors (methods)

func (*AsyncFSM) StartWork

func (fsm *AsyncFSM) StartWork()

type Cluster

type Cluster struct {
	// contains filtered or unexported fields
}

func NewCluster

func NewCluster(events *pubsub.PubSub[string, Event], clusterEventObserver *PersistentClusterObserver, cluster zmqcluster.Cluster) *Cluster

func (*Cluster) Start

func (ps *Cluster) Start()

type CounterListener

type CounterListener struct {
	phony.Inbox
	// contains filtered or unexported fields
}

func NewCounterListener

func NewCounterListener(events *pubsub.PubSub[string, Event]) *CounterListener

func (*CounterListener) OnNewCount

func (n *CounterListener) OnNewCount(ev percounter.CountEvent)

type Event

type Event struct {
	Timestamp  string                 `json:"timestamp"`
	Name       string                 `json:"name"`
	Properties map[string]interface{} `json:"properties"`
}

func GetReplicasEvent

func GetReplicasEvent(count int) Event

func NewEventWithParam

func NewEventWithParam(name string, p any) Event

func NewEventWithReason

func NewEventWithReason(name, reason string) Event

func NewSimpleEvent

func NewSimpleEvent(name string) Event

type FlyPeerLocator

type FlyPeerLocator struct {
	// contains filtered or unexported fields
}

func NewFlyPeerLocator

func NewFlyPeerLocator(flyDiscoveryDomainName string) *FlyPeerLocator

func (*FlyPeerLocator) GetMyIP

func (l *FlyPeerLocator) GetMyIP() string

func (*FlyPeerLocator) GetPeers

func (l *FlyPeerLocator) GetPeers() ([]string, int, error)

type LoadBalancer

type LoadBalancer struct {
	Servers            []TraefikServerResponse `json:"servers"`
	PassHostHeader     bool                    `json:"passHostHeader"`
	ResponseForwarding ResponseForwarding      `json:"responseForwarding"`
}

type PeerLocator

type PeerLocator interface {
	GetPeers() ([]string, int, error)
	GetMyIP() string
}

func ChoosePeerLocator

func ChoosePeerLocator() PeerLocator

type PersistentClusterObserver

type PersistentClusterObserver struct {
	phony.Inbox
	// contains filtered or unexported fields
}

func NewPersistentClusterObserver

func NewPersistentClusterObserver(identity string, myIP string, events *pubsub.PubSub[string, Event]) *PersistentClusterObserver

func (*PersistentClusterObserver) AfterMessageReceived

func (o *PersistentClusterObserver) AfterMessageReceived(peer string, msg []byte)

func (*PersistentClusterObserver) AfterMessageSent

func (o *PersistentClusterObserver) AfterMessageSent(peer string, msg []byte)

type ResponseForwarding

type ResponseForwarding struct {
	FlushInterval string `json:"flushInterval"`
}

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewServerWithOptions

func NewServerWithOptions(port string,
	events *pubsub.PubSub[string, Event],
	fs http.FileSystem,
	delay time.Duration) *Server

func (*Server) Run

func (s *Server) Run(port string)

func (*Server) WaitToDrainConnections

func (s *Server) WaitToDrainConnections()

type ServerStatus

type ServerStatus map[string]string

type TraefikPeerLocator

type TraefikPeerLocator struct {
	// contains filtered or unexported fields
}

func NewTraefikPeerLocator

func NewTraefikPeerLocator(traetraefikServicesUrl string) *TraefikPeerLocator

func (*TraefikPeerLocator) GetMyIP

func (l *TraefikPeerLocator) GetMyIP() string

func (*TraefikPeerLocator) GetPeers

func (l *TraefikPeerLocator) GetPeers() ([]string, int, error)

type TraefikReplicas

type TraefikReplicas struct {
	LoadBalancer LoadBalancer `json:"loadBalancer"`
	Status       string       `json:"status"`
	UsedBy       []string     `json:"usedBy"`
	ServerStatus ServerStatus `json:"serverStatus"`
	Name         string       `json:"name"`
	Provider     string       `json:"provider"`
	Type         string       `json:"type"`
}

via https://app.quicktype.io/

type TraefikServerResponse

type TraefikServerResponse struct {
	URL string `json:"url"`
}

type VisitorTracker

type VisitorTracker struct {
	phony.Inbox
	// contains filtered or unexported fields
}

func NewVisitorTracker

func NewVisitorTracker(events *pubsub.PubSub[string, Event]) *VisitorTracker

func (*VisitorTracker) Joined

func (v *VisitorTracker) Joined()

func (*VisitorTracker) Left

func (v *VisitorTracker) Left()