add an integration test that transfers a lot of small files

.github/workflows/interop.sh

@@ -4,22 +4,25 @@ set -e
 
 SERVER=$1
 CLIENT=$2
+TEST=$3
 SERVER_ADDR="/ip4/127.0.0.1/udp/12345/quic"
 
 for k1 in server*.key; do
   for k2 in client*.key; do
-    # Run the handshake failure test.
-    for k3 in other*.key; do
-      echo "Running with server $SERVER ($k1) and client $CLIENT ($k2). Test: handshake-failure"
-      ./$SERVER -role server -key $k1 -peerkey $k2 -addr $SERVER_ADDR -test handshake-failure &
-      PID=$!
-      ./$CLIENT -role client -key $k2 -peerkey $k3 -addr $SERVER_ADDR -test handshake-failure
-      kill $PID
-    done;
-    # Run the transfer test.
-    echo "Running with server $SERVER ($k1) and client $CLIENT ($k2)"
-    ./$SERVER -role server -key $k1 -peerkey $k2 -addr $SERVER_ADDR -test single-transfer &
-    ./$CLIENT -role client -key $k2 -peerkey $k1 -addr $SERVER_ADDR -test single-transfer
-    wait &
+    if [[ $TEST == "handshake-failure" ]]; then
+      for k3 in other*.key; do
+        echo -e "\nRunning with server $SERVER ($k1) and client $CLIENT ($k2). Test: handshake-failure"
+        ./$SERVER -role server -key $k1 -peerkey $k2 -addr $SERVER_ADDR -test handshake-failure &
+        PID=$!
+        time ./$CLIENT -role client -key $k2 -peerkey $k3 -addr $SERVER_ADDR -test handshake-failure
+        kill $PID
+        wait
+      done;
+    else
+      echo -e "\nRunning with server $SERVER ($k1) and client $CLIENT ($k2). Test: $TEST"
+      ./$SERVER -role server -key $k1 -peerkey $k2 -addr $SERVER_ADDR -test $TEST &
+      time ./$CLIENT -role client -key $k2 -peerkey $k1 -addr $SERVER_ADDR -test $TEST
+      wait
+    fi
   done;
 done;

.github/workflows/interop.yml

@@ -58,11 +58,17 @@ jobs:
           rm -rf integrationtests || true
           mv builder integrationtests
           git reflog --decorate -1
-          if [[ `git merge-base --is-ancestor HEAD 126c64772ba0aef0b2b6d58ff36e55a93f9253a7; echo $?` == "1" ]]; then
-            go build -o transport-go${{ matrix.cfg.go }}-${{ matrix.cfg.commit }} integrationtests/main.go
-          else
-            go build -tags oldstream -o transport-go${{ matrix.cfg.go }}-${{ matrix.cfg.commit }} integrationtests/main.go
+          TAGS=()
+          if [[ `git merge-base --is-ancestor HEAD 126c64772ba0aef0b2b6d58ff36e55a93f9253a7; echo $?` != "1" ]]; then
+            TAGS+=("oldstream")
           fi
+          if [[ "${{ matrix.cfg.retireBugBackwardsCompatiblityMode }}" == "true" ]]; then
+            TAGS+=("retirebugcompatmode")
+          fi
+          function join { local IFS="$1"; shift; echo "$*"; }
+          tags=$(join , ${TAGS[@]})
+          if [[ -n $tags ]]; then tags="-tags $tags"; fi
+          go build $tags -o transport-go${{ matrix.cfg.go }}-${{ matrix.cfg.commit }} integrationtests/main.go
       - name: Upload binary
         uses: actions/upload-artifact@v2
         with:
@@ -89,8 +95,12 @@ jobs:
         with:
           name: binary
           path: interop/
-      - name: Run interop
-        run: |
-          cd interop
-          chmod 744 transport*
-          ../.github/workflows/interop.sh transport-go${{ matrix.server.go }}-${{ matrix.server.commit }} transport-go${{ matrix.client.go }}-${{ matrix.client.commit }}
+      - name: Set permissions
+        run: chmod 744 interop/transport*
+      - name: 'Run test case: handshake-failure'
+        run: cd interop && ../.github/workflows/interop.sh transport-go${{ matrix.server.go }}-${{ matrix.server.commit }} transport-go${{ matrix.client.go }}-${{ matrix.client.commit }} handshake-failure
+      - name: 'Run test case: single-transfer'
+        run: cd interop && ../.github/workflows/interop.sh transport-go${{ matrix.server.go }}-${{ matrix.server.commit }} transport-go${{ matrix.client.go }}-${{ matrix.client.commit }} single-transfer
+      - name: 'Run test case: multi-transfer'
+        if: ${{ matrix.client.retireBugBackwardsCompatiblityMode || matrix.server.retireBugBackwardsCompatiblityMode }}
+        run: cd interop && ../.github/workflows/interop.sh transport-go${{ matrix.server.go }}-${{ matrix.server.commit }} transport-go${{ matrix.client.go }}-${{ matrix.client.commit }} multi-transfer

