wetware / casm

Universal middleware for decentralized computing

🧬 CASM

Cluster Assembly. Modular middleware for distributed computing

What is CASM?

CASM is short for Cluster Assembly. It is a low-level toolkit for developing efficient, reliable and secure distributed systems. It is entirely peer-to-peer and requires no coordinator nodes or other infrastructure. It is built using libp2p and integrates seamlessly into the Protocol Labs ecosystem.

Design

CASM appeals to developers in search of firm ground on which to build distributed systems. It offers zero-cost abstractions¹ that put you in control of trade-offs, while enforcing key properties of well-behaved systems.

In particular, the following invariants are preserved throughout the public API:

1. Cluster membership is dynamic.
2. Data is automatically signed and validated.
3. Network protocols are partition-available and low-latency.
4. Security is provided through Object Capabilities.

Users can stack additional guarantees in application logic. For example, you can build a consistent database out of CASM parts.

Features

CASM follows a modular, "Lego bricks" design, allowing you to pick and choose the pieces you want.

Feature	Package	Description
RPC	pkg/	Fast & extensible RPC for communicating between nodes, with capability-based security.
Bootstrap	pkg/boot	Pluggable strategies for discovering and joining clusters.
Peer Exchange	pkg/pex	Lightweight gossip-based protocol for randomly sampling peers. Ideal for building caches.
Clustering	pkg/cluster	Unstructured service providing a global view of the cluster2.

Getting Started

Installation

Run go get github.com/wetware/casm with modules enabled.

Getting Support

The best place to get help is on Matrix, or by asking a question on the Wetware Q&A Board.

We're friendly! Drop in and say hi! 👋

CASM Users

Organizations using CASM in production

Blocknative

Contributors

Footnotes

1. The term "zero-cost" is obviously a figure of speech, which is intended to emphasize the following point. As a matter of principle, CASM emphasizes "thin", non-leaky abstractions that do not significantly impact performance.
2. In the spirit of zero-cost abstractions, CASM's clustering protocol provides PA/EL guarantees. No effort is made to provide a consistent view between nodes, because (a) it is rarely needed in practice, and (b) this configuration provides you with the greatest flexibility. CASM provides an ideal foundation on which to build more specialized (including consistent) systems.

References

• UnsServ: Unstructured Peer-to-Peer Services [pdf]

casm's Issues

Fix ordering of records when using LowerBound

See hashicorp/go-memdb#113 (comment) for context. Not urgent, since we've hacked together a workaround.

Data race in PeX

Running TestPeX_NNodes with the race detector enabled will occasionally produce the following warning:

==================
WARNING: DATA RACE
Read at 0x00c0017c7090 by goroutine 315:
  runtime.slicecopy()
      /usr/local/go/src/runtime/slice.go:284 +0x0
  capnproto.org/go/capnp/v3.(*Message).Marshal()
      /Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/message.go:840 +0x984
  github.com/wetware/casm/pkg/pex.gossipStore.StoreRecords()
      /Users/lthibault/Go/src/github.com/wetware/casm/pkg/pex/store.go:96 +0x284
  github.com/wetware/casm/pkg/pex.(*gossiper).PushPull()
      /Users/lthibault/Go/src/github.com/wetware/casm/pkg/pex/gossip.go:202 +0x724
  github.com/wetware/casm/pkg/pex.(*gossiper).newHandler.func1()
      /Users/lthibault/Go/src/github.com/wetware/casm/pkg/pex/gossip.go:223 +0x2fe
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).SetStreamHandlerMatch.func1()
      /Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/p2p/host/basic/basic_host.go:583 +0x86
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).newStreamHandler·dwrap·3()
      /Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/p2p/host/basic/basic_host.go:412 +0x74

