tests: reliable workers tests for reordering and acks (#60)

Tests for down and up workers in reliable service, covering reordering
towards the upper layer and checking that any incoming packet is
eventually ACKed.

This PR also includes some needed test utilities that can be reused to
exercise different parts of the system.

---------

Co-authored-by: Simone Basso <bassosimone@gmail.com>

.github/workflows/build-refactor.yml

@@ -0,0 +1,57 @@
+name: build-refactor
+# this action is covering internal/ tree with go1.21
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  short-tests:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: setup go
+      uses: actions/setup-go@v2
+      with:
+        go-version: '1.21'
+    - name: Run short tests
+      run: go test --short -cover ./internal/...
+
+  gosec:
+    runs-on: ubuntu-latest
+    env:
+        GO111MODULE: on
+    steps:
+    - name: Checkout Source
+      uses: actions/checkout@v2
+    - name: Run Gosec security scanner
+      uses: securego/gosec@master
+      with:
+        args: '-no-fail ./...'
+
+  coverage-threshold:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: setup go
+      uses: actions/setup-go@v2
+      with:
+        go-version: '1.21'
+    - name: Ensure coverage threshold
+      run: make test-coverage-threshold-refactor
+
+  integration:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: setup go
+        uses: actions/setup-go@v2
+        with:
+          go-version: '1.21'
+      - name: run integration tests
+        run: go test -v ./tests/integration
+

.gitignore

@@ -8,8 +8,6 @@
 *.swo
 *.pem
 *.ovpn
+/*.out
 data/*
 measurements/*
-coverage.out
-coverage-ping.out
-cov-threshold.out

Makefile

@@ -3,7 +3,7 @@ TARGET ?= "1.1.1.1"
 COUNT ?= 5
 TIMEOUT ?= 10
 LOCAL_TARGET := $(shell ip -4 addr show docker0 | grep 'inet ' | awk '{print $$2}' | cut -f 1 -d /)
-COVERAGE_THRESHOLD := 88
+COVERAGE_THRESHOLD := 80
 FLAGS=-ldflags="-w -s -buildid=none -linkmode=external" -buildmode=pie -buildvcs=false
 
 build:
@@ -33,10 +33,17 @@ test:
 test-coverage:
 	go test -coverprofile=coverage.out ./vpn
 
+test-coverage-refactor:
+	go test -coverprofile=coverage.out ./internal/...
+
 test-coverage-threshold:
 	go test --short -coverprofile=cov-threshold.out ./vpn
 	./scripts/go-coverage-check.sh cov-threshold.out ${COVERAGE_THRESHOLD}
 
+test-coverage-threshold-refactor:
+	go test --short -coverprofile=cov-threshold-refactor.out ./internal/...
+	./scripts/go-coverage-check.sh cov-threshold-refactor.out ${COVERAGE_THRESHOLD}
+
 test-short:
 	go test -race -short -v ./...
 

internal/model/packet.go

@@ -32,6 +32,33 @@ const (
 	P_DATA_V2                                         // 9
 )
 
+// NewOpcodeFromString returns an opcode from a string representation, and an error if it cannot parse the opcode
+// representation. The zero return value is invalid and always coupled with a non-nil error.
+func NewOpcodeFromString(s string) (Opcode, error) {
+	switch s {
+	case "CONTROL_HARD_RESET_CLIENT_V1":
+		return P_CONTROL_HARD_RESET_CLIENT_V1, nil
+	case "CONTROL_HARD_RESET_SERVER_V1":
+		return P_CONTROL_HARD_RESET_SERVER_V1, nil
+	case "CONTROL_SOFT_RESET_V1":
+		return P_CONTROL_SOFT_RESET_V1, nil
+	case "CONTROL_V1":
+		return P_CONTROL_V1, nil
+	case "ACK_V1":
+		return P_ACK_V1, nil
+	case "DATA_V1":
+		return P_DATA_V1, nil
+	case "CONTROL_HARD_RESET_CLIENT_V2":
+		return P_CONTROL_HARD_RESET_CLIENT_V2, nil
+	case "CONTROL_HARD_RESET_SERVER_V2":
+		return P_CONTROL_HARD_RESET_SERVER_V2, nil
+	case "DATA_V2":
+		return P_DATA_V2, nil
+	default:
+		return 0, errors.New("unknown opcode")
+	}
+}
+
 // String returns the opcode string representation
 func (op Opcode) String() string {
 	switch op {

internal/reliabletransport/common_test.go

@@ -0,0 +1,47 @@
+package reliabletransport
+
+import (
+	"github.com/apex/log"
+	"github.com/ooni/minivpn/internal/bytesx"
+	"github.com/ooni/minivpn/internal/model"
+	"github.com/ooni/minivpn/internal/runtimex"
+	"github.com/ooni/minivpn/internal/session"
+	"github.com/ooni/minivpn/internal/workers"
+)
+
+//
+// Common utilities for tests in this package.
+//
+
+// initManagers initializes a workers manager and a session manager.
+func initManagers() (*workers.Manager, *session.Manager) {
+	w := workers.NewManager(log.Log)
+	s, err := session.NewManager(log.Log)
+	runtimex.PanicOnError(err, "cannot create session manager")
+	return w, s
+}
+
+// newRandomSessionID returns a random session ID to initialize mock sessions.
+func newRandomSessionID() model.SessionID {
+	b, err := bytesx.GenRandomBytes(8)
+	if err != nil {
+		panic(err)
+	}
+	return model.SessionID(b)
+}
+
+func ackSetFromInts(s []int) *ackSet {
+	acks := make([]model.PacketID, 0)
+	for _, i := range s {
+		acks = append(acks, model.PacketID(i))
+	}
+	return newACKSet(acks...)
+}
+
+func ackSetFromRange(start, total int) *ackSet {
+	acks := make([]model.PacketID, 0)
+	for i := 0; i < total; i++ {
+		acks = append(acks, model.PacketID(start+i))
+	}
+	return newACKSet(acks...)
+}

internal/reliabletransport/receiver.go

@@ -3,7 +3,6 @@ package reliabletransport
 import (
 	"bytes"
 	"fmt"
-	"log"
 	"sort"
 
 	"github.com/ooni/minivpn/internal/model"
@@ -42,7 +41,7 @@ func (ws *workersState) moveUpWorker() {
 			// We should be able to deterministically test how this affects the state machine.
 
 			// drop a packet that is not for our session
-			if !bytes.Equal(packet.LocalSessionID[:], ws.sessionManager.RemoteSessionID()) {
+			if !bytes.Equal(packet.RemoteSessionID[:], ws.sessionManager.LocalSessionID()) {
 				ws.logger.Warnf(
 					"%s: packet with invalid RemoteSessionID: expected %x; got %x",
 					workerName,
@@ -64,6 +63,7 @@ func (ws *workersState) moveUpWorker() {
 
 			if inserted := receiver.MaybeInsertIncoming(packet); !inserted {
 				// this packet was not inserted in the queue: we drop it
+				ws.logger.Debugf("Dropping packet: %v", packet.ID)
 				continue
 			}
 

internal/reliabletransport/reliable_ack_test.go

@@ -0,0 +1,164 @@
+package reliabletransport
+
+import (
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/apex/log"
+	"github.com/ooni/minivpn/internal/model"
+	"github.com/ooni/minivpn/internal/vpntest"
+)
+
+// test that everything that is received from below is eventually ACKed to the sender.
+func TestReliable_ACK(t *testing.T) {
+
+	log.SetLevel(log.DebugLevel)
+
+	type args struct {
+		inputSequence []string
+		start         int
+		wantacks      int
+	}
+
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "ten ordered packets in",
+			args: args{
+				inputSequence: []string{
+					"[1] CONTROL_V1 +1ms",
+					"[2] CONTROL_V1 +1ms",
+					"[3] CONTROL_V1 +1ms",
+					"[4] CONTROL_V1 +1ms",
+					"[5] CONTROL_V1 +1ms",
+					"[6] CONTROL_V1 +1ms",
+					"[7] CONTROL_V1 +1ms",
+					"[8] CONTROL_V1 +1ms",
+					"[9] CONTROL_V1 +1ms",
+					"[10] CONTROL_V1 +1ms",
+				},
+				start:    1,
+				wantacks: 10,
+			},
+		},
+		{
+			name: "five ordered packets with offset",
+			args: args{
+				inputSequence: []string{
+					"[100] CONTROL_V1 +1ms",
+					"[101] CONTROL_V1 +1ms",
+					"[102] CONTROL_V1 +1ms",
+					"[103] CONTROL_V1 +1ms",
+					"[104] CONTROL_V1 +1ms",
+				},
+				start:    100,
+				wantacks: 5,
+			},
+		},
+		{
+			name: "five reversed packets",
+			args: args{
+				inputSequence: []string{
+					"[5] CONTROL_V1 +1ms",
+					"[4] CONTROL_V1 +1ms",
+					"[3] CONTROL_V1 +1ms",
+					"[2] CONTROL_V1 +1ms",
+					"[1] CONTROL_V1 +1ms",
+				},
+				start:    1,
+				wantacks: 5,
+			},
+		},
+		{
+			name: "ten unordered packets with duplicates",
+			args: args{
+				inputSequence: []string{
+					"[5] CONTROL_V1 +1ms",
+					"[1] CONTROL_V1 +1ms",
+					"[5] CONTROL_V1 +1ms",
+					"[2] CONTROL_V1 +1ms",
+					"[1] CONTROL_V1 +1ms",
+					"[4] CONTROL_V1 +1ms",
+					"[2] CONTROL_V1 +1ms",
+					"[3] CONTROL_V1 +1ms",
+					"[3] CONTROL_V1 +1ms",
+					"[4] CONTROL_V1 +1ms",
+				},
+				start:    1,
+				wantacks: 5,
+			},
+		},
+		/*
+			{
+				name: "a burst of packets",
+				args: args{
+					inputSequence: []string{
+						"[5] CONTROL_V1 +1ms",
+						"[1] CONTROL_V1 +1ms",
+						"[3] CONTROL_V1 +1ms",
+						"[2] CONTROL_V1 +1ms",
+						"[4] CONTROL_V1 +1ms",
+					},
+					start:    1,
+					wantacks: 5,
+				},
+			},
+		*/
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &Service{}
+
+			// just to properly initialize it, we don't care about these
+			s.ControlToReliable = make(chan *model.Packet)
+			// this one up to control/tls also needs to be buffered because otherwise
+			// we'll block on the receiver when delivering up.
+			reliableToControl := make(chan *model.Packet, 1024)
+			s.ReliableToControl = &reliableToControl
+
+			// the only two channels we're going to be testing on this test
+			// we want to buffer enough to be safe writing to them.
+			dataIn := make(chan *model.Packet, 1024)
+			dataOut := make(chan *model.Packet, 1024)
+
+			s.MuxerToReliable = dataIn        // up
+			s.DataOrControlToMuxer = &dataOut // down
+
+			workers, session := initManagers()
+
+			t0 := time.Now()
+
+			// let the workers pump up the jam!
+			s.StartWorkers(log.Log, workers, session)
+
+			writer := vpntest.NewPacketWriter(dataIn)
+
+			// initialize a mock session ID for our peer
+			peerSessionID := newRandomSessionID()
+
+			writer.RemoteSessionID = model.SessionID(session.LocalSessionID())
+			writer.LocalSessionID = peerSessionID
+			session.SetRemoteSessionID(peerSessionID)
+
+			go writer.WriteSequence(tt.args.inputSequence)
+
+			reader := vpntest.NewPacketReader(dataOut)
+			witness := vpntest.NewWitness(reader)
+
+			if ok := witness.VerifyNumberOfACKs(tt.args.start, tt.args.wantacks, t0); !ok {
+				got := len(witness.Log().ACKs())
+				t.Errorf("TestACK: got = %v, want %v", got, tt.args.wantacks)
+			}
+			gotAckSet := ackSetFromInts(witness.Log().ACKs()).sorted()
+			wantAckSet := ackSetFromRange(tt.args.start, tt.args.wantacks).sorted()
+
+			if !slices.Equal(gotAckSet, wantAckSet) {
+				t.Errorf("TestACK: got = %v, want %v", gotAckSet, wantAckSet)
+
+			}
+		})
+	}
+}