Skip to content

Commit a5aafd1

Browse files
authored
fix(clmimicry): migrate to go-libp2p-pubsub v0.16.1 (#852)
v0.16 removed the ADD_PEER / REMOVE_PEER gossipsub tracer events (replaced by per-stream ON_NEW_OUTBOUND_STREAM / ON_CLOSED_OUTBOUND_STREAM). Bump the dependency and remove the now-dead producer-side handling for those two events; peer connect/disconnect remains covered by the host notifee (CONNECTED / DISCONNECTED). The xatu proto event types are kept for schema / historical-data backward compatibility.
1 parent 956c4ae commit a5aafd1

11 files changed

Lines changed: 23 additions & 468 deletions

go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ require (
3535
github.com/klauspost/compress v1.18.5
3636
github.com/lib/pq v1.10.9
3737
github.com/libp2p/go-libp2p v0.47.0
38-
github.com/libp2p/go-libp2p-pubsub v0.14.2
38+
github.com/libp2p/go-libp2p-pubsub v0.16.1-0.20260611143718-41b11d5cb1a7
3939
github.com/minio/minio-go/v7 v7.1.0
4040
github.com/mitchellh/hashstructure/v2 v2.0.2
4141
github.com/multiformats/go-multiaddr v0.16.1

go.sum

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -472,8 +472,8 @@ github.com/libp2p/go-libp2p-asn-util v0.4.1 h1:xqL7++IKD9TBFMgnLPZR6/6iYhawHKHl9
472472
github.com/libp2p/go-libp2p-asn-util v0.4.1/go.mod h1:d/NI6XZ9qxw67b4e+NgpQexCIiFYJjErASrYW4PFDN8=
473473
github.com/libp2p/go-libp2p-mplex v0.11.0 h1:0vwpLXRSfkTzshEjETIEgJaVxXvg+orbxYoIb3Ty5qM=
474474
github.com/libp2p/go-libp2p-mplex v0.11.0/go.mod h1:QrsdNY3lzjpdo9V1goJfPb0O65Nms0sUR8CDAO18f6k=
475-
github.com/libp2p/go-libp2p-pubsub v0.14.2 h1:nT5lFHPQOFJcp9CW8hpKtvbpQNdl2udJuzLQWbgRum8=
476-
github.com/libp2p/go-libp2p-pubsub v0.14.2/go.mod h1:MKPU5vMI8RRFyTP0HfdsF9cLmL1nHAeJm44AxJGJx44=
475+
github.com/libp2p/go-libp2p-pubsub v0.16.1-0.20260611143718-41b11d5cb1a7 h1:UMiJ408NqO9Sf2ANutEM3An8Em3K+qn78eoIgzY3PIY=
476+
github.com/libp2p/go-libp2p-pubsub v0.16.1-0.20260611143718-41b11d5cb1a7/go.mod h1:l00Tc/MXTM/dK69HFxKWIe0yaNwMy5OuU0dyuZaWJa0=
477477
github.com/libp2p/go-libp2p-testing v0.12.0 h1:EPvBb4kKMWO29qP4mZGyhVzUyR25dvfUIK5WDu6iPUA=
478478
github.com/libp2p/go-libp2p-testing v0.12.0/go.mod h1:KcGDRXyN7sQCllucn1cOOS+Dmm7ujhfEyXQL5lvkcPg=
479479
github.com/libp2p/go-mplex v0.7.0 h1:BDhFZdlk5tbr0oyFq/xv/NPGfjbnrsDam1EvutpBDbY=
@@ -492,8 +492,8 @@ github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35 h1:PpXWgLPs+Fqr32
492492
github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35/go.mod h1:autxFIvghDt3jPTLoqZ9OZ7s9qTGNAWmYCjVFWPX/zg=
493493
github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE=
494494
github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
495-
github.com/marcopolo/simnet v0.0.4 h1:50Kx4hS9kFGSRIbrt9xUS3NJX33EyPqHVmpXvaKLqrY=
496-
github.com/marcopolo/simnet v0.0.4/go.mod h1:tfQF1u2DmaB6WHODMtQaLtClEf3a296CKQLq5gAsIS0=
495+
github.com/marcopolo/simnet v0.0.7 h1:DpH8BMGsF9+1w13L8rvCaAhb6nYJdY+dIXncDrssvUs=
496+
github.com/marcopolo/simnet v0.0.7/go.mod h1:tfQF1u2DmaB6WHODMtQaLtClEf3a296CKQLq5gAsIS0=
497497
github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd h1:br0buuQ854V8u83wA0rVZ8ttrq5CpaPZdvrK0LP2lOk=
498498
github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd/go.mod h1:QuCEs1Nt24+FYQEqAAncTDPJIuGs+LxK1MCiFL25pMU=
499499
github.com/matoous/go-nanoid/v2 v2.1.0 h1:P64+dmq21hhWdtvZfEAofnvJULaRR1Yib0+PnU669bE=

pkg/clmimicry/README.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ Events with only message ID:
8585

8686
### Group D: No Sharding Key Events
8787
Events without sharding keys:
88-
- `ADD_PEER`, `REMOVE_PEER`, `CONNECTED`, `DISCONNECTED`
88+
- `CONNECTED`, `DISCONNECTED`
8989
- `RECV_RPC`, `SEND_RPC`, `DROP_RPC` (parent events only)
9090
- `HANDLE_METADATA`, `HANDLE_STATUS`
9191

pkg/clmimicry/event.go

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,8 +46,6 @@ const (
4646
// - Blob Sidecars (p2p.GossipBlobSidecarMessage)
4747
//
4848
// 2. libp2p pubsub protocol level events:
49-
// - "ADD_PEER": When a peer is added to the pubsub system
50-
// - "REMOVE_PEER": When a peer is removed from the pubsub system
5149
// - "RECV_RPC": When an RPC message is received
5250
// - "SEND_RPC": When an RPC message is sent
5351
// - "JOIN": When joining a pubsub topic

pkg/clmimicry/event_categorizer_test.go

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@ func TestEventCategorization(t *testing.T) {
2020
t.Logf("Group %d has %d events", group, len(events))
2121
}
2222

23-
// We should have categorized all 29 known events
24-
assert.Equal(t, totalEvents, 29, "Should have exactly 28 events categorized")
23+
// We should have categorized all 27 currently emitted events
24+
assert.Equal(t, 27, totalEvents, "Should have exactly 27 events categorized")
2525

2626
// Test specific group queries
2727
groupA := ec.GetGroupAEvents()
@@ -34,7 +34,7 @@ func TestEventCategorization(t *testing.T) {
3434
assert.Len(t, groupC, 2, "Group C should have 2 events")
3535

3636
groupD := ec.GetGroupDEvents()
37-
assert.GreaterOrEqual(t, len(groupD), 10, "Group D should have at least 10 events")
37+
assert.Len(t, groupD, 8, "Group D should have 8 events")
3838
}
3939

4040
func TestMetaEventIdentification(t *testing.T) {
@@ -53,7 +53,6 @@ func TestMetaEventIdentification(t *testing.T) {
5353
nonMetaEvents := []xatu.Event_Name{
5454
xatu.Event_LIBP2P_TRACE_PUBLISH_MESSAGE,
5555
xatu.Event_LIBP2P_TRACE_JOIN,
56-
xatu.Event_LIBP2P_TRACE_ADD_PEER,
5756
xatu.Event_LIBP2P_TRACE_CONNECTED,
5857
}
5958

@@ -113,8 +112,8 @@ func TestEventGroupCharacteristics(t *testing.T) {
113112
{
114113
name: "Group D events have neither",
115114
events: []xatu.Event_Name{
116-
xatu.Event_LIBP2P_TRACE_ADD_PEER,
117115
xatu.Event_LIBP2P_TRACE_CONNECTED,
116+
xatu.Event_LIBP2P_TRACE_DISCONNECTED,
118117
xatu.Event_LIBP2P_TRACE_RECV_RPC,
119118
},
120119
expectTopic: false,
@@ -164,8 +163,6 @@ func TestEventCompleteness(t *testing.T) {
164163
xatu.Event_LIBP2P_TRACE_RPC_META_CONTROL_IWANT,
165164
xatu.Event_LIBP2P_TRACE_RPC_META_CONTROL_IDONTWANT,
166165
// Group D
167-
xatu.Event_LIBP2P_TRACE_ADD_PEER,
168-
xatu.Event_LIBP2P_TRACE_REMOVE_PEER,
169166
xatu.Event_LIBP2P_TRACE_RECV_RPC,
170167
xatu.Event_LIBP2P_TRACE_SEND_RPC,
171168
xatu.Event_LIBP2P_TRACE_DROP_RPC,

pkg/clmimicry/event_catergorizer.go

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -70,8 +70,6 @@ func (ec *EventCategorizer) initializeEvents() {
7070
ec.addEvent(xatu.Event_LIBP2P_TRACE_RPC_META_CONTROL_IDONTWANT, GroupC, false, true, true)
7171

7272
// Group D: Events with no sharding keys
73-
ec.addEvent(xatu.Event_LIBP2P_TRACE_ADD_PEER, GroupD, false, false, false)
74-
ec.addEvent(xatu.Event_LIBP2P_TRACE_REMOVE_PEER, GroupD, false, false, false)
7573
ec.addEvent(xatu.Event_LIBP2P_TRACE_RECV_RPC, GroupD, false, false, false)
7674
ec.addEvent(xatu.Event_LIBP2P_TRACE_SEND_RPC, GroupD, false, false, false)
7775
ec.addEvent(xatu.Event_LIBP2P_TRACE_DROP_RPC, GroupD, false, false, false)

pkg/clmimicry/event_config.go

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,6 @@ type EventConfig struct {
1212
RpcMetaControlPruneEnabled bool `yaml:"rpcMetaControlPruneEnabled" default:"false"`
1313
RpcMetaSubscriptionEnabled bool `yaml:"rpcMetaSubscriptionEnabled" default:"false"`
1414
RpcMetaMessageEnabled bool `yaml:"rpcMetaMessageEnabled" default:"false"`
15-
AddPeerEnabled bool `yaml:"addPeerEnabled" default:"true"`
16-
RemovePeerEnabled bool `yaml:"removePeerEnabled" default:"true"`
1715
ConnectedEnabled bool `yaml:"connectedEnabled" default:"true"`
1816
DisconnectedEnabled bool `yaml:"disconnectedEnabled" default:"true"`
1917
IdentifyEnabled bool `yaml:"identifyEnabled" default:"true"`

pkg/clmimicry/event_libp2p.go

Lines changed: 11 additions & 117 deletions
Original file line numberDiff line numberDiff line change
@@ -22,15 +22,17 @@ var libp2pToXatuEventMap = map[string]string{
2222
pubsubpb.TraceEvent_REJECT_MESSAGE.String(): xatu.Event_LIBP2P_TRACE_REJECT_MESSAGE.String(),
2323
pubsubpb.TraceEvent_DUPLICATE_MESSAGE.String(): xatu.Event_LIBP2P_TRACE_DUPLICATE_MESSAGE.String(),
2424
pubsubpb.TraceEvent_DELIVER_MESSAGE.String(): xatu.Event_LIBP2P_TRACE_DELIVER_MESSAGE.String(),
25-
pubsubpb.TraceEvent_ADD_PEER.String(): xatu.Event_LIBP2P_TRACE_ADD_PEER.String(),
26-
pubsubpb.TraceEvent_REMOVE_PEER.String(): xatu.Event_LIBP2P_TRACE_REMOVE_PEER.String(),
27-
pubsubpb.TraceEvent_RECV_RPC.String(): xatu.Event_LIBP2P_TRACE_RECV_RPC.String(),
28-
pubsubpb.TraceEvent_SEND_RPC.String(): xatu.Event_LIBP2P_TRACE_SEND_RPC.String(),
29-
pubsubpb.TraceEvent_DROP_RPC.String(): xatu.Event_LIBP2P_TRACE_DROP_RPC.String(),
30-
pubsubpb.TraceEvent_JOIN.String(): xatu.Event_LIBP2P_TRACE_JOIN.String(),
31-
pubsubpb.TraceEvent_LEAVE.String(): xatu.Event_LIBP2P_TRACE_LEAVE.String(),
32-
pubsubpb.TraceEvent_GRAFT.String(): xatu.Event_LIBP2P_TRACE_GRAFT.String(),
33-
pubsubpb.TraceEvent_PRUNE.String(): xatu.Event_LIBP2P_TRACE_PRUNE.String(),
25+
// ADD_PEER / REMOVE_PEER were removed from the gossipsub tracer in go-libp2p-pubsub v0.16
26+
// (replaced by per-stream ON_NEW_OUTBOUND_STREAM / ON_CLOSED_OUTBOUND_STREAM events), so they
27+
// are no longer mapped here. Peer connect/disconnect remains available via the host notifee
28+
// (CONNECTED / DISCONNECTED).
29+
pubsubpb.TraceEvent_RECV_RPC.String(): xatu.Event_LIBP2P_TRACE_RECV_RPC.String(),
30+
pubsubpb.TraceEvent_SEND_RPC.String(): xatu.Event_LIBP2P_TRACE_SEND_RPC.String(),
31+
pubsubpb.TraceEvent_DROP_RPC.String(): xatu.Event_LIBP2P_TRACE_DROP_RPC.String(),
32+
pubsubpb.TraceEvent_JOIN.String(): xatu.Event_LIBP2P_TRACE_JOIN.String(),
33+
pubsubpb.TraceEvent_LEAVE.String(): xatu.Event_LIBP2P_TRACE_LEAVE.String(),
34+
pubsubpb.TraceEvent_GRAFT.String(): xatu.Event_LIBP2P_TRACE_GRAFT.String(),
35+
pubsubpb.TraceEvent_PRUNE.String(): xatu.Event_LIBP2P_TRACE_PRUNE.String(),
3436
}
3537

3638
// handleHermesLibp2pEvent handles libp2p pubsub protocol level events.
@@ -54,21 +56,6 @@ func (p *Processor) handleHermesLibp2pEvent(
5456
networkStr := getNetworkID(clientMeta)
5557

5658
switch xatuEvent {
57-
case xatu.Event_LIBP2P_TRACE_ADD_PEER.String():
58-
if !p.events.AddPeerEnabled {
59-
return nil
60-
}
61-
62-
// Record that we received this event
63-
p.metrics.AddEvent(xatuEvent, networkStr)
64-
65-
// Check if we should process this event based on trace/sharding config.
66-
if !p.ShouldTraceMessage(event, clientMeta, xatuEvent) {
67-
return nil
68-
}
69-
70-
return p.handleAddPeerEvent(ctx, clientMeta, traceMeta, event)
71-
7259
case xatu.Event_LIBP2P_TRACE_RECV_RPC.String():
7360
// Always process RPC events to extract child events, even if parent is disabled
7461
// This allows child events (like IHAVE) to be captured independently
@@ -84,21 +71,6 @@ func (p *Processor) handleHermesLibp2pEvent(
8471
// This allows child events (like IHAVE) to be captured independently
8572
return p.handleSendRPCEvent(ctx, clientMeta, traceMeta, event, xatuEvent, networkStr)
8673

87-
case xatu.Event_LIBP2P_TRACE_REMOVE_PEER.String():
88-
if !p.events.RemovePeerEnabled {
89-
return nil
90-
}
91-
92-
// Record that we received this event
93-
p.metrics.AddEvent(xatuEvent, networkStr)
94-
95-
// Check if we should process this event based on trace/sharding config.
96-
if !p.ShouldTraceMessage(event, clientMeta, xatuEvent) {
97-
return nil
98-
}
99-
100-
return p.handleRemovePeerEvent(ctx, clientMeta, traceMeta, event)
101-
10274
case xatu.Event_LIBP2P_TRACE_JOIN.String():
10375
if !p.events.JoinEnabled {
10476
return nil
@@ -216,45 +188,6 @@ func (p *Processor) handleHermesLibp2pEvent(
216188
return nil
217189
}
218190

219-
func (p *Processor) handleRemovePeerEvent(
220-
ctx context.Context,
221-
clientMeta *xatu.ClientMeta,
222-
traceMeta *libp2p.TraceEventMetadata,
223-
event *TraceEvent,
224-
) error {
225-
data, err := TraceEventToRemovePeer(event)
226-
if err != nil {
227-
return errors.Wrapf(err, "failed to convert event to remove peer event")
228-
}
229-
230-
metadata, ok := proto.Clone(clientMeta).(*xatu.ClientMeta)
231-
if !ok {
232-
return fmt.Errorf("failed to clone client metadata")
233-
}
234-
235-
metadata.AdditionalData = &xatu.ClientMeta_Libp2PTraceRemovePeer{
236-
Libp2PTraceRemovePeer: &xatu.ClientMeta_AdditionalLibP2PTraceRemovePeerData{
237-
Metadata: traceMeta,
238-
},
239-
}
240-
241-
decoratedEvent := &xatu.DecoratedEvent{
242-
Event: &xatu.Event{
243-
Name: xatu.Event_LIBP2P_TRACE_REMOVE_PEER,
244-
DateTime: timestamppb.New(event.Timestamp.Add(p.clockDrift)),
245-
Id: uuid.New().String(),
246-
},
247-
Meta: &xatu.Meta{
248-
Client: metadata,
249-
},
250-
Data: &xatu.DecoratedEvent_Libp2PTraceRemovePeer{
251-
Libp2PTraceRemovePeer: data,
252-
},
253-
}
254-
255-
return p.output.HandleDecoratedEvent(ctx, decoratedEvent)
256-
}
257-
258191
func (p *Processor) handleJoinEvent(
259192
ctx context.Context,
260193
clientMeta *xatu.ClientMeta,
@@ -500,45 +433,6 @@ func (p *Processor) handleSendRPCEvent(
500433
return nil
501434
}
502435

503-
func (p *Processor) handleAddPeerEvent(
504-
ctx context.Context,
505-
clientMeta *xatu.ClientMeta,
506-
traceMeta *libp2p.TraceEventMetadata,
507-
event *TraceEvent,
508-
) error {
509-
data, err := TraceEventToAddPeer(event)
510-
if err != nil {
511-
return errors.Wrapf(err, "failed to convert event to add_peer event")
512-
}
513-
514-
metadata, ok := proto.Clone(clientMeta).(*xatu.ClientMeta)
515-
if !ok {
516-
return fmt.Errorf("failed to clone client metadata")
517-
}
518-
519-
metadata.AdditionalData = &xatu.ClientMeta_Libp2PTraceAddPeer{
520-
Libp2PTraceAddPeer: &xatu.ClientMeta_AdditionalLibP2PTraceAddPeerData{
521-
Metadata: traceMeta,
522-
},
523-
}
524-
525-
decoratedEvent := &xatu.DecoratedEvent{
526-
Event: &xatu.Event{
527-
Name: xatu.Event_LIBP2P_TRACE_ADD_PEER,
528-
DateTime: timestamppb.New(event.Timestamp.Add(p.clockDrift)),
529-
Id: uuid.New().String(),
530-
},
531-
Meta: &xatu.Meta{
532-
Client: metadata,
533-
},
534-
Data: &xatu.DecoratedEvent_Libp2PTraceAddPeer{
535-
Libp2PTraceAddPeer: data,
536-
},
537-
}
538-
539-
return p.output.HandleDecoratedEvent(ctx, decoratedEvent)
540-
}
541-
542436
func (p *Processor) handleRecvRPCEvent(
543437
ctx context.Context,
544438
clientMeta *xatu.ClientMeta,

0 commit comments

Comments
 (0)