Previous write at 0x00c0017c7095 by goroutine 279:
  encoding/binary.littleEndian.PutUint64()
      /usr/local/go/src/encoding/binary/binary.go:89 +0x174
  capnproto.org/go/capnp/v3.(*Segment).writeUint64()
      /Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/segment.go:85 +0x74
  capnproto.org/go/capnp/v3.Struct.SetUint64()
      /Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/struct.go:277 +0xa4
  github.com/wetware/casm/internal/api/pex.Gossip.SetHop()
      /Users/lthibault/Go/src/github.com/wetware/casm/internal/api/pex/pex.capnp.go:41 +0x176
  github.com/wetware/casm/pkg/pex.(*GossipRecord).IncrHop()
      /Users/lthibault/Go/src/github.com/wetware/casm/pkg/pex/record.go:64 +0x26
  github.com/wetware/casm/pkg/pex.View.incrHops()
      /Users/lthibault/Go/src/github.com/wetware/casm/pkg/pex/record.go:158 +0x57
  github.com/wetware/casm/pkg/pex.(*gossiper).mutexMerge()
      /Users/lthibault/Go/src/github.com/wetware/casm/pkg/pex/gossip.go:306 +0x595
  github.com/wetware/casm/pkg/pex.(*gossiper).PushPull()
      /Users/lthibault/Go/src/github.com/wetware/casm/pkg/pex/gossip.go:201 +0x5ce
  github.com/wetware/casm/pkg/pex.(*gossiper).newHandler.func1()
      /Users/lthibault/Go/src/github.com/wetware/casm/pkg/pex/gossip.go:223 +0x2fe
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).SetStreamHandlerMatch.func1()
      /Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/p2p/host/basic/basic_host.go:583 +0x86
  github.com/libp2p/go-libp2p/p2p/host/basic.(*BasicHost).newStreamHandler·dwrap·3()
      /Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/p2p/host/basic/basic_host.go:412 +0x74

Speculative root-cause:

1. At least 2 concurrent gossip rounds, G1 & G2 are in-flight on a single host.
2. G1 calls incrHops() as last step of mutexMerge()
3. Concurrently, G2 has just exited mutexMerge() and calls StoreRecords => marshal record while hop is being incremented.

Panic in BenchmarkIterator

Unclear if this is an actual problem, or a bug in the test.

go test -benchmem -run=^$ -bench ^BenchmarkIterator$ github.com/wetware/casm/pkg/cluster

goos: darwin
goarch: amd64
pkg: github.com/wetware/casm/pkg/cluster
cpu: Intel(R) Core(TM) i7-1068NG7 CPU @ 2.30GHz
BenchmarkIterator-8   	fatal error: sync: unlock of unlocked mutex

goroutine 46 [running]:
sync.fatal({0x100cd7bda?, 0x0?})
	/usr/local/go/src/runtime/panic.go:1031 +0x1e
sync.(*Mutex).unlockSlow(0xc00123f390, 0xffffffff)
	/usr/local/go/src/sync/mutex.go:229 +0x3c
sync.(*Mutex).Unlock(...)
	/usr/local/go/src/sync/mutex.go:223
capnproto.org/go/capnp/v3.resolveHook(0x100f19830?)
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/capability.go:264 +0x5d
capnproto.org/go/capnp/v3.Client.State({0xc00132c038?})
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/capability.go:495 +0x85
github.com/wetware/casm/pkg/util/stream.promise.Failed({{0xc00132c038?}, 0xc001229dd0?})
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/util/stream/stream.go:138 +0x2a
github.com/wetware/casm/pkg/util/stream.(*Stream[...]).Call(0xc00013a0a0, {0x100f19830, 0xc0008bb140}, 0x4)
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/util/stream/stream.go:63 +0x1a5
github.com/wetware/casm/pkg/cluster.iterator.func1({0x129e89fa8?, 0x1017fe840?})
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/cluster/server.go:125 +0x5c
github.com/wetware/casm/pkg/cluster.Server.bind({{0x100f104c0?, 0xc000290140?}}, 0xc000090120, 0xc000130018)
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/cluster/server.go:102 +0xb1
github.com/wetware/casm/pkg/cluster.Server.Iter({{0x100f104c0?, 0xc000290140?}}, {0x100f19830, 0xc0008bb140}, {0xc00011cd80?})
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/cluster/server.go:55 +0x29c
github.com/wetware/casm/internal/api/routing.View_Methods.func2({0x100f19830?, 0xc0008bb140?}, 0xc0000b6240?)
	/Users/lthibault/Go/src/github.com/wetware/casm/internal/api/routing/routing.capnp.go:306 +0x38