.github/workflows/matrix.jsonc

@@ -1,11 +1,13 @@
 [
-  { "go": "1.13", "commit": "v0.6.0" },
-  { "go": "1.14", "commit": "v0.6.0" },
-  { "go": "1.13", "commit": "v0.7.0" },
-  { "go": "1.14", "commit": "v0.7.0" },
+  { "go": "1.13", "commit": "v0.6.0", "retireBugBackwardsCompatiblityMode": false },
+  { "go": "1.14", "commit": "v0.6.0", "retireBugBackwardsCompatiblityMode": false },
+  { "go": "1.13", "commit": "v0.7.0", "retireBugBackwardsCompatiblityMode": false },
+  { "go": "1.14", "commit": "v0.7.0", "retireBugBackwardsCompatiblityMode": false },
+  { "go": "1.13", "commit": "v0.7.0", "retireBugBackwardsCompatiblityMode": true },
+  { "go": "1.14", "commit": "v0.7.0", "retireBugBackwardsCompatiblityMode": true },
   // v0.7.1 was never released in IPFS.
-  { "go": "1.14", "commit": "v0.8.0" },
-  { "go": "1.15", "commit": "v0.8.0" },
-  { "go": "1.14", "commit": "HEAD" },
-  { "go": "1.15", "commit": "HEAD" }
+  { "go": "1.14", "commit": "v0.8.0", "retireBugBackwardsCompatiblityMode": true },
+  { "go": "1.15", "commit": "v0.8.0", "retireBugBackwardsCompatiblityMode": true },
+  { "go": "1.14", "commit": "HEAD", "retireBugBackwardsCompatiblityMode": true },
+  { "go": "1.15", "commit": "HEAD", "retireBugBackwardsCompatiblityMode": true }
 ]

go.mod

@@ -14,4 +14,5 @@ require (
 	github.com/onsi/ginkgo v1.14.0
 	github.com/onsi/gomega v1.10.1
 	golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a
+	golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e
 )

go.sum

@@ -320,6 +320,7 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

integrationtests/main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/transport"
 	libp2pquic "github.com/libp2p/go-libp2p-quic-transport"
 	ma "github.com/multiformats/go-multiaddr"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/libp2p/go-libp2p-quic-transport/integrationtests/stream"
 )
@@ -129,6 +130,8 @@ func runClient(hostKey crypto.PrivKey, serverKey crypto.PubKey, addr ma.Multiadd
 	switch test {
 	case "single-transfer":
 		return testSingleFileTransfer(tr, serverKey, addr)
+	case "multi-transfer":
+		return testMultipleFileTransfer(tr, serverKey, addr)
 	case "handshake-failure":
 		return testHandshakeFailure(tr, serverKey, addr)
 	default:
@@ -141,11 +144,11 @@ func testSingleFileTransfer(tr transport.Transport, serverKey crypto.PubKey, add
 	if err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
 	conn, err := tr.Dial(ctx, addr, serverPeerID)
 	if err != nil {
-		return err
+		return fmt.Errorf("Dial failed: %w", err)
 	}
 	defer conn.Close()
 	if !conn.RemotePublicKey().Equals(serverKey) {
@@ -177,6 +180,53 @@ func testSingleFileTransfer(tr transport.Transport, serverKey crypto.PubKey, add
 	return nil
 }
 
+func testMultipleFileTransfer(tr transport.Transport, serverKey crypto.PubKey, addr ma.Multiaddr) error {
+	serverPeerID, err := peer.IDFromPublicKey(serverKey)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	conn, err := tr.Dial(ctx, addr, serverPeerID)
+	if err != nil {
+		return fmt.Errorf("Dial failed: %w", err)
+	}
+	defer conn.Close()
+	if !conn.RemotePublicKey().Equals(serverKey) {
+		return errors.New("mismatching public keys")
+	}
+	if conn.RemotePeer() != serverPeerID {
+		return fmt.Errorf("remote Peer ID mismatch. Got %s, expected %s", conn.RemotePeer().Pretty(), serverPeerID.Pretty())
+	}
+	var g errgroup.Group
+	for i := 0; i < 2000; i++ {
+		g.Go(func() error {
+			st, err := conn.OpenStream()
+			if err != nil {
+				return err
+			}
+			str := stream.WrapStream(st)
+			data := make([]byte, 1<<9)
+			rand.Read(data)
+			if _, err := str.Write(data); err != nil {
+				return err
+			}
+			if err := str.CloseWrite(); err != nil {
+				return err
+			}
+			echoed, err := ioutil.ReadAll(str)
+			if err != nil {
+				return err
+			}
+			if !bytes.Equal(data, echoed) {
+				return errors.New("echoed data does not match")
+			}
+			return nil
+		})
+	}
+	return g.Wait()
+}
+
 func testHandshakeFailure(tr transport.Transport, serverKey crypto.PubKey, addr ma.Multiaddr) error {
 	serverPeerID, err := peer.IDFromPublicKey(serverKey)
 	if err != nil {