feat(enginenetx): add algorithms to filter, mix, and stream tactics (#…

…1556)

This diff extends the enginenetx package to add algorithms to filter,
mix, and stream tactics.

We will use this algorithms to simplify the implementation and make it
more composable.

This work is part of ooni/probe#2704.

internal/enginenetx/filter.go

@@ -0,0 +1,73 @@
+package enginenetx
+
+// filterOutNilTactics filters out nil tactics.
+//
+// This function returns a channel where we emit the edited
+// tactics, and which we clone when we're done.
+func filterOutNilTactics(input <-chan *httpsDialerTactic) <-chan *httpsDialerTactic {
+	output := make(chan *httpsDialerTactic)
+	go func() {
+		defer close(output)
+		for tx := range input {
+			if tx != nil {
+				output <- tx
+			}
+		}
+	}()
+	return output
+}
+
+// filterOnlyKeepUniqueTactics only keeps unique tactics.
+//
+// This function returns a channel where we emit the edited
+// tactics, and which we clone when we're done.
+func filterOnlyKeepUniqueTactics(input <-chan *httpsDialerTactic) <-chan *httpsDialerTactic {
+	output := make(chan *httpsDialerTactic)
+	go func() {
+
+		// make sure we close output chan
+		defer close(output)
+
+		// useful to make sure we don't emit two equal policy in a single run
+		uniq := make(map[string]int)
+
+		for tx := range input {
+			// handle the case in which we already emitted a tactic
+			key := tx.tacticSummaryKey()
+			if uniq[key] > 0 {
+				continue
+			}
+			uniq[key]++
+
+			// emit the tactic
+			output <- tx
+		}
+
+	}()
+	return output
+}
+
+// filterAssignInitialDelays assigns initial delays to tactics.
+//
+// This function returns a channel where we emit the edited
+// tactics, and which we clone when we're done.
+func filterAssignInitialDelays(input <-chan *httpsDialerTactic) <-chan *httpsDialerTactic {
+	output := make(chan *httpsDialerTactic)
+	go func() {
+
+		// make sure we close output chan
+		defer close(output)
+
+		index := 0
+		for tx := range input {
+			// rewrite the delays
+			tx.InitialDelay = happyEyeballsDelay(index)
+			index++
+
+			// emit the tactic
+			output <- tx
+		}
+
+	}()
+	return output
+}

internal/enginenetx/filter_test.go

@@ -0,0 +1,111 @@
+package enginenetx
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/ooni/probe-cli/v3/internal/testingx"
+)
+
+func TestFilterOutNilTactics(t *testing.T) {
+	inputs := []*httpsDialerTactic{
+		nil,
+		nil,
+		{
+			Address:        "130.192.91.211",
+			InitialDelay:   0,
+			Port:           "443",
+			SNI:            "x.org",
+			VerifyHostname: "api.ooni.io",
+		},
+		nil,
+		{
+			Address:        "130.192.91.211",
+			InitialDelay:   0,
+			Port:           "443",
+			SNI:            "www.polito.it",
+			VerifyHostname: "api.ooni.io",
+		},
+		nil,
+		nil,
+	}
+
+	expect := []*httpsDialerTactic{
+		inputs[2], inputs[4],
+	}
+
+	var output []*httpsDialerTactic
+	for tx := range filterOutNilTactics(streamTacticsFromSlice(inputs)) {
+		output = append(output, tx)
+	}
+
+	if diff := cmp.Diff(expect, output); diff != "" {
+		t.Fatal(diff)
+	}
+}
+
+func TestFilterOnlyKeepUniqueTactics(t *testing.T) {
+	templates := []*httpsDialerTactic{{
+		Address:        "130.192.91.211",
+		InitialDelay:   0,
+		Port:           "443",
+		SNI:            "www.example.com",
+		VerifyHostname: "api.ooni.io",
+	}, {
+		Address:        "130.192.91.211",
+		InitialDelay:   0,
+		Port:           "443",
+		SNI:            "www.kernel.org",
+		VerifyHostname: "api.ooni.io",
+	}, {
+		Address:        "130.192.91.211",
+		InitialDelay:   0,
+		Port:           "443",
+		SNI:            "x.org",
+		VerifyHostname: "api.ooni.io",
+	}, {
+		Address:        "130.192.91.211",
+		InitialDelay:   0,
+		Port:           "443",
+		SNI:            "www.polito.it",
+		VerifyHostname: "api.ooni.io",
+	}}
+
+	inputs := []*httpsDialerTactic{
+		templates[2], templates[1], templates[1],
+		templates[2], templates[2], templates[1],
+		templates[0], templates[1], templates[0],
+		templates[2], templates[1], templates[2],
+		templates[1], templates[0], templates[1],
+		templates[3], // only once at the end
+	}
+
+	expect := []*httpsDialerTactic{
+		templates[2], templates[1], templates[0], templates[3],
+	}
+
+	var output []*httpsDialerTactic
+	for tx := range filterOnlyKeepUniqueTactics(streamTacticsFromSlice(inputs)) {
+		output = append(output, tx)
+	}
+
+	if diff := cmp.Diff(expect, output); diff != "" {
+		t.Fatal(diff)
+	}
+}
+
+func TestFilterAssignInitalDelays(t *testing.T) {
+	inputs := []*httpsDialerTactic{}
+	ff := &testingx.FakeFiller{}
+	ff.Fill(&inputs)
+	idx := 0
+	for tx := range filterAssignInitialDelays(streamTacticsFromSlice(inputs)) {
+		if tx.InitialDelay != happyEyeballsDelay(idx) {
+			t.Fatal("unexpected .InitialDelay", tx.InitialDelay, "for", idx)
+		}
+		idx++
+	}
+	if idx < 1 {
+		t.Fatal("expected to see at least one entry")
+	}
+}