capnproto.org/go/capnp/v3/server.(*Server).handleCall(0x1000479d1?, {0x100f19830?, 0xc0008bb140?}, 0xc00011eb60)
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:206 +0x97
capnproto.org/go/capnp/v3/server.(*Server).handleCalls.func2(0x100f19830?, 0xc0009200c0?, {0x100f19830?, 0xc0008bb140?}, 0x1?)
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:182 +0x4a
capnproto.org/go/capnp/v3/server.(*Server).handleCalls(0xc00024a000, {0x100f19830?, 0xc000920080})
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:183 +0x145
created by capnproto.org/go/capnp/v3/server.New
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:122 +0x325

goroutine 1 [chan receive]:
testing.(*B).doBench(0xc000150240)
	/usr/local/go/src/testing/benchmark.go:285 +0x7f
testing.(*benchContext).processBench(0xc000283470, 0x238?)
	/usr/local/go/src/testing/benchmark.go:589 +0x3aa
testing.(*B).run(0xc000150240?)
	/usr/local/go/src/testing/benchmark.go:276 +0x67
testing.(*B).Run(0xc000294900, {0x100c1aa42?, 0x63892750?}, 0x100dc6390)
	/usr/local/go/src/testing/benchmark.go:677 +0x453
testing.runBenchmarks.func1(0xc000294900?)
	/usr/local/go/src/testing/benchmark.go:550 +0x6e
testing.(*B).runN(0xc000294900, 0x1)
	/usr/local/go/src/testing/benchmark.go:193 +0x102
testing.runBenchmarks({0x100cdd199, 0x23}, 0x101830480?, {0x10157de40, 0x1, 0x40?})
	/usr/local/go/src/testing/benchmark.go:559 +0x418
testing.(*M).Run(0xc0002b59a0)
	/usr/local/go/src/testing/testing.go:1733 +0x811
main.main()
	_testmain.go:63 +0x1aa

