From 4d4125194b086175965b26d1ec9131ae161d29f7 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Mon, 14 Sep 2026 01:39:35 +0530 Subject: [PATCH 1/2] fix: relay gossip contacts through entrypoints Forward verified, signed ContactInfo records so nodes joining through a Mithril entrypoint can discover each other. Bound relay storage and packet sizes, preserve contact update ordering, remove withdrawn service endpoints, and continue sending after individual UDP failures. --- pkg/gossip/client.go | 46 ++++++++++---- pkg/gossip/contact_info.go | 3 +- pkg/gossip/contact_relay.go | 116 ++++++++++++++++++++++++++++++++++++ 3 files changed, 152 insertions(+), 13 deletions(-) create mode 100644 pkg/gossip/contact_relay.go diff --git a/pkg/gossip/client.go b/pkg/gossip/client.go index 67c205c64..3e822cb83 100644 --- a/pkg/gossip/client.go +++ b/pkg/gossip/client.go @@ -54,6 +54,7 @@ type Client struct { contactMu sync.RWMutex contact *ContactInfo + relay contactRelay identityConflictMu sync.Mutex identityConflictGossip contactEndpoint @@ -475,13 +476,23 @@ func (c *Client) pushContact(conn *net.UDPConn) error { if err != nil { return err } - for _, peer := range c.currentPeers() { - if err := sendUDP(conn, packet, peer); err != nil { - c.txErrors.Add(1) + packets := [][]byte{packet} + if values := c.relay.values(wallclockMillis()); len(values) != 0 { + packet, err := encodePushMessage(c.pubkey, values) + if err != nil { return err } - c.recordTx() - c.txPushMessages.Add(1) + packets = append(packets, packet) + } + for _, peer := range c.currentPeers() { + for _, packet := range packets { + if err := sendUDP(conn, packet, peer); err != nil { + c.txErrors.Add(1) + break + } + c.recordTx() + c.txPushMessages.Add(1) + } } return nil } @@ -626,6 +637,10 @@ func (c *Client) handleContactRecord(record contactRecord, shredVersion uint16) if record.ShredVer != shredVersion || !record.GossipAddr.ok { return } + addr := record.GossipAddr.UDPAddr() + if addr == nil || addr.Port == 0 || addr.IP.IsUnspecified() || addr.IP.IsMulticast() { + return + } // CRDS is keyed by validator identity. If another process publishes that // same identity with different sockets, its newer ContactInfo replaces ours // cluster-wide: turbine and Votor traffic then move to the other process in @@ -637,6 +652,11 @@ func (c *Client) handleContactRecord(record contactRecord, shredVersion uint16) c.observeOwnContactRecord(record) return } + // The decoder verified the original signed bytes. Retain those bytes for + // relay, and do not let an older relayed contact replace newer endpoints. + if !c.relay.accept(record, wallclockMillis()) { + return + } // The entrypoint is commonly a validator itself. Do not add its gossip // socket to the rotating pull-peer table, but retain the service endpoints // from its signed ContactInfo so Turbine, repair, and Votor can route to it. @@ -742,16 +762,17 @@ func (c *Client) recordRepairPeer(contact *ContactInfo) { } func (c *Client) recordRepairPeerRecord(record contactRecord) { + c.repairPeerMu.Lock() + defer c.repairPeerMu.Unlock() if !record.ServeRepairAddr.ok || record.ServeRepairAddr.port == 0 { + delete(c.repairPeers, record.Pubkey) return } now := time.Now() key := record.Pubkey - c.repairPeerMu.Lock() if existing, ok := c.repairPeers[key]; ok && sameEndpointUDPAddr(record.ServeRepairAddr, existing.Addr) { existing.LastSeen = now c.repairPeers[key] = existing - c.repairPeerMu.Unlock() return } c.repairPeers[key] = RepairPeer{ @@ -759,7 +780,6 @@ func (c *Client) recordRepairPeerRecord(record contactRecord) { Addr: record.ServeRepairAddr.UDPAddr(), LastSeen: now, } - c.repairPeerMu.Unlock() } // TVUPeers returns non-expired TVU endpoints learned from gossip. @@ -841,30 +861,32 @@ func (c *Client) LookupAlpenglow(pubkey solana.PublicKey) (*net.UDPAddr, bool) { func (c *Client) recordTVUPeerRecord(record contactRecord) { addr := record.TVUAddr.UDPAddr() + c.tvuPeerMu.Lock() + defer c.tvuPeerMu.Unlock() if addr == nil { + delete(c.tvuPeers, record.Pubkey) return } - c.tvuPeerMu.Lock() c.tvuPeers[record.Pubkey] = TVUPeer{ Pubkey: record.Pubkey, TVUAddr: addr, LastSeen: time.Now(), } - c.tvuPeerMu.Unlock() } func (c *Client) recordAlpenglowPeerRecord(record contactRecord) { addr := record.Sockets[socketTagAlpenglow].UDPAddr() + c.alpenglowPeerMu.Lock() + defer c.alpenglowPeerMu.Unlock() if addr == nil { + delete(c.alpenglowPeers, record.Pubkey) return } - c.alpenglowPeerMu.Lock() c.alpenglowPeers[record.Pubkey] = AlpenglowPeer{ Pubkey: record.Pubkey, AlpenglowAddr: addr, LastSeen: time.Now(), } - c.alpenglowPeerMu.Unlock() } func (c *Client) recordTx() { diff --git a/pkg/gossip/contact_info.go b/pkg/gossip/contact_info.go index baed0d396..b2adaf7c4 100644 --- a/pkg/gossip/contact_info.go +++ b/pkg/gossip/contact_info.go @@ -87,6 +87,7 @@ type contactEndpoint struct { type contactRecord struct { Pubkey Pubkey Wallclock uint64 + Outset uint64 ShredVer uint16 GossipAddr contactEndpoint ServeRepairAddr contactEndpoint @@ -370,7 +371,7 @@ func decodeContactRecord(d *decoder) (contactRecord, error) { if record.Wallclock, err = d.varint(10); err != nil { return contactRecord{}, err } - if _, err := d.u64(); err != nil { + if record.Outset, err = d.u64(); err != nil { return contactRecord{}, err } if record.ShredVer, err = d.u16(); err != nil { diff --git a/pkg/gossip/contact_relay.go b/pkg/gossip/contact_relay.go new file mode 100644 index 000000000..5ac9b56cd --- /dev/null +++ b/pkg/gossip/contact_relay.go @@ -0,0 +1,116 @@ +package gossip + +import ( + "bytes" + "crypto/sha256" + "sort" + "sync" + "time" +) + +// Agave accepts pushed values within a 15-second wallclock window. +const contactPushWindow = uint64(15 * time.Second / time.Millisecond) + +const crdsMessageHeaderSize = 4 + 32 + 8 + +type relayContact struct { + record contactRecord + hash [32]byte + forwarded uint64 +} + +// contactRelay retains at most the existing gossip peer limit. Each push tick +// relays at most one additional datagram per peer, rotating through contacts. +type contactRelay struct { + mu sync.Mutex + contacts map[Pubkey]relayContact + sequence uint64 +} + +func recentContact(wallclock, now, maxAge uint64) bool { + if wallclock > now { + return wallclock-now < contactPushWindow + } + return now-wallclock <= maxAge +} + +// accept is called only after signature and shred-version verification. +func (r *contactRelay) accept(record contactRecord, now uint64) bool { + if !recentContact(record.Wallclock, now, uint64(peerExpirationWindow/time.Millisecond)) { + return false + } + if len(record.data)+len(record.signature)+crdsMessageHeaderSize > packetDataSize { + return false + } + h := sha256.New() + h.Write(record.signature[:]) + h.Write(record.data) + var hash [32]byte + copy(hash[:], h.Sum(nil)) + r.mu.Lock() + defer r.mu.Unlock() + old, exists := r.contacts[record.Pubkey] + if exists { + // ContactInfo orders restarts first, then wallclock, then the hash of + // the complete signed value, matching Agave's CRDS replacement rule. + if record.Outset < old.record.Outset || + (record.Outset == old.record.Outset && record.Wallclock < old.record.Wallclock) || + (record.Outset == old.record.Outset && record.Wallclock == old.record.Wallclock && bytes.Compare(hash[:], old.hash[:]) <= 0) { + return false + } + } else if len(r.contacts) >= maxKnownGossipPeers { + var oldest Pubkey + oldestWallclock := ^uint64(0) + for key, contact := range r.contacts { + if contact.record.Wallclock < oldestWallclock { + oldest, oldestWallclock = key, contact.record.Wallclock + } + } + delete(r.contacts, oldest) + } + if r.contacts == nil { + r.contacts = make(map[Pubkey]relayContact) + } + // The receive loop reuses its datagram buffer on the next packet. + record.data = bytes.Clone(record.data) + r.contacts[record.Pubkey] = relayContact{record: record, hash: hash, forwarded: old.forwarded} + return true +} + +func (r *contactRelay) values(now uint64) []CrdsValue { + r.mu.Lock() + defer r.mu.Unlock() + var keys []Pubkey + for key, contact := range r.contacts { + if !recentContact(contact.record.Wallclock, now, uint64(peerExpirationWindow/time.Millisecond)) { + delete(r.contacts, key) + continue + } + if recentContact(contact.record.Wallclock, now, contactPushWindow) { + keys = append(keys, key) + } + } + sort.Slice(keys, func(i, j int) bool { + a, b := r.contacts[keys[i]], r.contacts[keys[j]] + if a.forwarded != b.forwarded { + return a.forwarded < b.forwarded + } + return bytes.Compare(keys[i][:], keys[j][:]) < 0 + }) + var values []CrdsValue + size := crdsMessageHeaderSize + for _, key := range keys { + contact := r.contacts[key] + record := contact.record + n := len(record.signature) + len(record.data) + if size+n > packetDataSize { + break + } + values = append(values, CrdsValue{Signature: record.signature, Data: record.data}) + size += n + r.sequence++ + contact.forwarded = r.sequence + r.contacts[key] = contact + } + return values +} From 00b966c9967629220272a21550121bda5991efe9 Mon Sep 17 00:00:00 2001 From: Neeraj Godiyal Date: Mon, 14 Sep 2026 01:39:35 +0530 Subject: [PATCH 2/2] test: cover gossip peer discovery and contact relay Cover signed-record preservation, stale and invalid contacts, relay limits, endpoint withdrawal, peer restart, and UDP send failures. Add packet fuzz coverage and run networking race tests in CI. --- .github/workflows/go_build.yml | 3 + pkg/gossip/contact_relay_test.go | 361 +++++++++++++++++++++++++++++++ 2 files changed, 364 insertions(+) create mode 100644 pkg/gossip/contact_relay_test.go diff --git a/.github/workflows/go_build.yml b/.github/workflows/go_build.yml index 50383467f..eb5088e1e 100644 --- a/.github/workflows/go_build.yml +++ b/.github/workflows/go_build.yml @@ -17,3 +17,6 @@ jobs: - name: Build run: go build -v ./cmd/mithril + + - name: Test gossip and networking + run: go test -race ./pkg/gossip ./pkg/repair ./pkg/turbine ./pkg/blockstream diff --git a/pkg/gossip/contact_relay_test.go b/pkg/gossip/contact_relay_test.go new file mode 100644 index 000000000..762475cfc --- /dev/null +++ b/pkg/gossip/contact_relay_test.go @@ -0,0 +1,361 @@ +package gossip + +import ( + "bytes" + "context" + "crypto/ed25519" + "crypto/rand" + "net" + "sync" + "testing" + "time" + + "github.com/gagliardetto/solana-go" +) + +func relayFixture(t testing.TB) (*ContactInfo, ed25519.PrivateKey) { + t.Helper() + pub, key, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + info, err := NewContactInfo(Pubkey(pub), 4321, + &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 9000}, + &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 9001}) + if err != nil { + t.Fatal(err) + } + if err := info.SetSocket(socketTagServeRepair, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 9002}); err != nil { + t.Fatal(err) + } + if err := info.SetAlpenglowAddr(&net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 9003}); err != nil { + t.Fatal(err) + } + if err := info.SetTPUQUIC(&net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 9004}); err != nil { + t.Fatal(err) + } + return info, key +} + +func TestContactRelayPreservesSignedBytesAndOrdersUpdates(t *testing.T) { + info, key := relayFixture(t) + now := info.Wallclock + var relay contactRelay + record := contactRecordFromInfo(t, info, key) + original := bytes.Clone(record.data) + if !relay.accept(record, now) { + t.Fatal("first contact rejected") + } + clear(record.data) + values := relay.values(now) + if len(values) != 1 || !bytes.Equal(values[0].Data, original) { + t.Fatal("relay did not retain the original receive bytes") + } + packet, err := encodePushMessage(Pubkey{}, values) + if err != nil { + t.Fatal(err) + } + var forwarded contactRecord + decoded, err := decodePacketWithContactHandler(packet, func(record contactRecord) { forwarded = record }) + if err != nil || decoded.ContactCount != 1 { + t.Fatalf("relayed signature/decoding: %v, %+v", err, decoded) + } + if endpointString(forwarded.ServeRepairAddr) != "127.0.0.1:9002" || endpointString(forwarded.Sockets[socketTagAlpenglow]) != "127.0.0.1:9003" || endpointString(forwarded.Sockets[socketTagTPUQUIC]) != "127.0.0.1:9004" { + t.Fatalf("lost relayed service endpoints: %+v", forwarded) + } + if relay.accept(contactRecordFromInfo(t, info, key), now) { + t.Fatal("duplicate accepted as update") + } + if relay.accept(contactRecordFromInfo(t, info.CloneWithWallclock(now-1), key), now) { + t.Fatal("older wallclock accepted") + } + newer := info.CloneWithWallclock(now + 1) + if !relay.accept(contactRecordFromInfo(t, newer, key), now) { + t.Fatal("newer wallclock rejected") + } + restarted := info.CloneWithWallclock(now - 1) + restarted.Outset++ + if !relay.accept(contactRecordFromInfo(t, restarted, key), now) { + t.Fatal("newer process outset rejected") + } + if relay.accept(contactRecordFromInfo(t, newer.CloneWithWallclock(now+2), key), now) { + t.Fatal("old process overwrote restarted process") + } + if got := relay.values(now + contactPushWindow + 1); len(got) != 0 { + t.Fatal("expired push value relayed") + } + if got := relay.values(now + uint64(peerExpirationWindow/time.Millisecond) + 1); len(got) != 0 || len(relay.contacts) != 0 { + t.Fatal("expired cache entry retained") + } +} + +func TestContactRelayDeterministicTies(t *testing.T) { + info, key := relayFixture(t) + a := contactRecordFromInfo(t, info, key) + if err := info.SetSocket(socketTagTVU, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 9101}); err != nil { + t.Fatal(err) + } + b := contactRecordFromInfo(t, info, key) + var left, right contactRelay + left.accept(a, info.Wallclock) + left.accept(b, info.Wallclock) + right.accept(b, info.Wallclock) + right.accept(a, info.Wallclock) + if !bytes.Equal(left.values(info.Wallclock)[0].Data, right.values(info.Wallclock)[0].Data) { + t.Fatal("equal timestamps did not converge") + } +} + +func TestContactRelayRejectsInvalidContacts(t *testing.T) { + for _, name := range []string{"signature", "wrong cluster", "expired", "future", "zero port", "multicast", "unspecified"} { + t.Run(name, func(t *testing.T) { + client, err := NewClient(Config{Entrypoint: "127.0.0.1:8000", ShredVersion: 4321}) + if err != nil { + t.Fatal(err) + } + info, key := relayFixture(t) + switch name { + case "wrong cluster": + info.ShredVer++ + case "expired": + info.Wallclock -= uint64(peerExpirationWindow/time.Millisecond) + 1000 + case "future": + info.Wallclock += contactPushWindow + 1000 + case "zero port": + info.Sockets[0].Port = 0 + case "multicast": + info.Addrs[0] = net.ParseIP("224.0.0.1") + case "unspecified": + info.Addrs[0] = net.IPv4zero + } + value, err := signCrdsContactInfo(info, key) + if err != nil { + t.Fatal(err) + } + if name == "signature" { + value.Signature[0] ^= 1 + } + packet, err := encodePushMessage(info.Pubkey, []CrdsValue{value}) + if err != nil { + t.Fatal(err) + } + if err := client.handlePacket(nil, packet, &net.UDPAddr{}); err != nil { + t.Fatal(err) + } + if len(client.relay.contacts) != 0 || len(client.TVUPeers()) != 0 || client.Stats().AcceptedContacts != 0 { + t.Fatal("invalid contact entered relay or routing tables") + } + }) + } +} + +func TestContactRelayPacketAndStorageBounds(t *testing.T) { + var relay contactRelay + now := wallclockMillis() + for i := 0; i < maxKnownGossipPeers+10; i++ { + info, key := relayFixture(t) + if !relay.accept(contactRecordFromInfo(t, info, key), now) { + t.Fatal("valid contact rejected") + } + if len(relay.contacts) > maxKnownGossipPeers { + t.Fatal("relay exceeded capacity") + } + } + seen := make(map[Pubkey]bool) + for i := 0; i < maxKnownGossipPeers; i++ { + packet, err := encodePushMessage(Pubkey{}, relay.values(now)) + if err != nil { + t.Fatal(err) + } + if len(packet) > packetDataSize { + t.Fatal("relay packet exceeded wire limit") + } + decoded, err := decodePacket(packet) + if err != nil { + t.Fatal(err) + } + for _, contact := range decoded.Contacts { + seen[contact.Pubkey] = true + } + if len(seen) == maxKnownGossipPeers { + break + } + } + if len(seen) != maxKnownGossipPeers { + t.Fatalf("rotation reached %d of %d contacts", len(seen), maxKnownGossipPeers) + } +} + +func TestContactUpdateRemovesWithdrawnEndpoints(t *testing.T) { + client, err := NewClient(Config{Entrypoint: "127.0.0.1:8000", ShredVersion: 4321}) + if err != nil { + t.Fatal(err) + } + info, key := relayFixture(t) + client.handleContactRecord(contactRecordFromInfo(t, info, key), 4321) + if len(client.RepairPeers()) != 1 || len(client.AlpenglowPeers()) != 1 { + t.Fatal("initial endpoints missing") + } + updated, err := NewContactInfo(info.Pubkey, 4321, info.GossipAddr, nil) + if err != nil { + t.Fatal(err) + } + updated.Outset = info.Outset + 1 + client.handleContactRecord(contactRecordFromInfo(t, updated, key), 4321) + if len(client.RepairPeers()) != 0 || len(client.AlpenglowPeers()) != 0 || len(client.TVUPeers()) != 0 { + t.Fatal("withdrawn endpoints retained") + } + client.handleContactRecord(contactRecordFromInfo(t, info, key), 4321) + if len(client.RepairPeers()) != 0 { + t.Fatal("old relayed contact restored withdrawn endpoint") + } +} + +func waitForGossip(t *testing.T, label string, check func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if check() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("timed out: " + label) +} + +func runRelayClient(t *testing.T, seed string, key ed25519.PrivateKey) (*Client, func()) { + t.Helper() + socket, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + addr := socket.LocalAddr().String() + if err := socket.Close(); err != nil { + t.Fatal(err) + } + if seed == "" { + seed = addr + } + client, err := NewClient(Config{Entrypoint: seed, BindAddr: addr, TVUAddr: addr, AlpenglowAddr: addr, + AdvertisedIP: "127.0.0.1", ShredVersion: 4321, Identity: key, PushInterval: 50 * time.Millisecond, PingInterval: 50 * time.Millisecond}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- client.Run(ctx) }() + var once sync.Once + stop := func() { + once.Do(func() { + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("gossip Run: %v", err) + } + case <-time.After(time.Second): + t.Error("gossip did not stop") + } + }) + } + t.Cleanup(stop) + waitForGossip(t, "client startup", func() bool { return client.AdvertisedGossipAddr() != nil }) + return client, stop +} + +func TestGossipDiscoverySurvivesSeedExitAndPeerRestart(t *testing.T) { + seed, stopSeed := runRelayClient(t, "", nil) + left, stopLeft := runRelayClient(t, seed.AdvertisedGossipAddr().String(), nil) + right, _ := runRelayClient(t, seed.AdvertisedGossipAddr().String(), nil) + leftKey, rightKey := solana.PublicKey(left.Pubkey()), solana.PublicKey(right.Pubkey()) + waitForGossip(t, "joiners discover each other through seed", func() bool { + _, a := left.LookupAlpenglow(rightKey) + _, b := right.LookupAlpenglow(leftKey) + return a && b + }) + stopSeed() + time.Sleep(100 * time.Millisecond) + lrx, rrx := left.Stats().RxPackets, right.Stats().RxPackets + waitForGossip(t, "survivors exchange new packets", func() bool { return left.Stats().RxPackets > lrx && right.Stats().RxPackets > rrx }) + identity := left.Identity() + stopLeft() + restarted, _ := runRelayClient(t, right.AdvertisedGossipAddr().String(), identity) + waitForGossip(t, "restart replaces old endpoint", func() bool { + addr, ok := right.LookupAlpenglow(leftKey) + return ok && addr.String() == restarted.AdvertisedGossipAddr().String() + }) +} + +func TestGossipPushContinuesAfterUnreachablePeer(t *testing.T) { + sender, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + defer sender.Close() + receiver, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + defer receiver.Close() + client, err := NewClient(Config{Entrypoint: receiver.LocalAddr().String(), TVUAddr: sender.LocalAddr().String(), AdvertisedIP: "127.0.0.1", ShredVersion: 4321}) + if err != nil { + t.Fatal(err) + } + if err := client.initializeContact(sender.LocalAddr().(*net.UDPAddr)); err != nil { + t.Fatal(err) + } + client.recordPeer(receiver.LocalAddr().(*net.UDPAddr)) + // An IPv4-only socket cannot send to this IPv6 destination. + client.recordPeer(&net.UDPAddr{IP: net.ParseIP("::1"), Port: 9000}) + if err := client.pushContact(sender); err != nil { + t.Fatal(err) + } + if err := receiver.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + var packet [packetDataSize]byte + if _, _, err := receiver.ReadFromUDP(packet[:]); err != nil { + t.Fatal(err) + } + if client.Stats().TxErrors != 1 || client.Stats().TxPushMessages != 1 { + t.Fatalf("unexpected send counters: %+v", client.Stats()) + } +} + +func FuzzContactRelayPackets(f *testing.F) { + info, key := relayFixture(f) + _, localKey, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + f.Fatal(err) + } + value, err := signCrdsContactInfo(info, key) + if err != nil { + f.Fatal(err) + } + packet, err := encodePushMessage(info.Pubkey, []CrdsValue{value}) + if err != nil { + f.Fatal(err) + } + f.Add(packet) + f.Add([]byte{0, 0, 0, 0}) + f.Fuzz(func(t *testing.T, packet []byte) { + if len(packet) > packetDataSize { + return + } + client, err := NewClient(Config{Entrypoint: "127.0.0.1:8000", Identity: localKey, ShredVersion: 4321}) + if err != nil { + t.Fatal(err) + } + // Decode contact messages only; this check must never send a packet. + _, _ = decodePacketWithContactHandler(packet, func(record contactRecord) { client.handleContactRecord(record, 4321) }) + values := client.relay.values(info.Wallclock) + encoded, err := encodePushMessage(client.Pubkey(), values) + if err != nil { + t.Fatal(err) + } + decoded, err := decodePacket(encoded) + if err != nil || decoded.ContactCount != len(values) { + t.Fatalf("relay produced invalid signed values: %v", err) + } + }) +}