internal/enginenetx/mix.go

@@ -0,0 +1,19 @@
+package enginenetx
+
+// mixSequentially mixes entries from primary followed by entries from fallback.
+//
+// This function returns a channel where we emit the edited
+// tactics, and which we clone when we're done.
+func mixSequentially(primary, fallback <-chan *httpsDialerTactic) <-chan *httpsDialerTactic {
+	output := make(chan *httpsDialerTactic)
+	go func() {
+		defer close(output)
+		for tx := range primary {
+			output <- tx
+		}
+		for tx := range fallback {
+			output <- tx
+		}
+	}()
+	return output
+}

internal/enginenetx/mix_test.go

@@ -0,0 +1,29 @@
+package enginenetx
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/ooni/probe-cli/v3/internal/testingx"
+)
+
+func TestMixSequentially(t *testing.T) {
+	primary := []*httpsDialerTactic{}
+	fallback := []*httpsDialerTactic{}
+
+	ff := &testingx.FakeFiller{}
+	ff.Fill(&primary)
+	ff.Fill(&fallback)
+
+	expect := append([]*httpsDialerTactic{}, primary...)
+	expect = append(expect, fallback...)
+
+	var output []*httpsDialerTactic
+	for tx := range mixSequentially(streamTacticsFromSlice(primary), streamTacticsFromSlice(fallback)) {
+		output = append(output, tx)
+	}
+
+	if diff := cmp.Diff(expect, output); diff != "" {
+		t.Fatal(diff)
+	}
+}

internal/enginenetx/stream.go

@@ -0,0 +1,16 @@
+package enginenetx
+
+// streamTacticsFromSlice streams tactics from a given slice.
+//
+// This function returns a channel where we emit the edited
+// tactics, and which we clone when we're done.
+func streamTacticsFromSlice(input []*httpsDialerTactic) <-chan *httpsDialerTactic {
+	output := make(chan *httpsDialerTactic)
+	go func() {
+		defer close(output)
+		for _, tx := range input {
+			output <- tx
+		}
+	}()
+	return output
+}

internal/enginenetx/stream_test.go

@@ -0,0 +1,23 @@
+package enginenetx
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/ooni/probe-cli/v3/internal/testingx"
+)
+
+func TestStreamTacticsFromSlice(t *testing.T) {
+	input := []*httpsDialerTactic{}
+	ff := &testingx.FakeFiller{}
+	ff.Fill(&input)
+
+	var output []*httpsDialerTactic
+	for tx := range streamTacticsFromSlice(input) {
+		output = append(output, tx)
+	}
+
+	if diff := cmp.Diff(input, output); diff != "" {
+		t.Fatal(diff)
+	}
+}