goroutine 39 [runnable]:
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc0035400c0?, {{0xc003523a20?, 0x4?, 0x4?}, 0x20?}, {0x100f10220?, 0xc0008ced00?})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:232 +0x385
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc003521500, {{0xc003523a20?, 0x0?, 0x4?}, 0xc003522d30?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc00089c2a0, {{0xc003523a20?, 0x100054812?, 0x10?}, 0xc0000c1180?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc0000d0000, {{0xc003523a20?, 0x0?, 0x4?}, 0xc0000c0e90?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc00027bda0, {{0xc003523a20?, 0x0?, 0x4?}, 0xc0000c1d80?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc000123920, {{0xc003523a20?, 0x0?, 0x4?}, 0xc000040ac0?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc0002a1260, {{0xc003523a20?, 0x0?, 0x4?}, 0xc00090a9a0?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc0002a05a0, {{0xc003523a20?, 0x0?, 0x4?}, 0xc000257c20?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc000267c20, {{0xc003523a20?, 0x0?, 0x1299fffff?}, 0xc0002579c0?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc000912a80, {{0xc003523a20?, 0x129800f00?, 0x0?}, 0xc000125480?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc000266900, {{0xc003523a20?, 0x10001aa4b?, 0x1299fffff?}, 0xc000256ce0?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc0000bf2c0, {{0xc003523a20?, 0x0?, 0x1299fffff?}, 0xc0000c0420?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc000266300, {{0xc003523a20?, 0x129b85670?, 0x2?}, 0xc0002568e0?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).insert(0xc000123da0, {{0xc003523a20?, 0x10?, 0xc0035239e0?}, 0x10?}, {0x100f10220?, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:264 +0x25b
github.com/libp2p/go-cidranger.(*prefixTrie).Insert(0xc000123da0, {0x100f10220, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/trie.go:95 +0xdb
github.com/libp2p/go-cidranger.(*versionedRanger).Insert(0x100bdc4f7?, {0x100f10220, 0xc0008ced00})
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/version.go:29 +0xd4
github.com/libp2p/go-libp2p-asn-util.newAsnStore()
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/asn.go:61 +0x163
github.com/libp2p/go-libp2p-asn-util.newIndirectAsnStore.func1()
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/asn.go:89 +0x65
created by github.com/libp2p/go-libp2p-asn-util.newIndirectAsnStore
	/Users/lthibault/Go/pkg/mod/github.com/libp2p/[email protected]/asn.go:87 +0x9b

goroutine 40 [select]:
github.com/ipfs/go-log/writer.(*MirrorWriter).logRoutine(0xc000287a40)
	/Users/lthibault/Go/pkg/mod/github.com/ipfs/[email protected]/writer/writer.go:71 +0x11f
created by github.com/ipfs/go-log/writer.NewMirrorWriter
	/Users/lthibault/Go/pkg/mod/github.com/ipfs/[email protected]/writer/writer.go:36 +0xca

goroutine 24 [chan receive]:
github.com/wetware/casm/pkg/cluster.handler.Next(...)
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/cluster/client.go:98
github.com/wetware/casm/pkg/cluster.Iterator.Next(...)
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/cluster/client.go:77
github.com/wetware/casm/pkg/cluster_test.BenchmarkIterator(0xc000150240)
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/cluster/view_test.go:211 +0x5b5
testing.(*B).runN(0xc000150240, 0x2710)
	/usr/local/go/src/testing/benchmark.go:193 +0x102
testing.(*B).launch(0xc000150240)
	/usr/local/go/src/testing/benchmark.go:334 +0x1c5
created by testing.(*B).doBench
	/usr/local/go/src/testing/benchmark.go:284 +0x6c

goroutine 52 [semacquire]:
sync.runtime_SemacquireMutex(0x0?, 0x0?, 0x1008ea762?)
	/usr/local/go/src/runtime/sema.go:77 +0x25
sync.(*Mutex).lockSlow(0xc00013a0a8)
	/usr/local/go/src/sync/mutex.go:171 +0x165
sync.(*Mutex).Lock(...)
	/usr/local/go/src/sync/mutex.go:90
github.com/wetware/casm/pkg/util/stream.(*Stream[...]).next(0xc00013a0a0?)
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/util/stream/stream.go:95 +0x5a
github.com/wetware/casm/pkg/util/stream.(*Stream[...]).loop(0xc00013a0a0)
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/util/stream/stream.go:88 +0xdb
created by github.com/wetware/casm/pkg/util/stream.(*Stream[...]).Call
	/Users/lthibault/Go/src/github.com/wetware/casm/pkg/util/stream/stream.go:53 +0x179

goroutine 51 [select]:
capnproto.org/go/capnp/v3/server.(*Server).handleCalls.func1()
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:175 +0xb3
created by capnproto.org/go/capnp/v3/server.(*Server).handleCalls
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:173 +0x125

goroutine 47 [chan receive]:
capnproto.org/go/capnp/v3.(*ClientPromise).shutdown(...)
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/capability.go:685
capnproto.org/go/capnp/v3.(*Promise).Resolve(0x0?, {0x0, 0x0, 0x0, {0x0, 0x0}, 0x0, 0x0}, {0x0, 0x0})
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/answer.go:190 +0x145
capnproto.org/go/capnp/v3.(*Promise).Fulfill(...)
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/answer.go:130
capnproto.org/go/capnp/v3/server.(*structReturner).Return(0xc00123f310, {0x0?, 0x0})
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/answer.go:240 +0x2e5
capnproto.org/go/capnp/v3/server.(*Server).handleCall(0x1000479d1?, {0x100f19830?, 0xc002c42940?}, 0xc001321ee0)
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:214 +0x14f
capnproto.org/go/capnp/v3/server.(*Server).handleCalls.func2(0x100f19830?, 0xc0008bb140?, {0x100f19830?, 0xc002c42940?}, 0x1?)
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:182 +0x4a
capnproto.org/go/capnp/v3/server.(*Server).handleCalls(0xc00024a070, {0x100f19830?, 0xc000920140})
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:183 +0x145
created by capnproto.org/go/capnp/v3/server.New
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:122 +0x325

goroutine 1693 [select]:
capnproto.org/go/capnp/v3/server.(*Server).handleCalls.func1()
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:175 +0xb3
created by capnproto.org/go/capnp/v3/server.(*Server).handleCalls
	/Users/lthibault/Go/pkg/mod/capnproto.org/go/capnp/[email protected]/server/server.go:173 +0x125
exit status 2
FAIL	github.com/wetware/casm/pkg/cluster	0.322s
FAIL

Add piggybacked information to neighbors

Problem

In order to implement Gossiping, it is necessary to piggyback information together with the neighbors.

The gossiping I suggest to implement is based on the following work: The Peer Sampling Service: Experimental Evaluation of Unstructured Gossip-Based Implementations. There, periodically, every node of the network exchanges the neighbors list with one of its peers. Then, it merges the current neighbors list with the received peers during the exchange.When merging, the node only keeps the youngest peers, so that the oldest are evicted.

The pseudocode of gossiping is the following:

do forever
  wait(T time units)
  p ← selectPeer()
  // 0 is the initial hop count
  myDescriptor ← (myAddress, 0)
  buffer ← merge(neighbors,{myDescriptor})
  send buffer to p
  
  receive neighbors_p from p
  neighbors_p ← increaseHopCount(neighbors_p)
  buffer ← merge(neighbors_p ,neighbors)
  buffer ← sortByHopCount(buffer)
  updateNeighbors(neighbors[:MaxNeighbors])

Therefore, the exchanged neighbors must contain a field representing how old a neighbor is. Otherwise, it is not possible to decide what neighbors to keep when merging. For that, when exchanging neighbors we should piggyback a Hop field . Hop is increased whenever the neighbor is sent from one peer to another.

Solution

I suggest creating a new struct for representing peers that contains two fields peer.AddrInfo and Hop. Then, these structs could be stored in one of the two following places:

1. In neighborhood.go, so that in vtx contains Hop together with the peer.ID.
2. In net.go within the Overlay. This would require to maintain a second list of neighbors that works independently from neighborhood. This would require a loop that it is subscribed to the events, which updates the neighbors when based on the events.

Consider bootstrap record with separate signature domain.

A libp2p Record defines a "signature domain", with the following purpose and specification:

Signatures can be used for a variety of purposes, and a signature made for a specific purpose MUST NOT be considered valid for a different purpose.

Without this property, an attacker could convince a peer to sign a payload in one context and present it as valid in another, for example, presenting a signed address record as a pubsub message.

We separate signatures into "domains" by prefixing the data to be signed with a string unique to each domain. This string is not contained within the payload or the outer envelope structure. Instead, each libp2p subsystem that makes use of signed envelopes will provide their own domain string when constructing the envelope, and again when validating the envelope. If the domain string used to validate is different from the one used to sign, the signature validation will fail.

Domain strings may be any valid UTF-8 string, but should be fairly short and descriptive of their use case, for example "libp2p-routing-record".

(source)

I'm wondering if it might not be a good idea to separate the bootstrap domain from the ordinary routing domain. I have often remarked that libp2p's discovery system conflates bootstrapping and peer sampling, and it seems to me that these are different contexts, so I'm not sure how I feel about using peer.PeerRecord in the boot package.

I don't have a clear attack scenario in mind, so I'll instead appeal to a vague notion of "defense in depth". Adding a boot.Record type allows a node to sign the namespace(s) for which the record is valid, in addition to the peer ID, sequence number and address list. It might be defined by the following capnp schema:

struct Record {
    peerID @0  :Text;
    addrs  @1  :List(Data);  # binary-marshaled multiaddrs, identical to peer.Record
    seq    @3  :UInt64;
    ns     @4  :List(Text);  # list of namespaces
}

The domain record string could be casm.boot

I don't see any downsides and the effort is pretty low. Ultimately, I would want this record type to be used by all the services provided by pkg/boot, including PeX and the port scanner.

@aratz-lasa, what do you think?

Investigate traffic shaping in Testground

We need to support two cases:

1. Dynamically isolate individual nodes from the cluster
2. Dynamically introduce cluster partitions

This should be achievable using the Docker executor.

See also: #8

Reduce allocations when updating routing table

The routing table uses an immutable radix tree to store its indexes, giving it the (mostly) lock-free properties we have come to enjoy. However, this also means insertions -- and particularly updates -- allocate new tree nodes. Below is a benchmark of the routing table's Upsert() method.

cpu: Intel(R) Core(TM) i7-1068NG7 CPU @ 2.30GHz
BenchmarkRoutingTable_upsert/Insert-8         	   43494	     27782 ns/op	   36280 B/op	     210 allocs/op
BenchmarkRoutingTable_upsert/Drop-8           	 1000000	      1003 ns/op	     256 B/op	       6 allocs/op
BenchmarkRoutingTable_upsert/Re-Insert-8      	   48871	     24032 ns/op	   33753 B/op	     235 allocs/op

The 3rd benchmark (Re-Insert) tests an update operation on an existing record. This one is particularly interesting because in real-world access patterns, the bulk of operations is expected to be updates on existing records, with inserts and deletions happening only in short bursts (for instance, during partition merging). This suggests an optimization strategy.

Optimization Strategy

Rather than storing routing.Records directly in the database, we might consider storing an atomic.Value in the database that contains routing.Record, allowing us to mutate existing records on updates. This should be safe, since updates occur on records that are strictly identical, save for the increased sequence number. Three invariants guarantee that no other thread is reading the sequence number during an atomic mutation:

1. Atomic mutations MUST take place in a Write transaction
2. Sequence numbers MUST NOT be indexed
3. Iteration over read-only snapshots MUST NOT depend on Seq (e.g.: no filtering based on Seq).

Note that this mutation may make it appear to read-only iterators as if the sequence number has changed within a snapshot. Invariant 2 ensures that this does not cause any incorrect or dangling indexes. Invariant 3 ensures that no behaviors depend on mutable data that can be accessed from a read-only snapshot.

Switch to prometheus-based metrics

Libp2p appears to use https://github.com/prometheus/client_golang for metrics. We should adjust CASM to match.

Hopefully the casm.Metrics interface can be trivially adapted.

Verify context usage in fx.Lifecycle

I have noticed suspicious usage of the context passed into OnStart in the pex package. It appears as though the code expects this to be a long-lived context, which is not the case in Wetware.

Investigate error when exceeding 30 nodes in testground

Example error output:

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0xcf2005]
goroutine 131 [running]:
github.com/wetware/lab/pkg/boot.(*RedisDiscovery).syncRedis(0xc000224380, 0x11076b0, 0xc000392280, 0xed5442, 0xc, 0xc00018a000, 0x1b, 0x20, 0x0, 0x0)
/plan/pkg/boot/boot.go:78 +0x305
github.com/wetware/lab/pkg/boot.(*RedisDiscovery).Advertise(0xc000224380, 0x11076b0, 0xc000392240, 0xed5442, 0xc, 0x0, 0x0, 0x0, 0x0, 0x0, ...)
/plan/pkg/boot/boot.go:37 +0x17b
github.com/wetware/casm/pkg/pex.(*discover).newTopic.func2(0xc000135440, 0xed5442, 0xc, 0xc00029e960, 0x11076b0, 0xc000392240)
/go/pkg/mod/github.com/wetware/[email protected]/pkg/pex/discovery.go:193 +0x1b7
created by github.com/wetware/casm/pkg/pex.(*discover).newTopic
/go/pkg/mod/github.com/wetware/[email protected]/pkg/pex/discovery.go:184 +0x1da

Evaluate performance of PeX in a test setting

As the PeX implementation nears completion, I think our priority should be to evaluate its performance along the following dimensions:

1. Correctness
2. Resilience to partitions
3. Resilience to partial failure
4. Time complexity

I've used matrix to run simulations inside of unit tests (see: TestPeerExchange_Simulation) and am so far very pleased with the results. It seems like a good starting point for evaluating and optimizing PeX.

In contrast to Testground, Matrix is simpler to configure, can be run directly in unit tests/benchmarks, doesn't require any configuration on the dev machine and doesn't require setting up a separate repository (or fiddling with go.mod). The trade-off is that we use an in-process transport, so we cannot use it to test layer-3 networking behaviors. I don't think we care, though.

Below are a list of questions I have about PeX. These should dictate the tests/simulations that need doing. Please feel free to append to this list.

• How fast does PeX converge on a uniform distribution of records?
• How resilient is the overlay to partial failure without partitions?
• How long does a partition "remember" records from another partition?
• How does increasing the fanout during each gossip round affect the convergence rate?
• How does a pure-rand strategy affect the convergence rate relative to the current (hybrid) strategy?
• How does #7 compare to rand/hybrid strategies?

At this point, I think we should be going for low-hanging fruit. How can we apply the 80/20 rule here? I would ideally like to be confident in PeX by mid-September.

@aratz-lasa Thoughts? Can you look into this?

Block in Vat.Embargo until host unregisters all stream handlers

Vat.Embargo is used to remove a capability from the vat. Callers should be able to know exactly when this operation succeeds, so that they can reason about access revocation. Unfortunately, libp2p's Host.RemoveStreamHandler is asynchronous, so it is often the case that Embargo returns before the stream handler is actually disabled.

Two approaches to investigate:

• Is there a signal on event.Bus that we can wait for?
• Is there a network.Notifiee hook that we can tap into?

Add network mapping primitives

Libp2p conflates bootstrapping with discovery. The distinction is:

1. Bootstrapping: establishing an initial connection to the cluster.
2. Discovery: using the initial connection to establish subsequent connections to the cluster.

In practice, libp2p leaves bootstrapping as an exercise to the user. CASM should provide basic "building-block" style abstractions to compose bootstrap strategies.

The following should be included:

• Abstract bootstrap interface that satisfies discovery.Discovery (perhaps also Advertiser?).
• TCP port-scanning primitives that compose with the bootstrap interface.
• UDP port-scanning primitives that behave similarly.

We may also consider adding the following, either here or in Wetware:

• NetMapper type that is capable of actively crawling a network in search of bootstrap services.

Notes

Security

All bootstrap messages should be signed using libp2p's record.Record API. Signature verification should be specified by the protocol as non-optional, and enforced in implementation.

Design Principles

Bootstrap protocols should heavily optimize for network reachability. For UDP in particular, this means enforcing a payload MTU below 508 bytes, which guarantees that packets are universally deliverable as per IP spec.

Note that middleware boxes may annotate payloads, which could tip them over the 508-byte limit. We should therefore provide comfortable overhead (which also gives us some reserved space for future protocol extensions. A limit of 448 bytes seems appropriate.

Improve PeX stabilization time and partition-resistance.

Given PeX's design goals, I'm inclined to err on the side of partition-resistance when choosing policies and parameters. With regards to view selection, we know that a pure rand policy implies the following trade-off:

• Advantage: Partitions are slower to "forget" records from other partitions. This is helpful for repairing partitions and for reconnecting orphaned nodes. I think it is a significant advantage to be able to do this without a bootstrap service.
• Disadvantage: The cluster will be slower to converge on a uniform distribution of records. I believe the rate of convergence under rand is linear, vs exponential with tail. This is not awful, but it puts constraints on max cluster size.

After reading [0] and [1], I think we might be able to do better (Kermarrec to the rescue 😝 )!

A central observation of these papers is that the view size does not impact the rate at which information propagates in an overlay, and therefore has no effect on convergence time. However, the fan-out factor f is highly determining of convergence rate. A value of f > log(n) where n is maximum size of the cluster is sufficient to deliver gossip to every peer with a very high probability¹.

We currently set a value of f=1, as described in Jelasity et al., so it seems plausible that we should be able to increase this value and combine it with a pure rand policy². In theory, this should give us the best of both worlds (though clearly we should measure this).

Practically speaking, it seems like this should be a simple matter of performing peer exchange with f peers during each gossip round.

@aratz-lasa Thoughts?

Footnotes

1. I'm unclear on the appropriate base for the logarithm. Is it log10? The natural log? Log of f? Do you know?
2. Do we still need to track hop if we're only using rand? 🤔

References

[0] Lightweight probabilistic broadcast
[1] Epidemic information dissemination in distributed systems

TestRoutingTable_concurrent fails intermittently

To reproduce:

1. Uncomment test in routing_test.go
2. Run go test -timeout 10s ./...

Note that the test failure appears stochastic, possibly incurring a deadlock on occasion as well.

Set multicast interface based on GOOS

Currently we use the lo0 interface which is the one used in MacOS

Add support for LZ4-compressed transport to casm.Vat

Suppress `context cancelled` debug logs in Crawler

Example:

 $ ww client ls -json | jq
DEBU[0000] bootstrapping namespace                       ns=bn peer=12D3KooWAwzrBHs3TZrhLKiLn6P31WXFmVbBVbH6AZGZyTmqCD6U
DEBU[0000] found peer                                    addrs="[/ip4/10.0.1.204/udp/2020/quic /ip4/127.0.0.1/udp/2020/quic]" ns=bn peer=12D3KooWQcRsSiUZrxjz9qkjfwtefyueNFjftKmr4HV2TLYbUiRK
DEBU[0000] failed to send request packet                 error="context canceled" ns=bn peer=12D3KooWAwzrBHs3TZrhLKiLn6P31WXFmVbBVbH6AZGZyTmqCD6U to="10.0.1.75:8822"
{
  "server": "53a3d5808f5453ea",
  "peer": "12D3KooWQabEDcSzmZgbqJ2zQo3aAJcSxu1fEyAx83SYHSmbfhgt",
  "host": "wetware-a1-stage",
  "meta": {
    "az": "us-east-1a",
    "instance": "wetware-a1-stage",
    "region": "us-east-1"
  }
}
{
  "server": "fecd4b380b840255",
  "peer": "12D3KooWQcRsSiUZrxjz9qkjfwtefyueNFjftKmr4HV2TLYbUiRK",
  "host": "wetware-a0-stage",
  "meta": {
    "az": "us-east-1a",
    "instance": "wetware-a0-stage",
    "region": "us-east-1"
  }
}

TestSampler/Abort is flaky

=== CONT  TestSampler/Abort
    sampler_test.go:110: stopped
    sampler_test.go:75: 
        	Error Trace:	/home/runner/work/casm/casm/pkg/debug/sampler_test.go:75
        	Error:      	Not equal: 
        	            	expected: 50ms
        	            	actual  : 100ms
        	Test:       	TestSampler/Abort
        	Messages:   	should run for 50ms

I don't think there's an actual problem here. It's likely that the test is too sensitive.

Silence EOF errors from boot socket when restarting.

We currently log failed when unmarshalling the envelope: unexpected EOF at ERROR level when a casm-based project (namely: wetware) restarts. These aren't really errors, as they simply indicate the socket has been closed, and should be logged at DEBUG level instead.

Open questions:

• How can we distinguish between this and an actual failure? (Can PacketConn ever fail?)

Remove go.uber.org/atomic

Now that we require go1.19, we can rely on updated sync/atomic API instead of importing go.uber.org/atomic in most places. One use of go.uber.org/atomic however remains in pkg/cluster/routing. The default routing.Table implementation uses Uber's atomic.Time, which does not have a direct equivalent in sync/atomic.

One possibility is to use atomic.Pointer[time.Time], but as I was reading over the code, it suddenly occurred to me that using atomics is going to result in each item time.Time instance being heap-allocated. It might actually be better to use a mutex here, as this would allow us to copy a concrete time.Time struct. Writers will of course block readers, but the operation is fast enough that we probably don't care.

Update QUIC multiaddrs

Same as https://github.com/wetware/ww/issues/68, but this time in casm.Server().

Add listen addr to casm discover template