tests: reliable workers tests for reordering and acks

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