feat(p2p): add versioned handlers function

Author: janos

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
+}

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

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
+}