refactor: flatten and separate (#353)

* refactor(atomicx): move outside the engine package

After merging probe-engine into probe-cli, my impression is that we have
too much unnecessary nesting of packages in this repository.

The idea of this commit and of a bunch of following commits will instead
be to reduce the nesting and simplify the structure.

While there, improve the documentation.

* fix: always use the atomicx package

For consistency, never use sync/atomic and always use ./internal/atomicx
so we can just grep and make sure we're not risking to crash if we make
a subtle mistake on a 32 bit platform.

While there, mention in the contributing guidelines that we want to
always prefer the ./internal/atomicx package over sync/atomic.

* fix(atomicx): remove unnecessary constructor

We don't need a constructor here. The default constructed `&Int64{}`
instance is already usable and the constructor does not add anything to
what we are doing, rather it just creates extra confusion.

* cleanup(atomicx): we are not using Float64

Because atomicx.Float64 is unused, we can safely zap it.

* cleanup(atomicx): simplify impl and improve tests

We can simplify the implementation by using defer and by letting
the Load() method call Add(0).

We can improve tests by making many goroutines updated the
atomic int64 value concurrently.

* refactor(fsx): can live in the ./internal pkg

Let us reduce the amount of nesting. While there, ensure that the
package only exports the bare minimum, and improve the documentation
of the tests, to ease reading the code.

* refactor: move runtimex to ./internal

* refactor: move shellx into the ./internal package

While there, remove unnecessary dependency between packages.

While there, specify in the contributing guidelines that
one should use x/sys/execabs instead of os/exec.

* refactor: move ooapi into the ./internal pkg

* refactor(humanize): move to ./internal and better docs

* refactor: move platform to ./internal

* refactor(randx): move to ./internal

* refactor(multierror): move into the ./internal pkg

* refactor(kvstore): all kvstores in ./internal

Rather than having part of the kvstore inside ./internal/engine/kvstore
and part in ./internal/engine/kvstore.go, let us put every piece of code
that is kvstore related into the ./internal/kvstore package.

* fix(kvstore): always return ErrNoSuchKey on Get() error

It should help to use the kvstore everywhere removing all the
copies that are lingering around the tree.

* sessionresolver: make KVStore mandatory

Simplifies implementation. While there, use the ./internal/kvstore
package rather than having our private implementation.

* fix(ooapi): use the ./internal/kvstore package

* fix(platform): better documentation

CONTRIBUTING.md

@@ -55,6 +55,12 @@ is documented. At the minimum document all the exported symbols.
 Make sure you commit `go.mod` and `go.sum` changes. Make sure you
 run `go mod tidy` to minimize such changes.
 
+## Implementation requirements
+
+Please, use `./internal/atomicx` rather than `atomic/sync`.
+
+Do now use `os/exec`, use `x/sys/execabs`.
+
 ## Code testing requirements
 
 Make sure all tests pass with `go test -race ./...` run from the

cmd/ooniprobe/internal/autorun/autorun_darwin.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/apex/log"
 	"github.com/ooni/probe-cli/v3/cmd/ooniprobe/internal/utils"
-	"github.com/ooni/probe-cli/v3/internal/engine/shellx"
+	"github.com/ooni/probe-cli/v3/internal/shellx"
 	"golang.org/x/sys/execabs"
 	"golang.org/x/sys/unix"
 )

cmd/ooniprobe/internal/nettests/dnscheck.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/dnscheck"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/run"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 // DNSCheck nettest implementation.

cmd/ooniprobe/internal/ooni/ooni.go

