From 730027f11c684d913f752905048b00bee7095dd7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jano=C5=A1=20Gulja=C5=A1?= Date: Sat, 25 Jul 2026 22:14:35 +0200 Subject: [PATCH 1/2] feat(p2p): add versioned handlers function --- pkg/p2p/example_test.go | 148 ++++++++++++++++++++++++++++++++++++++++ pkg/p2p/p2p.go | 35 ++++++++++ pkg/p2p/p2p_test.go | 147 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 330 insertions(+) create mode 100644 pkg/p2p/example_test.go diff --git a/pkg/p2p/example_test.go b/pkg/p2p/example_test.go new file mode 100644 index 00000000000..9f8cf845a8b --- /dev/null +++ b/pkg/p2p/example_test.go @@ -0,0 +1,148 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package p2p_test + +import ( + "context" + "fmt" + + "github.com/coreos/go-semver/semver" + "github.com/ethersphere/bee/v2/pkg/p2p" + "github.com/ethersphere/bee/v2/pkg/p2p/protobuf" + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +const ( + exampleProtocolName = "versionedping" + exampleProtocolVersion = "1.2.0" + exampleStreamName = "ping" +) + +// ExampleService represents a full Bee protocol service (structured like pkg/pingpong) +// supporting 3 version levels: +// - v1.2.0: Current version +// - v1.1.0: Legacy version 1.1 +// - v1.0.0: Legacy version 1.0 +type ExampleService struct { + streamer p2p.Streamer +} + +func NewExampleService(streamer p2p.Streamer) *ExampleService { + return &ExampleService{ + streamer: streamer, + } +} + +func (s *ExampleService) Protocol() p2p.ProtocolSpec { + return p2p.ProtocolSpec{ + Name: exampleProtocolName, + Version: exampleProtocolVersion, + StreamSpecs: []p2p.StreamSpec{ + { + Name: exampleStreamName, + Handler: p2p.NewVersionedHandlersFunc( + p2p.VersionedHandler{ + Version: semver.New("1.2.0"), // Server handler for >= 1.2.0 + Handler: func(ctx context.Context, p p2p.Peer, stream p2p.Stream) error { + w, r := protobuf.NewWriterAndReader(stream) + _, _ = w, r + fmt.Println("Server received ping on v1.2.0 handler") + return stream.FullClose() + }, + }, + p2p.VersionedHandler{ + Version: semver.New("1.1.0"), // Server handler for legacy 1.1.0 + Handler: func(ctx context.Context, p p2p.Peer, stream p2p.Stream) error { + w, r := protobuf.NewWriterAndReader(stream) + _, _ = w, r + fmt.Println("Server received ping on v1.1.0 legacy handler") + return stream.FullClose() + }, + }, + p2p.VersionedHandler{ + Version: semver.New("1.0.0"), // Server handler for legacy 1.0.0 + Handler: func(ctx context.Context, p p2p.Peer, stream p2p.Stream) error { + w, r := protobuf.NewWriterAndReader(stream) + _, _ = w, r + fmt.Println("Server received ping on v1.0.0 legacy handler") + return stream.FullClose() + }, + }, + ), + }, + }, + } +} + +func (s *ExampleService) Ping(ctx context.Context, peer swarm.Address) error { + stream, err := s.streamer.NewStream(ctx, peer, nil, exampleProtocolName, "1.1.0", exampleStreamName) + if err != nil { + return err + } + defer stream.Close() + + pingClient := p2p.NewVersionedHandlersFunc( + p2p.VersionedHandler{ + Version: semver.New("1.2.0"), // Current version client (>= 1.2.0) + Handler: func(ctx context.Context, _ p2p.Peer, stream p2p.Stream) error { + w, r := protobuf.NewWriterAndReader(stream) + _, _ = w, r + fmt.Println("Client sent ping using v1.2.0 format") + return nil + }, + }, + p2p.VersionedHandler{ + Version: semver.New("1.1.0"), // Legacy v1.1.0 client (1.1.0 <= v < 1.2.0) + Handler: func(ctx context.Context, _ p2p.Peer, stream p2p.Stream) error { + w, r := protobuf.NewWriterAndReader(stream) + _, _ = w, r + fmt.Println("Client sent ping using v1.1.0 legacy format") + return nil + }, + }, + p2p.VersionedHandler{ + Version: semver.New("1.0.0"), // Legacy v1.0.0 client (1.0.0 <= v < 1.1.0) + Handler: func(ctx context.Context, _ p2p.Peer, stream p2p.Stream) error { + w, r := protobuf.NewWriterAndReader(stream) + _, _ = w, r + fmt.Println("Client sent ping using v1.0.0 legacy format") + return nil + }, + }, + ) + + return pingClient(ctx, p2p.Peer{Address: peer}, stream) +} + +// Example_versionedProtocol demonstrates constructing a versioned P2P protocol service +// and executing versioned message exchange between client and server nodes. +func Example_versionedProtocol() { + ctx := context.Background() + + serverSvc := NewExampleService(nil) + serverSpec := serverSvc.Protocol() + serverHandler := serverSpec.StreamSpecs[0].Handler + + // Client streamer connects to server where version 1.1.0 is negotiated: + clientStreamer := &mockStreamer{ + supportedVersions: map[string]bool{ + "1.1.0": true, + }, + } + clientSvc := NewExampleService(clientStreamer) + + // Client sends ping request (client-side dispatcher automatically selects v1.1.0 format) + _ = clientSvc.Ping(ctx, swarm.ZeroAddress) + + // Server receives incoming stream (server-side dispatcher automatically selects v1.1.0 handler) + incomingStream := mockStream{ + version: "1.1.0", + } + _ = serverHandler(ctx, p2p.Peer{Address: swarm.ZeroAddress}, incomingStream) + + // Output: + // Client sent ping using v1.1.0 legacy format + // Server received ping on v1.1.0 legacy handler +} diff --git a/pkg/p2p/p2p.go b/pkg/p2p/p2p.go index 17114540902..150f2c3655d 100644 --- a/pkg/p2p/p2p.go +++ b/pkg/p2p/p2p.go @@ -11,6 +11,7 @@ import ( "errors" "fmt" "io" + "sort" "time" "github.com/coreos/go-semver/semver" @@ -208,6 +209,40 @@ type HandlerFunc func(context.Context, Peer, Stream) error // HandlerMiddleware decorates a HandlerFunc by returning a new one. type HandlerMiddleware func(HandlerFunc) HandlerFunc +// VersionedHandler represents a HandlerFunc associated with a minimum supported Version threshold. +type VersionedHandler struct { + Version *semver.Version + Handler HandlerFunc +} + +// NewVersionedHandlersFunc creates a new HandlerFunc that dispatches stream execution +// based on the stream version. +// +// Handlers are evaluated in descending order of Version. The first handler where +// stream.Version >= handler.Version will be executed. +func NewVersionedHandlersFunc(handlers ...VersionedHandler) HandlerFunc { + sorted := make([]VersionedHandler, len(handlers)) + copy(sorted, handlers) + sort.Slice(sorted, func(i, j int) bool { + return sorted[j].Version.LessThan(*sorted[i].Version) + }) + + return func(ctx context.Context, p Peer, stream Stream) error { + v, err := stream.Version() + if err != nil { + return fmt.Errorf("get stream version: %w", err) + } + + for _, h := range sorted { + if !v.LessThan(*h.Version) { + return h.Handler(ctx, p, stream) + } + } + + return fmt.Errorf("no handler found for stream version: %s", v.String()) + } +} + // HeadlerFunc is returning response headers based on the received request // headers. type HeadlerFunc func(Headers, swarm.Address) Headers diff --git a/pkg/p2p/p2p_test.go b/pkg/p2p/p2p_test.go index 3c6ce22ca22..709bb495d79 100644 --- a/pkg/p2p/p2p_test.go +++ b/pkg/p2p/p2p_test.go @@ -5,9 +5,13 @@ package p2p_test import ( + "context" + "errors" "testing" + "github.com/coreos/go-semver/semver" "github.com/ethersphere/bee/v2/pkg/p2p" + "github.com/ethersphere/bee/v2/pkg/swarm" "github.com/libp2p/go-libp2p/core/network" ) @@ -36,3 +40,146 @@ func TestReachabilityStatus_String(t *testing.T) { } } } + +func TestNewVersionedHandlersFunc(t *testing.T) { + t.Parallel() + + var executed string + + makeHandler := func(name string) p2p.HandlerFunc { + return func(context.Context, p2p.Peer, p2p.Stream) error { + executed = name + return nil + } + } + + // Register handlers in intentionally unordered sequence to test automatic sorting + handlers := []p2p.VersionedHandler{ + {Version: semver.New("1.0.0"), Handler: makeHandler("v1.0.0")}, + {Version: semver.New("1.2.0"), Handler: makeHandler("v1.2.0")}, + {Version: semver.New("1.1.0"), Handler: makeHandler("v1.1.0")}, + } + + dispatcher := p2p.NewVersionedHandlersFunc(handlers...) + + tests := []struct { + name string + streamVersion string + wantExecuted string + wantErr bool + }{ + { + name: "exact match for highest version (1.2.0)", + streamVersion: "1.2.0", + wantExecuted: "v1.2.0", + }, + { + name: "newer patch version routes to highest version (1.2.5 -> v1.2.0)", + streamVersion: "1.2.5", + wantExecuted: "v1.2.0", + }, + { + name: "future minor version routes to highest version (1.3.0 -> v1.2.0)", + streamVersion: "1.3.0", + wantExecuted: "v1.2.0", + }, + { + name: "exact match for intermediate version (1.1.0)", + streamVersion: "1.1.0", + wantExecuted: "v1.1.0", + }, + { + name: "intermediate patch version (1.1.4 -> v1.1.0)", + streamVersion: "1.1.4", + wantExecuted: "v1.1.0", + }, + { + name: "exact match for lowest version (1.0.0)", + streamVersion: "1.0.0", + wantExecuted: "v1.0.0", + }, + { + name: "lowest version patch (1.0.9 -> v1.0.0)", + streamVersion: "1.0.9", + wantExecuted: "v1.0.0", + }, + { + name: "version below lowest registered version returns error (0.9.0)", + streamVersion: "0.9.0", + wantErr: true, + }, + { + name: "error when stream version cannot be retrieved", + streamVersion: "", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + executed = "" + err := dispatcher(context.Background(), p2p.Peer{}, mockStream{version: tt.streamVersion}) + + if tt.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + if executed != "" { + t.Fatalf("expected no handler to execute, but %q executed", executed) + } + return + } + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if executed != tt.wantExecuted { + t.Fatalf("executed handler = %q, want %q", executed, tt.wantExecuted) + } + }) + } +} + +type mockStream struct { + p2p.Stream + version string + closeFn func() error +} + +func (m mockStream) Version() (*semver.Version, error) { + if m.version == "" { + return nil, errors.New("missing version") + } + return semver.NewVersion(m.version) +} + +func (m mockStream) Close() error { + if m.closeFn != nil { + return m.closeFn() + } + return nil +} + +func (m mockStream) FullClose() error { + return m.Close() +} + +type mockStreamer struct { + p2p.Streamer + supportedVersions map[string]bool + closed bool +} + +func (m *mockStreamer) NewStream(_ context.Context, _ swarm.Address, _ p2p.Headers, _, version, _ string) (p2p.Stream, error) { + if !m.supportedVersions[version] { + return nil, errors.New("protocol version not supported") + } + return mockStream{ + version: version, + closeFn: func() error { + m.closed = true + return nil + }, + }, nil +} From 45afd0ce32b09f4e4e7f9d299efcbfceebeb2a45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jano=C5=A1=20Gulja=C5=A1?= Date: Thu, 30 Jul 2026 15:32:11 +0200 Subject: [PATCH 2/2] chore: rename example test file to be more clear --- pkg/p2p/{example_test.go => example_versioned_test.go} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename pkg/p2p/{example_test.go => example_versioned_test.go} (100%) diff --git a/pkg/p2p/example_test.go b/pkg/p2p/example_versioned_test.go similarity index 100% rename from pkg/p2p/example_test.go rename to pkg/p2p/example_versioned_test.go