feat(p2p): expose peer protocol version for backward compatibilities (#5539)

Author: janos

pkg/p2p/libp2p/internal/handshake/mock/stream.go

@@ -7,6 +7,7 @@ package mock
 import (
 	"bytes"
 
+	"github.com/coreos/go-semver/semver"
 	"github.com/ethersphere/bee/v2/pkg/p2p"
 )
 
@@ -72,3 +73,7 @@ func (s *Stream) FullClose() error {
 func (s *Stream) Reset() error {
 	return nil
 }
+
+func (s *Stream) Version() (*semver.Version, error) {
+	return nil, nil
+}

pkg/p2p/libp2p/stream.go

@@ -7,8 +7,10 @@ package libp2p
 import (
 	"errors"
 	"io"
+	"strings"
 	"time"
 
+	"github.com/coreos/go-semver/semver"
 	"github.com/ethersphere/bee/v2/pkg/p2p"
 	"github.com/libp2p/go-libp2p/core/network"
 )
@@ -38,6 +40,15 @@ func (s *stream) ResponseHeaders() p2p.Headers {
 	return s.responseHeaders
 }
 
+func (s *stream) Version() (*semver.Version, error) {
+	parts := strings.Split(string(s.Protocol()), "/")
+	partsLen := len(parts)
+	if partsLen < 2 {
+		return nil, errors.New("invalid protocol version")
+	}
+	return semver.NewVersion(parts[partsLen-2])
+}
+
 func (s *stream) Reset() error {
 	defer s.metrics.StreamResetCount.Inc()
 	return s.Stream.Reset()

pkg/p2p/libp2p/stream_test.go

@@ -0,0 +1,90 @@
+// Copyright 2020 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 libp2p_test
+
+import (
+	"testing"
+
+	"github.com/coreos/go-semver/semver"
+	"github.com/ethersphere/bee/v2/pkg/p2p/libp2p"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/protocol"
+)
+
+type mockNetStream struct {
+	network.Stream
+	protocol protocol.ID
+}
+
+func (m *mockNetStream) Protocol() protocol.ID {
+	return m.protocol
+}
+
+func TestStreamVersion(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name       string
+		protocolID protocol.ID
+		wantMajor  int64
+		wantMinor  int64
+		wantPatch  int64
+		wantErr    bool
+	}{
+		{
+			name:       "valid standard version",
+			protocolID: "/swarm/pingpong/1.2.3/ping",
+			wantMajor:  1,
+			wantMinor:  2,
+			wantPatch:  3,
+		},
+		{
+			name:       "valid rc version",
+			protocolID: "/swarm/pingpong/2.0.0-rc1/ping",
+			wantMajor:  2,
+			wantMinor:  0,
+			wantPatch:  0,
+		},
+		{
+			name:       "invalid version format",
+			protocolID: "/swarm/pingpong/abc/ping",
+			wantErr:    true,
+		},
+		{
+			name:       "too short protocol ID",
+			protocolID: "/ping",
+			wantErr:    true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			s := &mockNetStream{protocol: tc.protocolID}
+			srv := &libp2p.Service{}
+			wrapped := srv.WrapStream(s)
+
+			v, err := wrapped.Version()
+			if tc.wantErr {
+				if err == nil {
+					t.Fatal("expected error, got nil")
+				}
+				return
+			}
+
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+
+			if v == nil {
+				t.Fatal("expected version to be non-nil")
+			}
+
+			expected := semver.Version{Major: tc.wantMajor, Minor: tc.wantMinor}
+			if v.Major != expected.Major || v.Minor != expected.Minor {
+				t.Errorf("got version %v, want %v", v, expected)
+			}
+		})
+	}
+}

pkg/p2p/p2p.go

@@ -13,6 +13,7 @@ import (
 	"io"
 	"time"
 
+	"github.com/coreos/go-semver/semver"
 	"github.com/ethersphere/bee/v2/pkg/bzz"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	"github.com/libp2p/go-libp2p/core/network"
@@ -166,6 +167,7 @@ type Stream interface {
 	Headers() Headers
 	FullClose() error
 	Reset() error
+	Version() (*semver.Version, error)
 }
 
 // ProtocolSpec defines a collection of Stream specifications with handlers.

pkg/p2p/protobuf/protobuf_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"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/p2p/protobuf/internal/pb"
@@ -347,6 +348,10 @@ func (noopWriteCloser) ResponseHeaders() p2p.Headers {
 	return nil
 }
 
+func (noopWriteCloser) Version() (*semver.Version, error) {
+	return nil, nil
+}
+
 func (noopWriteCloser) Close() error {
 	return nil
 }
@@ -379,6 +384,10 @@ func (noopReadCloser) ResponseHeaders() p2p.Headers {
 	return nil
 }
 
+func (noopReadCloser) Version() (*semver.Version, error) {
+	return nil, nil
+}
+
 func (noopReadCloser) Close() error {
 	return nil
 }

pkg/p2p/streamtest/streamtest.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/coreos/go-semver/semver"
 	"github.com/ethersphere/bee/v2/pkg/p2p"
 	"github.com/ethersphere/bee/v2/pkg/spinlock"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
@@ -122,10 +123,12 @@ func (r *Recorder) NewStream(ctx context.Context, addr swarm.Address, h p2p.Head
 		}
 	}
 
+	version, versionErr := semver.NewVersion(protocolVersion)
+
 	recordIn := newRecord(r.messageLatency)
 	recordOut := newRecord(r.messageLatency)
-	streamOut := newStream(recordIn, recordOut)
-	streamIn := newStream(recordOut, recordIn)
+	streamOut := newStream(recordIn, recordOut, version, versionErr)
+	streamIn := newStream(recordOut, recordIn, version, versionErr)
 
 	var handler p2p.HandlerFunc
 	var headler p2p.HeadlerFunc
@@ -260,10 +263,12 @@ type stream struct {
 	responseHeaders p2p.Headers
 	closed          bool
 	lock            sync.Mutex
+	version         *semver.Version
+	versionErr      error
 }
 
-func newStream(in, out *record) *stream {
-	return &stream{in: in, out: out}
+func newStream(in, out *record, version *semver.Version, versionErr error) *stream {
+	return &stream{in: in, out: out, version: version, versionErr: versionErr}
 }
 
 func (s *stream) Read(p []byte) (int, error) {
@@ -290,6 +295,10 @@ func (s *stream) ResponseHeaders() p2p.Headers {
 	return s.responseHeaders
 }
 
+func (s *stream) Version() (*semver.Version, error) {
+	return s.version, s.versionErr
+}
+
 func (s *stream) Close() error {
 	s.lock.Lock()
 	defer s.lock.Unlock()

pkg/p2p/streamtest/streamtest_test.go

@@ -15,6 +15,7 @@ import (
 	"testing/synctest"
 	"time"
 
+	"github.com/coreos/go-semver/semver"
 	"github.com/ethersphere/bee/v2/pkg/p2p"
 	"github.com/ethersphere/bee/v2/pkg/p2p/streamtest"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
@@ -878,3 +879,45 @@ func testRecords(t *testing.T, records []*streamtest.Record, want [][2]string, w
 		}
 	}
 }
+
+func TestStreamVersion(t *testing.T) {
+	t.Parallel()
+
+	recorder := streamtest.New(
+		streamtest.WithProtocols(
+			newTestProtocol(func(_ context.Context, peer p2p.Peer, stream p2p.Stream) error {
+				v, err := stream.Version()
+				if err != nil {
+					t.Errorf("handler: unexpected error: %v", err)
+				}
+				if v == nil || v.String() != "1.0.1" {
+					t.Errorf("handler: got version %v, want 1.0.1", v)
+				}
+				return nil
+			}),
+		),
+	)
+
+	stream, err := recorder.NewStream(context.Background(), swarm.ZeroAddress, nil, testProtocolName, testProtocolVersion, testStreamName)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stream.Close()
+
+	v, err := stream.Version()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if v == nil {
+		t.Fatal("nil version")
+	}
+
+	if v.String() != "1.0.1" {
+		t.Fatalf("got string version %v, want 1.0.1", v)
+	}
+
+	if !v.Equal(semver.Version{Major: 1, Minor: 0, Patch: 1}) {
+		t.Fatalf("got semver version %v, want 1.0.1", v)
+	}
+}