@@ -6,7 +6,6 @@ import (
 	"io/ioutil"
 	"os"
 	"os/signal"
-	"sync/atomic"
 	"syscall"
 
 	"github.com/apex/log"
@@ -14,8 +13,10 @@ import (
 	"github.com/ooni/probe-cli/v3/cmd/ooniprobe/internal/database"
 	"github.com/ooni/probe-cli/v3/cmd/ooniprobe/internal/enginex"
 	"github.com/ooni/probe-cli/v3/cmd/ooniprobe/internal/utils"
-	engine "github.com/ooni/probe-cli/v3/internal/engine"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/engine"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/assetsdir"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 	"github.com/pkg/errors"
 	"upper.io/db.v3/lib/sqlbuilder"
 )
@@ -53,11 +54,7 @@ type Probe struct {
 	dbPath     string
 	configPath string
 
-	// We need to use a int32 in order to use the atomic.AddInt32/LoadInt32
+	isTerminated *atomicx.Int64
-	// operations to ensure consistent reads of the variables. We do not use
-	// a 64 bit integer here because that may lead to crashes with 32 bit
-	// OSes as documented in https://golang.org/pkg/sync/atomic/#pkg-note-BUG.
-	isTerminatedAtomicInt int32
 
 	softwareName    string
 	softwareVersion string
@@ -96,13 +93,12 @@ func (p *Probe) TempDir() string {
 // IsTerminated checks to see if the isTerminatedAtomicInt is set to a non zero
 // value and therefore we have received the signal to shutdown the running test
 func (p *Probe) IsTerminated() bool {
-	i := atomic.LoadInt32(&p.isTerminatedAtomicInt)
+	return p.isTerminated.Load() != 0
-	return i != 0
 }
 
 // Terminate interrupts the running context
 func (p *Probe) Terminate() {
-	atomic.AddInt32(&p.isTerminatedAtomicInt, 1)
+	p.isTerminated.Add(1)
 }
 
 // ListenForSignals will listen for SIGINT and SIGTERM. When it receives those
@@ -203,7 +199,7 @@ func (p *Probe) Init(softwareName, softwareVersion string) error {
 // current configuration inside the context. The caller must close
 // the session when done using it, by calling sess.Close().
 func (p *Probe) NewSession(ctx context.Context) (*engine.Session, error) {
-	kvstore, err := engine.NewFileSystemKVStore(
+	kvstore, err := kvstore.NewFS(
 		utils.EngineDir(p.home),
 	)
 	if err != nil {
@@ -234,10 +230,10 @@ func (p *Probe) NewProbeEngine(ctx context.Context) (ProbeEngine, error) {
 // NewProbe creates a new probe instance.
 func NewProbe(configPath string, homePath string) *Probe {
 	return &Probe{
-		home:                  homePath,
+		home:         homePath,
-		config:                &config.Config{},
+		config:       &config.Config{},
-		configPath:            configPath,
+		configPath:   configPath,
-		isTerminatedAtomicInt: 0,
+		isTerminated: &atomicx.Int64{},
 	}
 }
 

internal/atomicx/atomicx.go

@@ -0,0 +1,43 @@
+// Package atomicx extends sync/atomic.
+//
+// Sync/atomic fails when using int64 atomic operations on 32 bit platforms
+// when the access is not aligned. As specified in the documentation, in
+// fact, "it is the caller's responsibility to arrange for 64-bit alignment
+// of 64-bit words accessed atomically". For more information on this
+// issue, see https://golang.org/pkg/sync/atomic/#pkg-note-BUG.
+//
+// As explained in CONTRIBUTING.md, probe-cli SHOULD use this package rather
+// than sync/atomic to avoid these alignment issues on 32 bit.
+//
+// It is of course possible to write atomic code using 64 bit variables on a
+// 32 bit platform, but that's difficult to do correctly. This package
+// provides an easier-to-use interface. We use allocated
+// structures protected by a mutex that encapsulate a int64 value.
+//
+// While there we also added support for atomic float64 operations, again
+// by using structures protected by a mutex variable.
+package atomicx
+
+import "sync"
+
+// Int64 is an int64 with atomic semantics.
+type Int64 struct {
+	// mu provides mutual exclusion.
+	mu sync.Mutex
+
+	// v is the underlying value.
+	v int64
+}
+
+// Add behaves like atomic.AddInt64.
+func (i64 *Int64) Add(delta int64) int64 {
+	i64.mu.Lock()
+	defer i64.mu.Unlock()
+	i64.v += delta
+	return i64.v
+}
+
+// Load behaves like atomic.LoadInt64.
+func (i64 *Int64) Load() (v int64) {
+	return i64.Add(0)
+}

internal/atomicx/atomicx_test.go

@@ -0,0 +1,28 @@
+package atomicx_test
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
+)
+
+func TestInt64(t *testing.T) {
+	v := &atomicx.Int64{}
+	var wg sync.WaitGroup
+	// many goroutines update the value in parallel
+	for i := 0; i < 31; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			v.Add(1)
+		}()
+	}
+	wg.Wait()
+	if v.Add(3) != 34 {
+		t.Fatal("unexpected result")
+	}
+	if v.Load() != 34 {
+		t.Fatal("unexpected result")
+	}
+}

internal/cmd/apitool/main.go

@@ -16,11 +16,11 @@ import (
 	"os"
 
 	"github.com/apex/log"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/httpx"
-	"github.com/ooni/probe-cli/v3/internal/engine/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/probeservices"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/version"
 )
 
@@ -34,9 +34,9 @@ func newclient() probeservices.Client {
 			Logger:     log.Log,
 			UserAgent:  ua,
 		},
-		LoginCalls:    atomicx.NewInt64(),
+		LoginCalls:    &atomicx.Int64{},
-		RegisterCalls: atomicx.NewInt64(),
+		RegisterCalls: &atomicx.Int64{},
-		StateFile:     probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore()),
+		StateFile:     probeservices.NewStateFile(&kvstore.Memory{}),
 	}
 }
 

internal/cmd/jafar/iptables/iptables.go

@@ -4,7 +4,7 @@
 package iptables
 
 import (
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 type shell interface {

internal/cmd/jafar/iptables/iptables_integration_test.go

@@ -17,7 +17,7 @@ import (
 	"github.com/apex/log"
 	"github.com/ooni/probe-cli/v3/internal/cmd/jafar/resolver"
 	"github.com/ooni/probe-cli/v3/internal/cmd/jafar/uncensored"
-	"github.com/ooni/probe-cli/v3/internal/engine/shellx"
+	"github.com/ooni/probe-cli/v3/internal/shellx"
 )
 
 func init() {

internal/cmd/jafar/iptables/iptables_linux.go

@@ -3,8 +3,8 @@
 package iptables
 
 import (
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
-	"github.com/ooni/probe-cli/v3/internal/engine/shellx"
+	"github.com/ooni/probe-cli/v3/internal/shellx"
 )
 
 type linuxShell struct{}

internal/cmd/jafar/main.go

@@ -26,8 +26,8 @@ import (
 	"github.com/ooni/probe-cli/v3/internal/cmd/jafar/resolver"
 	"github.com/ooni/probe-cli/v3/internal/cmd/jafar/tlsproxy"
 	"github.com/ooni/probe-cli/v3/internal/cmd/jafar/uncensored"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
-	"github.com/ooni/probe-cli/v3/internal/engine/shellx"
+	"github.com/ooni/probe-cli/v3/internal/shellx"
 )
 
 var (

internal/cmd/jafar/main_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 
 	"github.com/ooni/probe-cli/v3/internal/cmd/jafar/iptables"
-	"github.com/ooni/probe-cli/v3/internal/engine/shellx"
+	"github.com/ooni/probe-cli/v3/internal/shellx"
 )
 
 func ensureWeStartOverWithIPTables() {

internal/cmd/jafar/uncensored/uncensored.go

@@ -10,7 +10,7 @@ import (
 	"github.com/apex/log"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 // Client is DNS, HTTP, and TCP client.

internal/cmd/miniooni/libminiooni.go

@@ -17,10 +17,11 @@ import (
 
 	"github.com/apex/log"
 	"github.com/ooni/probe-cli/v3/internal/engine"
-	"github.com/ooni/probe-cli/v3/internal/engine/humanizex"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/assetsdir"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/selfcensor"
+	"github.com/ooni/probe-cli/v3/internal/humanize"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/version"
 	"github.com/pborman/getopt/v2"
 )
@@ -348,7 +349,7 @@ func MainWithConfiguration(experimentName string, currentOptions Options) {
 	}
 
 	kvstore2dir := filepath.Join(miniooniDir, "kvstore2")
-	kvstore, err := engine.NewFileSystemKVStore(kvstore2dir)
+	kvstore, err := kvstore.NewFS(kvstore2dir)
 	fatalOnError(err, "cannot create kvstore2 directory")
 
 	tunnelDir := filepath.Join(miniooniDir, "tunnel")
@@ -377,8 +378,8 @@ func MainWithConfiguration(experimentName string, currentOptions Options) {
 	defer func() {
 		sess.Close()
 		log.Infof("whole session: recv %s, sent %s",
-			humanizex.SI(sess.KibiBytesReceived()*1024, "byte"),
+			humanize.SI(sess.KibiBytesReceived()*1024, "byte"),
-			humanizex.SI(sess.KibiBytesSent()*1024, "byte"),
+			humanize.SI(sess.KibiBytesSent()*1024, "byte"),
 		)
 	}()
 	log.Debugf("miniooni temporary directory: %s", sess.TempDir())
@@ -426,8 +427,8 @@ func MainWithConfiguration(experimentName string, currentOptions Options) {
 	experiment := builder.NewExperiment()
 	defer func() {
 		log.Infof("experiment: recv %s, sent %s",
-			humanizex.SI(experiment.KibiBytesReceived()*1024, "byte"),
+			humanize.SI(experiment.KibiBytesReceived()*1024, "byte"),
-			humanizex.SI(experiment.KibiBytesSent()*1024, "byte"),
+			humanize.SI(experiment.KibiBytesSent()*1024, "byte"),
 		)
 	}()
 

internal/cmd/oohelper/internal/client.go

@@ -15,7 +15,7 @@ import (
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/webconnectivity"
 	"github.com/ooni/probe-cli/v3/internal/engine/httpheader"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 	"github.com/ooni/probe-cli/v3/internal/version"
 )
 

internal/cmd/oohelper/internal/fake_test.go

@@ -8,7 +8,7 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 )
 
@@ -19,11 +19,11 @@ type FakeResolver struct {
 }
 
 func NewFakeResolverThatFails() FakeResolver {
-	return FakeResolver{NumFailures: atomicx.NewInt64(), Err: ErrNotFound}
+	return FakeResolver{NumFailures: &atomicx.Int64{}, Err: ErrNotFound}
 }
 
 func NewFakeResolverWithResult(r []string) FakeResolver {
-	return FakeResolver{NumFailures: atomicx.NewInt64(), Result: r}
+	return FakeResolver{NumFailures: &atomicx.Int64{}, Result: r}
 }
 
 var ErrNotFound = &net.DNSError{

internal/cmd/oohelper/oohelper.go

@@ -12,7 +12,7 @@ import (
 	"github.com/apex/log"
 	"github.com/ooni/probe-cli/v3/internal/cmd/oohelper/internal"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 var (

internal/cmd/oohelperd/internal/fake_test.go

@@ -8,7 +8,7 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 )
 
@@ -19,11 +19,11 @@ type FakeResolver struct {
 }
 
 func NewFakeResolverThatFails() FakeResolver {
-	return FakeResolver{NumFailures: atomicx.NewInt64(), Err: ErrNotFound}
+	return FakeResolver{NumFailures: &atomicx.Int64{}, Err: ErrNotFound}
 }
 
 func NewFakeResolverWithResult(r []string) FakeResolver {
-	return FakeResolver{NumFailures: atomicx.NewInt64(), Result: r}
+	return FakeResolver{NumFailures: &atomicx.Int64{}, Result: r}
 }
 
 var ErrNotFound = &net.DNSError{

internal/engine/atomicx/README.md

@@ -1,3 +0,0 @@
-# Package github.com/ooni/probe-engine/atomicx
-
-Atomic int64/float64 that works also on 32 bit platforms.

internal/engine/atomicx/atomicx.go

@@ -1,68 +0,0 @@
-// Package atomicx contains atomic int64/float64 that work also on 32 bit
-// platforms. The main reason for rolling out this package is to avoid potential
-// crashes when using 32 bit devices where we are atomically accessing a 64 bit
-// variable that is not aligned. The solution to this issue is rather crude: use
-// a normal variable and protect it using a normal mutex. While this could be
-// disappointing in general, it seems fine to be done in our context where
-// we mainly use atomic semantics for counting.
-package atomicx
-
-import (
-	"sync"
-)
-
-// Int64 is an int64 with atomic semantics.
-type Int64 struct {
-	mu sync.Mutex
-	v  int64
-}
-
-// NewInt64 creates a new int64 with atomic semantics.
-func NewInt64() *Int64 {
-	return new(Int64)
-}
-
-// Add behaves like atomic.AddInt64
-func (i64 *Int64) Add(delta int64) (newvalue int64) {
-	i64.mu.Lock()
-	i64.v += delta
-	newvalue = i64.v
-	i64.mu.Unlock()
-	return
-}
-
-// Load behaves like atomic.LoadInt64
-func (i64 *Int64) Load() (v int64) {
-	i64.mu.Lock()
-	v = i64.v
-	i64.mu.Unlock()
-	return
-}
-
-// Float64 is an float64 with atomic semantics.
-type Float64 struct {
-	mu sync.Mutex
-	v  float64
-}
-
-// NewFloat64 creates a new float64 with atomic semantics.
-func NewFloat64() *Float64 {
-	return new(Float64)
-}
-
-// Add behaves like AtomicInt64.Add but for float64
-func (f64 *Float64) Add(delta float64) (newvalue float64) {
-	f64.mu.Lock()
-	f64.v += delta
-	newvalue = f64.v
-	f64.mu.Unlock()
-	return
-}
-
-// Load behaves like LoadInt64.Load buf for float64
-func (f64 *Float64) Load() (v float64) {
-	f64.mu.Lock()
-	v = f64.v
-	f64.mu.Unlock()
-	return
-}

internal/engine/atomicx/atomicx_test.go

@@ -1,50 +0,0 @@
-package atomicx_test
-
-import (
-	"testing"
-	"time"
-
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
-)
-
-func TestInt64(t *testing.T) {
-	// TODO(bassosimone): how to write tests with race conditions
-	// and be confident that they're WAI? Here I hope this test is
-	// run with `-race` and I'm doing something that AFAICT will
-	// be flagged as race if we were not be using mutexes.
-	v := atomicx.NewInt64()
-	go func() {
-		v.Add(17)
-	}()
-	go func() {
-		v.Add(14)
-	}()
-	time.Sleep(1 * time.Second)
-	if v.Add(3) != 34 {
-		t.Fatal("unexpected result")
-	}
-	if v.Load() != 34 {
-		t.Fatal("unexpected result")
-	}
-}
-
-func TestFloat64(t *testing.T) {
-	// TODO(bassosimone): how to write tests with race conditions
-	// and be confident that they're WAI? Here I hope this test is
-	// run with `-race` and I'm doing something that AFAICT will
-	// be flagged as race if we were not be using mutexes.
-	v := atomicx.NewFloat64()
-	go func() {
-		v.Add(17.0)
-	}()
-	go func() {
-		v.Add(14.0)
-	}()
-	time.Sleep(1 * time.Second)
-	if r := v.Add(3); r < 33.9 && r > 34.1 {
-		t.Fatal("unexpected result")
-	}
-	if v.Load() < 33.9 && v.Load() > 34.1 {
-		t.Fatal("unexpected result")
-	}
-}

internal/engine/experiment.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"github.com/ooni/probe-cli/v3/internal/engine/geolocate"
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/platform"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/bytecounter"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/dialer"
@@ -168,7 +167,7 @@ func (e *Experiment) newMeasurement(input string) *model.Measurement {
 	}
 	m.AddAnnotation("engine_name", "ooniprobe-engine")
 	m.AddAnnotation("engine_version", version.Version)
-	m.AddAnnotation("platform", platform.Name())
+	m.AddAnnotation("platform", e.session.Platform())
 	return m
 }
 

internal/engine/experiment/dash/dash.go

@@ -15,11 +15,11 @@ import (
 	"time"
 
 	"github.com/montanaflynn/stats"
-	"github.com/ooni/probe-cli/v3/internal/engine/humanizex"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/errorx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/trace"
+	"github.com/ooni/probe-cli/v3/internal/humanize"
 )
 
 const (
@@ -181,7 +181,7 @@ func (r runner) measure(
 		total += current.Received
 		avgspeed := 8 * float64(total) / time.Now().Sub(begin).Seconds()
 		percentage := float64(current.Iteration) / float64(numIterations)
-		message := fmt.Sprintf("streaming: speed: %s", humanizex.SI(avgspeed, "bit/s"))
+		message := fmt.Sprintf("streaming: speed: %s", humanize.SI(avgspeed, "bit/s"))
 		r.callbacks.OnProgress(percentage, message)
 		current.Iteration++
 		speed := float64(current.Received) / float64(current.Elapsed)

internal/engine/experiment/dnscheck/dnscheck.go

@@ -11,15 +11,15 @@ import (
 	"net/url"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/archival"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/trace"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 const (
@@ -31,7 +31,7 @@ const (
 // Endpoints keeps track of repeatedly measured endpoints.
 type Endpoints struct {
 	WaitTime  time.Duration
-	count     uint32
+	count     *atomicx.Int64
 	nextVisit map[string]time.Time
 	mu        sync.Mutex
 }
@@ -48,7 +48,10 @@ func (e *Endpoints) maybeSleep(resolverURL string, logger model.Logger) {
 		return
 	}
 	sleepTime := nextTime.Sub(now)
-	atomic.AddUint32(&e.count, 1)
+	if e.count == nil {
+		e.count = &atomicx.Int64{}
+	}
+	e.count.Add(1)
 	logger.Infof("waiting %v before testing %s again", sleepTime, resolverURL)
 	time.Sleep(sleepTime)
 }

internal/engine/experiment/dnscheck/dnscheck_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"net/url"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -221,7 +220,7 @@ func TestDNSCheckWait(t *testing.T) {
 	}
 	run("dot://one.one.one.one")
 	run("dot://1dot1dot1dot1.cloudflare-dns.com")
-	if atomic.LoadUint32(&endpoints.count) < 1 {
+	if endpoints.count.Load() < 1 {
 		t.Fatal("did not sleep")
 	}
 }

internal/engine/experiment/hhfm/hhfm.go

@@ -17,7 +17,7 @@ import (
 
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/httpheader"
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/randx"
+	"github.com/ooni/probe-cli/v3/internal/randx"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/archival"

internal/engine/experiment/hirl/hirl.go

@@ -11,7 +11,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/randx"
+	"github.com/ooni/probe-cli/v3/internal/randx"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/archival"

internal/engine/experiment/ndt7/ndt7.go

@@ -11,10 +11,10 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/humanizex"
 	"github.com/ooni/probe-cli/v3/internal/engine/internal/mlablocatev2"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
+	"github.com/ooni/probe-cli/v3/internal/humanize"
 )
 
 const (
@@ -127,7 +127,7 @@ func (m *Measurer) doDownload(
 			// 50% of the whole experiment, hence the `/2.0`.
 			percentage := elapsed / paramMaxRuntimeUpperBound / 2.0
 			speed := float64(count) * 8.0 / elapsed
-			message := fmt.Sprintf(" download: speed %s", humanizex.SI(
+			message := fmt.Sprintf(" download: speed %s", humanize.SI(
 				float64(speed), "bit/s"))
 			tk.Summary.Download = speed / 1e03 /* bit/s => kbit/s */
 			callbacks.OnProgress(percentage, message)
@@ -197,7 +197,7 @@ func (m *Measurer) doUpload(
 			// the whole experiment, hence `0.5 +` and `/2.0`.
 			percentage := 0.5 + elapsed/paramMaxRuntimeUpperBound/2.0
 			speed := float64(count) * 8.0 / elapsed
-			message := fmt.Sprintf("   upload: speed %s", humanizex.SI(
+			message := fmt.Sprintf("   upload: speed %s", humanize.SI(
 				float64(speed), "bit/s"))
 			tk.Summary.Upload = speed / 1e03 /* bit/s => kbit/s */
 			callbacks.OnProgress(percentage, message)

internal/engine/experiment/psiphon/psiphon_test.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/apex/log"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/psiphon"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/internal/mockable"
@@ -83,7 +83,7 @@ func TestRunWillPrintSomethingWithCancelledContext(t *testing.T) {
 		time.Sleep(2 * time.Second)
 		cancel() // fail after we've given the printer a chance to run
 	}
-	observer := observerCallbacks{progress: atomicx.NewInt64()}
+	observer := observerCallbacks{progress: &atomicx.Int64{}}
 	err := measurer.Run(ctx, newfakesession(), measurement, observer)
 	if !errors.Is(err, context.Canceled) {
 		t.Fatal("expected another error here")

internal/engine/experiment/telegram/telegram_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 
 	"github.com/apex/log"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/telegram"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/internal/mockable"
@@ -271,9 +271,9 @@ func TestUpdateWebWithMixedResults(t *testing.T) {
 }
 
 func TestWeConfigureWebChecksToFailOnHTTPError(t *testing.T) {
-	called := atomicx.NewInt64()
+	called := &atomicx.Int64{}
-	failOnErrorHTTPS := atomicx.NewInt64()
+	failOnErrorHTTPS := &atomicx.Int64{}
-	failOnErrorHTTP := atomicx.NewInt64()
+	failOnErrorHTTP := &atomicx.Int64{}
 	measurer := telegram.Measurer{
 		Config: telegram.Config{},
 		Getter: func(ctx context.Context, g urlgetter.Getter) (urlgetter.TestKeys, error) {

internal/engine/experiment/tlstool/internal/splitter_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/tlstool/internal"
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/randx"
+	"github.com/ooni/probe-cli/v3/internal/randx"
 )
 
 func TestSplitter84restSmall(t *testing.T) {

internal/engine/experiment/tlstool/tlstool.go

@@ -19,7 +19,7 @@ import (
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/archival"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 const (

internal/engine/experiment/tor/tor.go

@@ -12,14 +12,14 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/httpheader"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/netxlogger"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/oonidatamodel"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/oonitemplates"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/errorx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 const (
@@ -264,7 +264,7 @@ func newResultsCollector(
 ) *resultsCollector {
 	rc := &resultsCollector{
 		callbacks:     callbacks,
-		completed:     atomicx.NewInt64(),
+		completed:     &atomicx.Int64{},
 		measurement:   measurement,
 		sess:          sess,
 		targetresults: make(map[string]TargetResults),

internal/engine/experiment/urlgetter/runner.go

@@ -13,7 +13,7 @@ import (
 	"github.com/ooni/probe-cli/v3/internal/engine/httpheader"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/errorx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 const httpRequestFailed = "http_request_failed"

internal/engine/experiment/urlgetter/runner_test.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/httpheader"
 )
@@ -235,7 +235,7 @@ func TestRunnerHTTPCannotReadBodyWinsOver400(t *testing.T) {
 
 func TestRunnerWeCanForceUserAgent(t *testing.T) {
 	expected := "antani/1.23.4-dev"
-	found := atomicx.NewInt64()
+	found := &atomicx.Int64{}
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.Header.Get("User-Agent") == expected {
 			found.Add(1)
@@ -262,7 +262,7 @@ func TestRunnerWeCanForceUserAgent(t *testing.T) {
 
 func TestRunnerDefaultUserAgent(t *testing.T) {
 	expected := httpheader.UserAgent()
-	found := atomicx.NewInt64()
+	found := &atomicx.Int64{}
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.Header.Get("User-Agent") == expected {
 			found.Add(1)

internal/engine/experiment/webconnectivity/endpoints.go

@@ -4,7 +4,7 @@ import (
 	"net"
 	"net/url"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 // EndpointInfo describes a TCP/TLS endpoint.

internal/engine/experiment/webconnectivity/endpoints_test.go

@@ -6,12 +6,12 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/webconnectivity"
 )
 
 func TestNewEndpointPortPanicsWithInvalidScheme(t *testing.T) {
-	counter := atomicx.NewInt64()
+	counter := &atomicx.Int64{}
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
@@ -30,7 +30,7 @@ func TestNewEndpointPortPanicsWithInvalidScheme(t *testing.T) {
 }
 
 func TestNewEndpointPortPanicsWithInvalidHost(t *testing.T) {
-	counter := atomicx.NewInt64()
+	counter := &atomicx.Int64{}
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {

internal/engine/experiment/webconnectivity/httpanalysis_test.go

@@ -6,7 +6,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/webconnectivity"
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/randx"
+	"github.com/ooni/probe-cli/v3/internal/randx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/archival"
 )
 

internal/engine/experiment/whatsapp/whatsapp.go

@@ -15,7 +15,7 @@ import (
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/internal/httpfailure"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 const (

internal/engine/experiment/whatsapp/whatsapp_test.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/apex/log"
 	"github.com/google/go-cmp/cmp"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/whatsapp"
 	"github.com/ooni/probe-cli/v3/internal/engine/internal/httpfailure"
@@ -555,7 +555,7 @@ func TestTestKeysOnlyWebHTTPFailureTooManyURLs(t *testing.T) {
 }
 
 func TestWeConfigureWebChecksCorrectly(t *testing.T) {
-	called := atomicx.NewInt64()
+	called := &atomicx.Int64{}
 	emptyConfig := urlgetter.Config{}
 	configWithFailOnHTTPError := urlgetter.Config{FailOnHTTPError: true}
 	configWithNoFollowRedirects := urlgetter.Config{NoFollowRedirects: true}

internal/engine/geolocate/geolocate.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 	"github.com/ooni/probe-cli/v3/internal/version"
 )
 

internal/engine/geolocate/iplookup.go

@@ -9,7 +9,7 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/multierror"
+	"github.com/ooni/probe-cli/v3/internal/multierror"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 )
 

internal/engine/humanizex/humanizex_test.go

@@ -1,34 +0,0 @@
-package humanizex_test
-
-import (
-	"testing"
-
-	"github.com/ooni/probe-cli/v3/internal/engine/humanizex"
-)
-
-func TestGood(t *testing.T) {
-	if humanizex.SI(128, "bit/s") != "128  bit/s" {
-		t.Fatal("unexpected result")
-	}
-	if humanizex.SI(1280, "bit/s") != "  1 kbit/s" {
-		t.Fatal("unexpected result")
-	}
-	if humanizex.SI(12800, "bit/s") != " 13 kbit/s" {
-		t.Fatal("unexpected result")
-	}
-	if humanizex.SI(128000, "bit/s") != "128 kbit/s" {
-		t.Fatal("unexpected result")
-	}
-	if humanizex.SI(1280000, "bit/s") != "  1 Mbit/s" {
-		t.Fatal("unexpected result")
-	}
-	if humanizex.SI(12800000, "bit/s") != " 13 Mbit/s" {
-		t.Fatal("unexpected result")
-	}
-	if humanizex.SI(128000000, "bit/s") != "128 Mbit/s" {
-		t.Fatal("unexpected result")
-	}
-	if humanizex.SI(1280000000, "bit/s") != "  1 Gbit/s" {
-		t.Fatal("unexpected result")
-	}
-}

internal/engine/inputloader.go

@@ -8,8 +8,8 @@ import (
 	"io/fs"
 
 	"github.com/apex/log"
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/fsx"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
+	"github.com/ooni/probe-cli/v3/internal/fsx"
 )
 
 // These errors are returned by the InputLoader.
@@ -154,7 +154,7 @@ func (il *InputLoader) loadLocal() ([]model.URLInfo, error) {
 		inputs = append(inputs, model.URLInfo{URL: input})
 	}
 	for _, filepath := range il.SourceFiles {
-		extra, err := il.readfile(filepath, fsx.Open)
+		extra, err := il.readfile(filepath, fsx.OpenFile)
 		if err != nil {
 			return nil, err
 		}

internal/engine/inputloader_network_test.go

@@ -6,8 +6,8 @@ import (
 
 	"github.com/apex/log"
 	engine "github.com/ooni/probe-cli/v3/internal/engine"
-	"github.com/ooni/probe-cli/v3/internal/engine/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 )
 
 func TestInputLoaderInputOrQueryBackendWithNoInput(t *testing.T) {
@@ -19,7 +19,7 @@ func TestInputLoaderInputOrQueryBackendWithNoInput(t *testing.T) {
 			Address: "https://ams-pg-test.ooni.org/",
 			Type:    "https",
 		}},
-		KVStore:         kvstore.NewMemoryKeyValueStore(),
+		KVStore:         &kvstore.Memory{},
 		Logger:          log.Log,
 		SoftwareName:    "miniooni",
 		SoftwareVersion: "0.1.0-dev",

internal/engine/inputloader_test.go

@@ -11,8 +11,8 @@ import (
 
 	"github.com/apex/log"
 	"github.com/google/go-cmp/cmp"
-	"github.com/ooni/probe-cli/v3/internal/engine/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 )
 
 func TestInputLoaderInputNoneWithStaticInputs(t *testing.T) {
@@ -237,7 +237,7 @@ func TestInputLoaderInputOrQueryBackendWithInput(t *testing.T) {
 
 func TestInputLoaderInputOrQueryBackendWithNoInputAndCancelledContext(t *testing.T) {
 	sess, err := NewSession(context.Background(), SessionConfig{
-		KVStore:         kvstore.NewMemoryKeyValueStore(),
+		KVStore:         &kvstore.Memory{},
 		Logger:          log.Log,
 		SoftwareName:    "miniooni",
 		SoftwareVersion: "0.1.0-dev",

internal/engine/inputprocessor.go

@@ -2,7 +2,6 @@ package engine
 
 import (
 	"context"
-	"sync/atomic"
 	"time"
 
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
@@ -68,11 +67,6 @@ type InputProcessor struct {
 	// Submitter is the code that will submit measurements
 	// to the OONI collector.
 	Submitter InputProcessorSubmitterWrapper
-
-	// terminatedByMaxRuntime is an internal atomic variabile
-	// incremented when we're terminated by MaxRuntime. We
-	// only use this variable when testing.
-	terminatedByMaxRuntime int32
 }
 
 // InputProcessorSaverWrapper is InputProcessor's
@@ -129,29 +123,41 @@ func (ipsw inputProcessorSubmitterWrapper) Submit(
 // though is free to choose different policies by configuring
 // the Experiment, Submitter, and Saver fields properly.
 func (ip *InputProcessor) Run(ctx context.Context) error {
+	_, err := ip.run(ctx)
+	return err
+}
+
+// These are the reasons why run could stop.
+const (
+	stopNormal = (1 << iota)
+	stopMaxRuntime
+)
+
+// run is like Run but, in addition to returning an error, it
+// also returns the reason why we stopped.
+func (ip *InputProcessor) run(ctx context.Context) (int, error) {
 	start := time.Now()
 	for idx, url := range ip.Inputs {
 		if ip.MaxRuntime > 0 && time.Since(start) > ip.MaxRuntime {
-			atomic.AddInt32(&ip.terminatedByMaxRuntime, 1)
+			return stopMaxRuntime, nil
-			return nil
 		}
 		input := url.URL
 		meas, err := ip.Experiment.MeasureWithContext(ctx, idx, input)
 		if err != nil {
-			return err
+			return 0, err
 		}
 		meas.AddAnnotations(ip.Annotations)
 		meas.Options = ip.Options
 		err = ip.Submitter.Submit(ctx, idx, meas)
 		if err != nil {
-			return err
+			return 0, err
 		}
 		// Note: must be after submission because submission modifies
 		// the measurement to include the report ID.
 		err = ip.Saver.SaveMeasurement(idx, meas)
 		if err != nil {
-			return err
+			return 0, err
 		}
 	}
-	return nil
+	return stopNormal, nil
 }

internal/engine/inputprocessor_test.go

@@ -150,10 +150,11 @@ func TestInputProcessorGood(t *testing.T) {
 		Submitter: NewInputProcessorSubmitterWrapper(submitter),
 	}
 	ctx := context.Background()
-	if err := ip.Run(ctx); err != nil {
+	reason, err := ip.run(ctx)
+	if err != nil {
 		t.Fatal(err)
 	}
-	if ip.terminatedByMaxRuntime > 0 {
+	if reason != stopNormal {
 		t.Fatal("terminated by max runtime!?")
 	}
 	if len(fipe.M) != 2 || len(saver.M) != 2 || len(submitter.M) != 2 {
@@ -192,10 +193,11 @@ func TestInputProcessorMaxRuntime(t *testing.T) {
 		Submitter:  NewInputProcessorSubmitterWrapper(submitter),
 	}
 	ctx := context.Background()
-	if err := ip.Run(ctx); err != nil {
+	reason, err := ip.run(ctx)
+	if err != nil {
 		t.Fatal(err)
 	}
-	if ip.terminatedByMaxRuntime <= 0 {
+	if reason != stopMaxRuntime {
 		t.Fatal("not terminated by max runtime")
 	}
 }

internal/engine/internal/fsx/fsx_test.go

@@ -1,76 +0,0 @@
-package fsx_test
-
-import (
-	"errors"
-	"io/fs"
-	"os"
-	"sync/atomic"
-	"syscall"
-	"testing"
-
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/fsx"
-)
-
-var StateBaseDir = "./testdata/"
-
-type FailingStatFS struct {
-	CloseCount *int32
-}
-
-type FailingStatFile struct {
-	CloseCount *int32
-}
-
-var errStatFailed = errors.New("stat failed")
-
-func (FailingStatFile) Stat() (os.FileInfo, error) {
-	return nil, errStatFailed
-}
-
-func (f FailingStatFS) Open(pathname string) (fs.File, error) {
-	return FailingStatFile(f), nil
-}
-
-func (fs FailingStatFile) Close() error {
-	if fs.CloseCount != nil {
-		atomic.AddInt32(fs.CloseCount, 1)
-	}
-	return nil
-}
-
-func (FailingStatFile) Read([]byte) (int, error) {
-	return 0, nil
-}
-
-func TestOpenWithFailingStat(t *testing.T) {
-	var count int32
-	_, err := fsx.OpenWithFS(FailingStatFS{CloseCount: &count}, StateBaseDir+"testfile.txt")
-	if !errors.Is(err, errStatFailed) {
-		t.Errorf("expected error with invalid FS: %+v", err)
-	}
-	if count != 1 {
-		t.Error("expected counter to be equal to 1")
-	}
-}
-
-func TestOpenNonexistentFile(t *testing.T) {
-	_, err := fsx.Open(StateBaseDir + "invalidtestfile.txt")
-	if !errors.Is(err, syscall.ENOENT) {
-		t.Errorf("not the error we expected")
-	}
-}
-
-func TestOpenDirectoryShouldFail(t *testing.T) {
-	_, err := fsx.Open(StateBaseDir)
-	if !errors.Is(err, syscall.EISDIR) {
-		t.Fatalf("not the error we expected: %+v", err)
-	}
-}
-
-func TestOpeningExistingFileShouldWork(t *testing.T) {
-	file, err := fsx.Open(StateBaseDir + "testfile.txt")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer file.Close()
-}

internal/engine/internal/mockable/mockable.go

@@ -6,9 +6,9 @@ import (
 	"net/http"
 	"net/url"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/probeservices"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 )
 
 // Session allows to mock sessions.
@@ -67,7 +67,7 @@ func (sess *Session) FetchURLList(
 
 // KeyValueStore returns the configured key-value store.
 func (sess *Session) KeyValueStore() model.KeyValueStore {
-	return kvstore.NewMemoryKeyValueStore()
+	return &kvstore.Memory{}
 }
 
 // Logger implements ExperimentSession.Logger

internal/engine/internal/sessionresolver/integration_test.go

@@ -5,13 +5,16 @@ import (
 	"testing"
 
 	"github.com/ooni/probe-cli/v3/internal/engine/internal/sessionresolver"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 )
 
 func TestSessionResolverGood(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skip test in short mode")
 	}
-	reso := &sessionresolver.Resolver{}
+	reso := &sessionresolver.Resolver{
+		KVStore: &kvstore.Memory{},
+	}
 	defer reso.CloseIdleConnections()
 	if reso.Network() != "sessionresolver" {
 		t.Fatal("unexpected Network")

internal/engine/internal/sessionresolver/memkvstore.go

@@ -1,43 +0,0 @@
-package sessionresolver
-
-import (
-	"errors"
-	"fmt"
-	"sync"
-)
-
-func (r *Resolver) kvstore() KVStore {
-	defer r.mu.Unlock()
-	r.mu.Lock()
-	if r.KVStore == nil {
-		r.KVStore = &memkvstore{}
-	}
-	return r.KVStore
-}
-
-var errMemkvstoreNotFound = errors.New("memkvstore: not found")
-
-type memkvstore struct {
-	m  map[string][]byte
-	mu sync.Mutex
-}
-
-func (kvs *memkvstore) Get(key string) ([]byte, error) {
-	defer kvs.mu.Unlock()
-	kvs.mu.Lock()
-	out, good := kvs.m[key]
-	if !good {
-		return nil, fmt.Errorf("%w: %s", errMemkvstoreNotFound, key)
-	}
-	return out, nil
-}
-
-func (kvs *memkvstore) Set(key string, value []byte) error {
-	defer kvs.mu.Unlock()
-	kvs.mu.Lock()
-	if kvs.m == nil {
-		kvs.m = make(map[string][]byte)
-	}
-	kvs.m[key] = value
-	return nil
-}

internal/engine/internal/sessionresolver/memkvstore_test.go

@@ -1,47 +0,0 @@
-package sessionresolver
-
-import (
-	"errors"
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-)
-
-func TestKVStoreCustom(t *testing.T) {
-	kvs := &memkvstore{}
-	reso := &Resolver{KVStore: kvs}
-	o := reso.kvstore()
-	if o != kvs {
-		t.Fatal("not the kvstore we expected")
-	}
-}
-
-func TestMemkvstoreGetNotFound(t *testing.T) {
-	reso := &Resolver{}
-	key := "antani"
-	out, err := reso.kvstore().Get(key)
-	if !errors.Is(err, errMemkvstoreNotFound) {
-		t.Fatal("not the error we expected", err)
-	}
-	if out != nil {
-		t.Fatal("expected nil here")
-	}
-}
-
-func TestMemkvstoreRoundTrip(t *testing.T) {
-	reso := &Resolver{}
-	key := []string{"antani", "mascetti"}
-	value := [][]byte{[]byte(`mascetti`), []byte(`antani`)}
-	for idx := 0; idx < 2; idx++ {
-		if err := reso.kvstore().Set(key[idx], value[idx]); err != nil {
-			t.Fatal(err)
-		}
-		out, err := reso.kvstore().Get(key[idx])
-		if err != nil {
-			t.Fatal(err)
-		}
-		if diff := cmp.Diff(value[idx], out); diff != "" {
-			t.Fatal(diff)
-		}
-	}
-}

internal/engine/internal/sessionresolver/sessionresolver.go

@@ -33,9 +33,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/multierror"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/bytecounter"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/multierror"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 // Resolver is the session resolver. Resolver will try to use
@@ -45,6 +45,9 @@ import (
 // and therefore we can generally give preference to underlying
 // DoT/DoH resolvers that work better.
 //
+// Make sure you fill the mandatory fields (indicated below)
+// before using this data structure.
+//
 // You MUST NOT modify public fields of this structure once it
 // has been created, because that MAY lead to data races.
 //
@@ -57,10 +60,9 @@ type Resolver struct {
 	// field is not set, then we won't count the bytes.
 	ByteCounter *bytecounter.Counter
 
-	// KVStore is the optional key-value store where you
+	// KVStore is the MANDATORY key-value store where you
 	// want us to write statistics about which resolver is
-	// working better in your network. If this field is
+	// working better in your network.
-	// not set, then we'll use a in-memory store.
 	KVStore KVStore
 
 	// Logger is the optional logger you want us to use

internal/engine/internal/sessionresolver/sessionresolver_test.go

@@ -6,11 +6,12 @@ import (
 	"net"
 	"net/url"
 	"strings"
-	"sync/atomic"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/multierror"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
+	"github.com/ooni/probe-cli/v3/internal/multierror"
 )
 
 func TestNetworkWorks(t *testing.T) {
@@ -30,7 +31,7 @@ func TestAddressWorks(t *testing.T) {
 func TestTypicalUsageWithFailure(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel() // fail immediately
-	reso := &Resolver{}
+	reso := &Resolver{KVStore: &kvstore.Memory{}}
 	addrs, err := reso.LookupHost(ctx, "ooni.org")
 	if !errors.Is(err, ErrLookupHost) {
 		t.Fatal("not the error we expected", err)
@@ -82,6 +83,7 @@ func TestTypicalUsageWithSuccess(t *testing.T) {
 	expected := []string{"8.8.8.8", "8.8.4.4"}
 	ctx := context.Background()
 	reso := &Resolver{
+		KVStore: &kvstore.Memory{},
 		dnsClientMaker: &fakeDNSClientMaker{
 			reso: &FakeResolver{Data: expected},
 		},
@@ -252,7 +254,7 @@ func TestMaybeConfusionManyEntries(t *testing.T) {
 
 func TestResolverWorksWithProxy(t *testing.T) {
 	var (
-		works      int32
+		works      = &atomicx.Int64{}
 		startuperr = make(chan error)
 		listench   = make(chan net.Listener)
 		done       = make(chan interface{})
@@ -273,7 +275,7 @@ func TestResolverWorksWithProxy(t *testing.T) {
 				// shutdown by the main goroutine.
 				return
 			}
-			atomic.AddInt32(&works, 1)
+			works.Add(1)
 			conn.Close()
 		}
 	}()
@@ -283,10 +285,13 @@ func TestResolverWorksWithProxy(t *testing.T) {
 	}
 	listener := <-listench
 	// use the proxy
-	reso := &Resolver{ProxyURL: &url.URL{
+	reso := &Resolver{
-		Scheme: "socks5",
+		ProxyURL: &url.URL{
-		Host:   listener.Addr().String(),
+			Scheme: "socks5",
-	}}
+			Host:   listener.Addr().String(),
+		},
+		KVStore: &kvstore.Memory{},
+	}
 	ctx := context.Background()
 	addrs, err := reso.LookupHost(ctx, "ooni.org")
 	// cleanly shutdown the listener
@@ -299,7 +304,7 @@ func TestResolverWorksWithProxy(t *testing.T) {
 	if addrs != nil {
 		t.Fatal("expected nil addrs")
 	}
-	if works < 1 {
+	if works.Load() < 1 {
 		t.Fatal("expected to see a positive number of entries here")
 	}
 }

internal/engine/internal/sessionresolver/state.go

@@ -18,9 +18,15 @@ type resolverinfo struct {
 	Score float64
 }
 
+// ErrNilKVStore indicates that the KVStore is nil.
+var ErrNilKVStore = errors.New("sessionresolver: kvstore is nil")
+
 // readstate reads the resolver state from disk
 func (r *Resolver) readstate() ([]*resolverinfo, error) {
-	data, err := r.kvstore().Get(storekey)
+	if r.KVStore == nil {
+		return nil, ErrNilKVStore
+	}
+	data, err := r.KVStore.Get(storekey)
 	if err != nil {
 		return nil, err
 	}
@@ -85,9 +91,12 @@ func (r *Resolver) readstatedefault() []*resolverinfo {
 
 // writestate writes the state on the kvstore.
 func (r *Resolver) writestate(ri []*resolverinfo) error {
+	if r.KVStore == nil {
+		return ErrNilKVStore
+	}
 	data, err := r.getCodec().Encode(ri)
 	if err != nil {
 		return err
 	}
-	return r.kvstore().Set(storekey, data)
+	return r.KVStore.Set(storekey, data)
 }

internal/engine/internal/sessionresolver/state_test.go

@@ -3,12 +3,25 @@ package sessionresolver
 import (
 	"errors"
 	"testing"
+
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 )
 
-func TestReadStateNothingInKVStore(t *testing.T) {
+func TestReadStateNoKVStore(t *testing.T) {
-	reso := &Resolver{KVStore: &memkvstore{}}
+	reso := &Resolver{}
 	out, err := reso.readstate()
-	if !errors.Is(err, errMemkvstoreNotFound) {
+	if !errors.Is(err, ErrNilKVStore) {
+		t.Fatal("not the error we expected", err)
+	}
+	if out != nil {
+		t.Fatal("expected nil here")
+	}
+}
+
+func TestReadStateNothingInKVStore(t *testing.T) {
+	reso := &Resolver{KVStore: &kvstore.Memory{}}
+	out, err := reso.readstate()
+	if !errors.Is(err, kvstore.ErrNoSuchKey) {
 		t.Fatal("not the error we expected", err)
 	}
 	if out != nil {
@@ -19,7 +32,7 @@ func TestReadStateNothingInKVStore(t *testing.T) {
 func TestReadStateDecodeError(t *testing.T) {
 	errMocked := errors.New("mocked error")
 	reso := &Resolver{
-		KVStore: &memkvstore{},
+		KVStore: &kvstore.Memory{},
 		codec:   &FakeCodec{DecodeErr: errMocked},
 	}
 	if err := reso.KVStore.Set(storekey, []byte(`[]`)); err != nil {
@@ -35,9 +48,9 @@ func TestReadStateDecodeError(t *testing.T) {
 }
 
 func TestReadStateAndPruneReadStateError(t *testing.T) {
-	reso := &Resolver{KVStore: &memkvstore{}}
+	reso := &Resolver{KVStore: &kvstore.Memory{}}
 	out, err := reso.readstateandprune()
-	if !errors.Is(err, errMemkvstoreNotFound) {
+	if !errors.Is(err, kvstore.ErrNoSuchKey) {
 		t.Fatal("not the error we expected", err)
 	}
 	if out != nil {
@@ -46,7 +59,7 @@ func TestReadStateAndPruneReadStateError(t *testing.T) {
 }
 
 func TestReadStateAndPruneWithUnsupportedEntries(t *testing.T) {
-	reso := &Resolver{KVStore: &memkvstore{}}
+	reso := &Resolver{KVStore: &kvstore.Memory{}}
 	var in []*resolverinfo
 	in = append(in, &resolverinfo{})
 	if err := reso.writestate(in); err != nil {
@@ -62,7 +75,7 @@ func TestReadStateAndPruneWithUnsupportedEntries(t *testing.T) {
 }
 
 func TestReadStateDefaultWithMissingEntries(t *testing.T) {
-	reso := &Resolver{KVStore: &memkvstore{}}
+	reso := &Resolver{KVStore: &kvstore.Memory{}}
 	// let us simulate that we have just one entry here
 	existingURL := "https://dns.google/dns-query"
 	existingScore := 0.88
@@ -100,12 +113,27 @@ func TestReadStateDefaultWithMissingEntries(t *testing.T) {
 	}
 }
 
+func TestWriteStateNoKVStore(t *testing.T) {
+	reso := &Resolver{}
+	existingURL := "https://dns.google/dns-query"
+	existingScore := 0.88
+	var in []*resolverinfo
+	in = append(in, &resolverinfo{
+		URL:   existingURL,
+		Score: existingScore,
+	})
+	if err := reso.writestate(in); !errors.Is(err, ErrNilKVStore) {
+		t.Fatal("not the error we expected", err)
+	}
+}
+
 func TestWriteStateCannotSerialize(t *testing.T) {
 	errMocked := errors.New("mocked error")
 	reso := &Resolver{
 		codec: &FakeCodec{
 			EncodeErr: errMocked,
 		},
+		KVStore: &kvstore.Memory{},
 	}
 	existingURL := "https://dns.google/dns-query"
 	existingScore := 0.88

internal/engine/kvstore.go

@@ -1,13 +1,5 @@
 package engine
 
-import (
-	"bytes"
-	"os"
-	"path/filepath"
-
-	"github.com/rogpeppe/go-internal/lockedfile"
-)
-
 // KVStore is a simple, atomic key-value store. The user of
 // probe-engine should supply an implementation of this interface,
 // which will be used by probe-engine to store specific data.
@@ -15,30 +7,3 @@ type KVStore interface {
 	Get(key string) (value []byte, err error)
 	Set(key string, value []byte) (err error)
 }
-
-// FileSystemKVStore is a directory based KVStore
-type FileSystemKVStore struct {
-	basedir string
-}
-
-// NewFileSystemKVStore creates a new FileSystemKVStore.
-func NewFileSystemKVStore(basedir string) (kvs *FileSystemKVStore, err error) {
-	if err = os.MkdirAll(basedir, 0700); err == nil {
-		kvs = &FileSystemKVStore{basedir: basedir}
-	}
-	return
-}
-
-func (kvs *FileSystemKVStore) filename(key string) string {
-	return filepath.Join(kvs.basedir, key)
-}
-
-// Get returns the specified key's value
-func (kvs *FileSystemKVStore) Get(key string) ([]byte, error) {
-	return lockedfile.Read(kvs.filename(key))
-}
-
-// Set sets the value of a specific key
-func (kvs *FileSystemKVStore) Set(key string, value []byte) error {
-	return lockedfile.Write(kvs.filename(key), bytes.NewReader(value), 0600)
-}

internal/engine/kvstore/kvstore.go

@@ -1,44 +0,0 @@
-// Package kvstore contains key-value stores
-package kvstore
-
-import (
-	"errors"
-	"sync"
-)
-
-// MemoryKeyValueStore is an in-memory key-value store
-type MemoryKeyValueStore struct {
-	m  map[string][]byte
-	mu sync.Mutex
-}
-
-// NewMemoryKeyValueStore creates a new in-memory key-value store
-func NewMemoryKeyValueStore() *MemoryKeyValueStore {
-	return &MemoryKeyValueStore{
-		m: make(map[string][]byte),
-	}
-}
-
-// Get returns a key from the key value store
-func (kvs *MemoryKeyValueStore) Get(key string) ([]byte, error) {
-	var (
-		err   error
-		ok    bool
-		value []byte
-	)
-	kvs.mu.Lock()
-	defer kvs.mu.Unlock()
-	value, ok = kvs.m[key]
-	if !ok {
-		err = errors.New("no such key")
-	}
-	return value, err
-}
-
-// Set sets a key into the key value store
-func (kvs *MemoryKeyValueStore) Set(key string, value []byte) error {
-	kvs.mu.Lock()
-	defer kvs.mu.Unlock()
-	kvs.m[key] = value
-	return nil
-}

internal/engine/kvstore_test.go

@@ -1,31 +0,0 @@
-package engine
-
-import (
-	"bytes"
-	"path/filepath"
-	"testing"
-)
-
-func TestKVStoreIntegration(t *testing.T) {
-	var (
-		err     error
-		kvstore KVStore
-	)
-	kvstore, err = NewFileSystemKVStore(
-		filepath.Join("testdata", "kvstore2"),
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-	value := []byte("foobar")
-	if err := kvstore.Set("antani", value); err != nil {
-		t.Fatal(err)
-	}
-	ovalue, err := kvstore.Get("antani")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if !bytes.Equal(ovalue, value) {
-		t.Fatal("invalid value")
-	}
-}

internal/engine/legacy/netx/dialid/dialid.go

@@ -3,12 +3,12 @@ package dialid
 import (
 	"context"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 )
 
 type contextkey struct{}
 
-var id = atomicx.NewInt64()
+var id = &atomicx.Int64{}
 
 // WithDialID returns a copy of ctx with DialID
 func WithDialID(ctx context.Context) context.Context {

internal/engine/legacy/netx/handlers/handlers.go

@@ -7,7 +7,7 @@ import (
 	"sync"
 
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/netx/modelx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 type stdoutHandler struct{}

internal/engine/legacy/netx/oldhttptransport/tracetripper.go

@@ -10,7 +10,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/netx/connid"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/netx/dialid"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/netx/modelx"
@@ -28,7 +28,7 @@ type TraceTripper struct {
 // NewTraceTripper creates a new Transport.
 func NewTraceTripper(roundTripper http.RoundTripper) *TraceTripper {
 	return &TraceTripper{
-		readAllErrs:  atomicx.NewInt64(),
+		readAllErrs:  &atomicx.Int64{},
 		readAll:      ioutil.ReadAll,
 		roundTripper: roundTripper,
 	}

internal/engine/legacy/netx/transactionid/transactionid.go

@@ -4,12 +4,12 @@ package transactionid
 import (
 	"context"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 )
 
 type contextkey struct{}
 
-var id = atomicx.NewInt64()
+var id = &atomicx.Int64{}
 
 // WithTransactionID returns a copy of ctx with TransactionID
 func WithTransactionID(ctx context.Context) context.Context {

internal/engine/legacy/oonitemplates/oonitemplates.go

@@ -20,11 +20,11 @@ import (
 	"time"
 
 	goptlib "git.torproject.org/pluggable-transports/goptlib.git"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/netx/handlers"
 	"github.com/ooni/probe-cli/v3/internal/engine/legacy/netx/modelx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 	"gitlab.com/yawning/obfs4.git/transports"
 	obfs4base "gitlab.com/yawning/obfs4.git/transports/base"
 )
@@ -37,7 +37,7 @@ type channelHandler struct {
 func newChannelHandler(ch chan<- modelx.Measurement) *channelHandler {
 	return &channelHandler{
 		ch:         ch,
-		lateWrites: atomicx.NewInt64(),
+		lateWrites: &atomicx.Int64{},
 	}
 }
 

internal/engine/netx/bytecounter/bytecounter.go

@@ -1,6 +1,6 @@
 package bytecounter
 
-import "github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+import "github.com/ooni/probe-cli/v3/internal/atomicx"
 
 // Counter counts bytes sent and received.
 type Counter struct {
@@ -10,7 +10,7 @@ type Counter struct {
 
 // New creates a new Counter.
 func New() *Counter {
-	return &Counter{Received: atomicx.NewInt64(), Sent: atomicx.NewInt64()}
+	return &Counter{Received: &atomicx.Int64{}, Sent: &atomicx.Int64{}}
 }
 
 // CountBytesSent adds count to the bytes sent counter.

internal/engine/netx/netx.go

@@ -39,7 +39,7 @@ import (
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/resolver"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/selfcensor"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/trace"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 // Logger is the logger assumed by this package

internal/engine/netx/resolver/bogon.go

@@ -5,7 +5,7 @@ import (
 	"net"
 
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/errorx"
-	"github.com/ooni/probe-cli/v3/internal/engine/runtimex"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 var privateIPBlocks []*net.IPNet

internal/engine/netx/resolver/fake_test.go

@@ -6,7 +6,7 @@ import (
 	"net"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 )
 
 type FakeDialer struct {
@@ -109,11 +109,11 @@ type FakeResolver struct {
 }
 
 func NewFakeResolverThatFails() FakeResolver {
-	return FakeResolver{NumFailures: atomicx.NewInt64(), Err: errNotFound}
+	return FakeResolver{NumFailures: &atomicx.Int64{}, Err: errNotFound}
 }
 
 func NewFakeResolverWithResult(r []string) FakeResolver {
-	return FakeResolver{NumFailures: atomicx.NewInt64(), Result: r}
+	return FakeResolver{NumFailures: &atomicx.Int64{}, Result: r}
 }
 
 var errNotFound = &net.DNSError{

internal/engine/netx/resolver/serial.go

@@ -6,7 +6,7 @@ import (
 	"net"
 
 	"github.com/miekg/dns"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 )
 
 // RoundTripper represents an abstract DNS transport.
@@ -38,7 +38,7 @@ func NewSerialResolver(t RoundTripper) SerialResolver {
 	return SerialResolver{
 		Encoder:     MiekgEncoder{},
 		Decoder:     MiekgDecoder{},
-		NumTimeouts: atomicx.NewInt64(),
+		NumTimeouts: &atomicx.Int64{},
 		Txp:         t,
 	}
 }

internal/engine/netx/selfcensor/selfcensor.go

@@ -31,7 +31,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 )
 
 // Spec indicates what self censorship techniques to implement.
@@ -61,8 +61,8 @@ type Spec struct {
 }
 
 var (
-	attempts *atomicx.Int64 = atomicx.NewInt64()
+	attempts *atomicx.Int64 = &atomicx.Int64{}
-	enabled  *atomicx.Int64 = atomicx.NewInt64()
+	enabled  *atomicx.Int64 = &atomicx.Int64{}
 	mu       sync.Mutex
 	spec     *Spec
 )

internal/engine/ooapi/README.md

@@ -1,5 +0,0 @@
-# Package ./internal/engine/ooapi
-
-Automatically generated API clients for speaking with OONI servers.
-
-Please, run `go doc ./internal/engine/ooapi` to see API documentation.

internal/engine/ooapi/kvstore_test.go

@@ -1,34 +0,0 @@
-package ooapi
-
-import (
-	"errors"
-	"fmt"
-	"sync"
-)
-
-var errMemkvstoreNotFound = errors.New("memkvstore: not found")
-
-type MemKVStore struct {
-	m  map[string][]byte
-	mu sync.Mutex
-}
-
-func (kvs *MemKVStore) Get(key string) ([]byte, error) {
-	defer kvs.mu.Unlock()
-	kvs.mu.Lock()
-	out, good := kvs.m[key]
-	if !good {
-		return nil, fmt.Errorf("%w: %s", errMemkvstoreNotFound, key)
-	}
-	return out, nil
-}
-
-func (kvs *MemKVStore) Set(key string, value []byte) error {
-	defer kvs.mu.Unlock()
-	kvs.mu.Lock()
-	if kvs.m == nil {
-		kvs.m = make(map[string][]byte)
-	}
-	kvs.m[key] = value
-	return nil
-}

internal/engine/probeservices/checkreportid_test.go

@@ -7,10 +7,10 @@ import (
 	"testing"
 
 	"github.com/apex/log"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/httpx"
-	"github.com/ooni/probe-cli/v3/internal/engine/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/engine/probeservices"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 )
 
 func TestCheckReportIDWorkingAsIntended(t *testing.T) {
@@ -21,9 +21,9 @@ func TestCheckReportIDWorkingAsIntended(t *testing.T) {
 			Logger:     log.Log,
 			UserAgent:  "miniooni/0.1.0-dev",
 		},
-		LoginCalls:    atomicx.NewInt64(),
+		LoginCalls:    &atomicx.Int64{},
-		RegisterCalls: atomicx.NewInt64(),
+		RegisterCalls: &atomicx.Int64{},
-		StateFile:     probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore()),
+		StateFile:     probeservices.NewStateFile(&kvstore.Memory{}),
 	}
 	reportID := `20201209T052225Z_urlgetter_IT_30722_n1_E1VUhMz08SEkgYFU`
 	ctx := context.Background()
@@ -44,9 +44,9 @@ func TestCheckReportIDWorkingWithCancelledContext(t *testing.T) {
 			Logger:     log.Log,
 			UserAgent:  "miniooni/0.1.0-dev",
 		},
-		LoginCalls:    atomicx.NewInt64(),
+		LoginCalls:    &atomicx.Int64{},
-		RegisterCalls: atomicx.NewInt64(),
+		RegisterCalls: &atomicx.Int64{},
-		StateFile:     probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore()),
+		StateFile:     probeservices.NewStateFile(&kvstore.Memory{}),
 	}
 	reportID := `20201209T052225Z_urlgetter_IT_30722_n1_E1VUhMz08SEkgYFU`
 	ctx, cancel := context.WithCancel(context.Background())

internal/engine/probeservices/measurementmeta_test.go

@@ -8,10 +8,10 @@ import (
 	"testing"
 
 	"github.com/apex/log"
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/httpx"
-	"github.com/ooni/probe-cli/v3/internal/engine/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/engine/probeservices"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 )
 
 func TestGetMeasurementMetaWorkingAsIntended(t *testing.T) {
@@ -22,9 +22,9 @@ func TestGetMeasurementMetaWorkingAsIntended(t *testing.T) {
 			Logger:     log.Log,
 			UserAgent:  "miniooni/0.1.0-dev",
 		},
-		LoginCalls:    atomicx.NewInt64(),
+		LoginCalls:    &atomicx.Int64{},
-		RegisterCalls: atomicx.NewInt64(),
+		RegisterCalls: &atomicx.Int64{},
-		StateFile:     probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore()),
+		StateFile:     probeservices.NewStateFile(&kvstore.Memory{}),
 	}
 	config := probeservices.MeasurementMetaConfig{
 		ReportID: `20201209T052225Z_urlgetter_IT_30722_n1_E1VUhMz08SEkgYFU`,
@@ -90,9 +90,9 @@ func TestGetMeasurementMetaWorkingWithCancelledContext(t *testing.T) {
 			Logger:     log.Log,
 			UserAgent:  "miniooni/0.1.0-dev",
 		},
-		LoginCalls:    atomicx.NewInt64(),
+		LoginCalls:    &atomicx.Int64{},
-		RegisterCalls: atomicx.NewInt64(),
+		RegisterCalls: &atomicx.Int64{},
-		StateFile:     probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore()),
+		StateFile:     probeservices.NewStateFile(&kvstore.Memory{}),
 	}
 	config := probeservices.MeasurementMetaConfig{
 		ReportID: `20201209T052225Z_urlgetter_IT_30722_n1_E1VUhMz08SEkgYFU`,

internal/engine/probeservices/probeservices.go

@@ -28,7 +28,7 @@ import (
 	"net/http"
 	"net/url"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/httpx"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 )
@@ -98,8 +98,8 @@ func NewClient(sess Session, endpoint model.Service) (*Client, error) {
 			ProxyURL:   sess.ProxyURL(),
 			UserAgent:  sess.UserAgent(),
 		},
-		LoginCalls:    atomicx.NewInt64(),
+		LoginCalls:    &atomicx.Int64{},
-		RegisterCalls: atomicx.NewInt64(),
+		RegisterCalls: &atomicx.Int64{},
 		StateFile:     NewStateFile(sess.KeyValueStore()),
 	}
 	switch endpoint.Type {

internal/engine/probeservices/register.go

@@ -3,7 +3,7 @@ package probeservices
 import (
 	"context"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/randx"
+	"github.com/ooni/probe-cli/v3/internal/randx"
 )
 
 type registerRequest struct {

internal/engine/probeservices/statefile_test.go

@@ -7,8 +7,8 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
-	"github.com/ooni/probe-cli/v3/internal/engine/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/engine/probeservices"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
 )
 
 func TestStateAuth(t *testing.T) {
@@ -67,7 +67,7 @@ func TestStateCredentials(t *testing.T) {
 func TestStateFileMemoryIntegration(t *testing.T) {
 	// Does the StateFile have the property that we can write
 	// values into it and then read again the same files?
-	sf := probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore())
+	sf := probeservices.NewStateFile(&kvstore.Memory{})
 	s := probeservices.State{
 		Expire:   time.Now(),
 		Password: "xy",
@@ -85,7 +85,7 @@ func TestStateFileMemoryIntegration(t *testing.T) {
 }
 
 func TestStateFileSetMarshalError(t *testing.T) {
-	sf := probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore())
+	sf := probeservices.NewStateFile(&kvstore.Memory{})
 	s := probeservices.State{
 		Expire:   time.Now(),
 		Password: "xy",
@@ -102,7 +102,7 @@ func TestStateFileSetMarshalError(t *testing.T) {
 }
 
 func TestStateFileGetKVStoreGetError(t *testing.T) {
-	sf := probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore())
+	sf := probeservices.NewStateFile(&kvstore.Memory{})
 	expected := errors.New("mocked error")
 	failingfunc := func(string) ([]byte, error) {
 		return nil, expected
@@ -126,7 +126,7 @@ func TestStateFileGetKVStoreGetError(t *testing.T) {
 }
 
 func TestStateFileGetUnmarshalError(t *testing.T) {
-	sf := probeservices.NewStateFile(kvstore.NewMemoryKeyValueStore())
+	sf := probeservices.NewStateFile(&kvstore.Memory{})
 	if err := sf.Set(probeservices.State{}); err != nil {
 		t.Fatal(err)
 	}

internal/engine/session.go

@@ -10,16 +10,16 @@ import (
 	"os"
 	"sync"
 
-	"github.com/ooni/probe-cli/v3/internal/engine/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/geolocate"
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/platform"
 	"github.com/ooni/probe-cli/v3/internal/engine/internal/sessionresolver"
-	"github.com/ooni/probe-cli/v3/internal/engine/kvstore"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx"
 	"github.com/ooni/probe-cli/v3/internal/engine/netx/bytecounter"
 	"github.com/ooni/probe-cli/v3/internal/engine/probeservices"
 	"github.com/ooni/probe-cli/v3/internal/engine/tunnel"
+	"github.com/ooni/probe-cli/v3/internal/kvstore"
+	"github.com/ooni/probe-cli/v3/internal/platform"
 	"github.com/ooni/probe-cli/v3/internal/version"
 )
 
@@ -52,7 +52,7 @@ type Session struct {
 	availableTestHelpers     map[string][]model.Service
 	byteCounter              *bytecounter.Counter
 	httpDefaultTransport     netx.HTTPRoundTripper
-	kvStore                  model.KeyValueStore
+	kvStore                  KVStore
 	location                 *geolocate.Results
 	logger                   model.Logger
 	proxyURL                 *url.URL
@@ -142,7 +142,7 @@ func NewSession(ctx context.Context, config SessionConfig) (*Session, error) {
 		return nil, errors.New("SoftwareVersion is empty")
 	}
 	if config.KVStore == nil {
-		config.KVStore = kvstore.NewMemoryKeyValueStore()
+		config.KVStore = &kvstore.Memory{}
 	}
 	// Implementation note: if config.TempDir is empty, then Go will
 	// use the temporary directory on the current system. This should
@@ -157,7 +157,7 @@ func NewSession(ctx context.Context, config SessionConfig) (*Session, error) {
 		byteCounter:             bytecounter.New(),
 		kvStore:                 config.KVStore,
 		logger:                  config.Logger,
-		queryProbeServicesCount: atomicx.NewInt64(),
+		queryProbeServicesCount: &atomicx.Int64{},
 		softwareName:            config.SoftwareName,
 		softwareVersion:         config.SoftwareVersion,
 		tempDir:                 tempDir,

internal/engine/submitter_test.go

@@ -3,10 +3,10 @@ package engine
 import (
 	"context"
 	"errors"
-	"sync/atomic"
 	"testing"
 
 	"github.com/apex/log"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/engine/model"
 )
 
@@ -28,12 +28,14 @@ func TestSubmitterNotEnabled(t *testing.T) {
 }
 
 type FakeSubmitter struct {
-	Calls uint32
+	Calls *atomicx.Int64
 	Error error
 }
 
 func (fs *FakeSubmitter) Submit(ctx context.Context, m *model.Measurement) error {
-	atomic.AddUint32(&fs.Calls, 1)
+	if fs.Calls != nil {
+		fs.Calls.Add(1)
+	}
 	return fs.Error
 }
 
@@ -68,7 +70,10 @@ func TestNewSubmitterFails(t *testing.T) {
 func TestNewSubmitterWithFailedSubmission(t *testing.T) {
 	expected := errors.New("mocked error")
 	ctx := context.Background()
-	fakeSubmitter := &FakeSubmitter{Error: expected}
+	fakeSubmitter := &FakeSubmitter{
+		Calls: &atomicx.Int64{},
+		Error: expected,
+	}
 	submitter, err := NewSubmitter(ctx, SubmitterConfig{
 		Enabled: true,
 		Logger:  log.Log,
@@ -82,7 +87,7 @@ func TestNewSubmitterWithFailedSubmission(t *testing.T) {
 	if !errors.Is(err, expected) {
 		t.Fatalf("not the error we expected: %+v", err)
 	}
-	if fakeSubmitter.Calls != 1 {
+	if fakeSubmitter.Calls.Load() != 1 {
 		t.Fatal("unexpected number of calls")
 	}
 }

internal/fsx/fsx.go

@@ -1,4 +1,4 @@
-// Package fsx contains file system extension
+// Package fsx contains io/fs extensions.
 package fsx
 
 import (
@@ -8,13 +8,15 @@ import (
 	"syscall"
 )
 
-// Open is a wrapper for os.Open that ensures that we're opening a file.
+// OpenFile is a wrapper for os.OpenFile that ensures that
-func Open(pathname string) (fs.File, error) {
+// we're opening a file rather than a directory. If you are
-	return OpenWithFS(filesystem{}, pathname)
+// opening a directory, this func will return an error.
+func OpenFile(pathname string) (fs.File, error) {
+	return openWithFS(filesystem{}, pathname)
 }
 
-// OpenWithFS is like Open but with explicit file system argument.
+// openWithFS is like Open but with explicit file system argument.
-func OpenWithFS(fs fs.FS, pathname string) (fs.File, error) {
+func openWithFS(fs fs.FS, pathname string) (fs.File, error) {
 	file, err := fs.Open(pathname)
 	if err != nil {
 		return nil, err

internal/fsx/fsx_test.go

@@ -0,0 +1,84 @@
+package fsx
+
+import (
+	"errors"
+	"io/fs"
+	"os"
+	"syscall"
+	"testing"
+
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
+)
+
+// baseDir is the base directory we use for testing.
+var baseDir = "./testdata/"
+
+// failingStatFS is a fs.FS returning a file where stat() fails.
+type failingStatFS struct {
+	CloseCount *atomicx.Int64
+}
+
+// failingStatFile is a fs.File where stat() fails.
+type failingStatFile struct {
+	CloseCount *atomicx.Int64
+}
+
+// errStatFailed is the internal error indicating that stat() failed.
+var errStatFailed = errors.New("stat failed")
+
+// Stat is a stat implementation that fails.
+func (failingStatFile) Stat() (os.FileInfo, error) {
+	return nil, errStatFailed
+}
+
+// Open opens a fake file whose Stat fails.
+func (f failingStatFS) Open(pathname string) (fs.File, error) {
+	return failingStatFile(f), nil
+}
+
+// Close closes the failingStatFile.
+func (fs failingStatFile) Close() error {
+	if fs.CloseCount != nil {
+		fs.CloseCount.Add(1)
+	}
+	return nil
+}
+
+// Read implements fs.File.Read.
+func (failingStatFile) Read([]byte) (int, error) {
+	return 0, errors.New("shouldn't be called")
+}
+
+func TestOpenWithFailingStat(t *testing.T) {
+	count := &atomicx.Int64{}
+	_, err := openWithFS(
+		failingStatFS{CloseCount: count}, baseDir+"testfile.txt")
+	if !errors.Is(err, errStatFailed) {
+		t.Error("expected error with invalid FS", err)
+	}
+	if count.Load() != 1 {
+		t.Error("expected close counter to be equal to 1")
+	}
+}
+
+func TestOpenNonexistentFile(t *testing.T) {
+	_, err := OpenFile(baseDir + "invalidtestfile.txt")
+	if !errors.Is(err, syscall.ENOENT) {
+		t.Errorf("not the error we expected")
+	}
+}
+
+func TestOpenDirectoryShouldFail(t *testing.T) {
+	_, err := OpenFile(baseDir)
+	if !errors.Is(err, syscall.EISDIR) {
+		t.Fatalf("not the error we expected: %+v", err)
+	}
+}
+
+func TestOpeningExistingFileShouldWork(t *testing.T) {
+	file, err := OpenFile(baseDir + "testfile.txt")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer file.Close()
+}

internal/humanize/humanizex.go

@@ -1,14 +1,17 @@
-// Package humanizex is like dustin/go-humanize
+// Package humanize is like dustin/go-humanize.
-package humanizex
+package humanize
 
 import "fmt"
 
-// SI is like dustin/go-humanize.SI
+// SI is like dustin/go-humanize.SI but its implementation is
+// specially tailored for printing download speeds.
 func SI(value float64, unit string) string {
 	value, prefix := reduce(value)
 	return fmt.Sprintf("%3.0f %s%s", value, prefix, unit)
 }
 
+// reduce reduces value to a base value and a unit prefix. For
+// example, reduce(1055) returns (1.055, "k").
 func reduce(value float64) (float64, string) {
 	if value < 1e03 {
 		return value, " "

internal/humanize/humanizex_test.go

@@ -0,0 +1,30 @@
+package humanize
+
+import "testing"
+
+func TestGood(t *testing.T) {
+	if SI(128, "bit/s") != "128  bit/s" {
+		t.Fatal("unexpected result")
+	}
+	if SI(1280, "bit/s") != "  1 kbit/s" {
+		t.Fatal("unexpected result")
+	}
+	if SI(12800, "bit/s") != " 13 kbit/s" {
+		t.Fatal("unexpected result")
+	}
+	if SI(128000, "bit/s") != "128 kbit/s" {
+		t.Fatal("unexpected result")
+	}
+	if SI(1280000, "bit/s") != "  1 Mbit/s" {
+		t.Fatal("unexpected result")
+	}
+	if SI(12800000, "bit/s") != " 13 Mbit/s" {
+		t.Fatal("unexpected result")
+	}
+	if SI(128000000, "bit/s") != "128 Mbit/s" {
+		t.Fatal("unexpected result")
+	}
+	if SI(1280000000, "bit/s") != "  1 Gbit/s" {
+		t.Fatal("unexpected result")
+	}
+}

internal/kvstore/fs.go

@@ -0,0 +1,53 @@
+package kvstore
+
+import (
+	"bytes"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+
+	"github.com/rogpeppe/go-internal/lockedfile"
+)
+
+// FS is a file-system based KVStore.
+type FS struct {
+	basedir string
+}
+
+// NewFS creates a new kvstore.FileSystem.
+func NewFS(basedir string) (kvs *FS, err error) {
+	return newFileSystem(basedir, os.MkdirAll)
+}
+
+// osMkdirAll is the type of os.MkdirAll.
+type osMkdirAll func(path string, perm fs.FileMode) error
+
+// newFileSystem is like NewFileSystem with a customizable
+// osMkdirAll function for creating the kvstore dir.
+func newFileSystem(basedir string, mkdir osMkdirAll) (*FS, error) {
+	if err := mkdir(basedir, 0700); err != nil {
+		return nil, err
+	}
+	return &FS{basedir: basedir}, nil
+}
+
+// filename returns the filename for a given key.
+func (kvs *FS) filename(key string) string {
+	return filepath.Join(kvs.basedir, key)
+}
+
+// Get returns the specified key's value. In case of error, the
+// error type is such that errors.Is(err, ErrNoSuchKey).
+func (kvs *FS) Get(key string) ([]byte, error) {
+	data, err := lockedfile.Read(kvs.filename(key))
+	if err != nil {
+		return nil, fmt.Errorf("%w: %s", ErrNoSuchKey, err.Error())
+	}
+	return data, nil
+}
+
+// Set sets the value of a specific key.
+func (kvs *FS) Set(key string, value []byte) error {
+	return lockedfile.Write(kvs.filename(key), bytes.NewReader(value), 0600)
+}

internal/kvstore/fs_test.go

@@ -0,0 +1,67 @@
+package kvstore
+
+import (
+	"bytes"
+	"errors"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestFileSystemGood(t *testing.T) {
+	dirpath := filepath.Join("testdata", "kvstore2")
+	if err := os.RemoveAll(dirpath); err != nil {
+		t.Fatal(err)
+	}
+	kvstore, err := NewFS(dirpath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	value := []byte("foobar")
+	if err := kvstore.Set("antani", value); err != nil {
+		t.Fatal(err)
+	}
+	ovalue, err := kvstore.Get("antani")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(ovalue, value) {
+		t.Fatal("invalid value")
+	}
+}
+
+func TestFileSystemNoSuchKey(t *testing.T) {
+	dirpath := filepath.Join("testdata", "kvstore2")
+	if err := os.RemoveAll(dirpath); err != nil {
+		t.Fatal(err)
+	}
+	kvstore, err := NewFS(dirpath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	value, err := kvstore.Get("antani")
+	if !errors.Is(err, ErrNoSuchKey) {
+		t.Fatal("not the error we expected", err)
+	}
+	if value != nil {
+		t.Fatal("expected nil value")
+	}
+}
+
+func TestFileSystemWithFailure(t *testing.T) {
+	expect := errors.New("mocked error")
+	mkdir := func(path string, perm fs.FileMode) error {
+		return expect
+	}
+	kvstore, err := newFileSystem(
+		filepath.Join("testdata", "kvstore2"),
+		mkdir,
+	)
+	if !errors.Is(err, expect) {
+		t.Fatal("not the error we expected", err)
+	}
+	if kvstore != nil {
+		t.Fatal("expected nil here")
+	}
+}

internal/kvstore/memory.go

@@ -0,0 +1,42 @@
+// Package kvstore contains key-value stores.
+package kvstore
+
+import (
+	"errors"
+	"sync"
+)
+
+// ErrNoSuchKey indicates that there's no value for the given key.
+var ErrNoSuchKey = errors.New("no such key")
+
+// Memory is an in-memory key-value store.
+type Memory struct {
+	// m is the underlying map.
+	m map[string][]byte
+
+	// mu provides mutual exclusion
+	mu sync.Mutex
+}
+
+// Get returns the specified key's value. In case of error, the
+// error type is such that errors.Is(err, ErrNoSuchKey).
+func (kvs *Memory) Get(key string) ([]byte, error) {
+	kvs.mu.Lock()
+	defer kvs.mu.Unlock()
+	value, ok := kvs.m[key]
+	if !ok {
+		return nil, ErrNoSuchKey
+	}
+	return value, nil
+}
+
+// Set sets a key into the key-value store
+func (kvs *Memory) Set(key string, value []byte) error {
+	kvs.mu.Lock()
+	defer kvs.mu.Unlock()
+	if kvs.m == nil {
+		kvs.m = make(map[string][]byte)
+	}
+	kvs.m[key] = value
+	return nil
+}

internal/kvstore/memory_test.go

@@ -1,11 +1,14 @@
 package kvstore
 
-import "testing"
+import (
+	"errors"
+	"testing"
+)
 
 func TestNoSuchKey(t *testing.T) {
-	kvs := NewMemoryKeyValueStore()
+	kvs := &Memory{}
 	value, err := kvs.Get("nonexistent")
-	if err == nil {
+	if !errors.Is(err, ErrNoSuchKey) {
 		t.Fatal("expected an error here")
 	}
 	if value != nil {
@@ -14,7 +17,7 @@ func TestNoSuchKey(t *testing.T) {
 }
 
 func TestExistingKey(t *testing.T) {
-	kvs := NewMemoryKeyValueStore()
+	kvs := &Memory{}
 	if err := kvs.Set("antani", []byte("mascetti")); err != nil {
 		t.Fatal(err)
 	}

internal/multierror/multierror.go

@@ -9,11 +9,14 @@ import (
 
 // Union is the logical union of several errors. The Union will
 // appear to be the Root error, except that it will actually
-// be possible to look deeper and see specific sub errors that
+// be possible to look deeper and see specific child errors that
 // occurred using errors.As and errors.Is.
 type Union struct {
+	// Children contains the underlying errors.
 	Children []error
-	Root     error
+
+	// Root is the root error.
+	Root error
 }
 
 // New creates a new Union error instance.
@@ -37,8 +40,8 @@ func (err *Union) AddWithPrefix(prefix string, child error) {
 	err.Add(fmt.Errorf("%s: %w", prefix, child))
 }
 
-// Is returns whether the Union error contains at least one child
+// Is returns true (1) if the err.Root error is target or (2) if
-// error that is exactly the specified target error.
+// any err.Children error is target.
 func (err Union) Is(target error) bool {
 	if errors.Is(err.Root, target) {
 		return true

internal/multierror/multierror_test.go

@@ -8,7 +8,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
-	"github.com/ooni/probe-cli/v3/internal/engine/internal/multierror"
+	"github.com/ooni/probe-cli/v3/internal/multierror"
 )
 
 func TestEmpty(t *testing.T) {