fix(enginenetx): periodically trim statistics (ooni#1317)

If we're running the enginenetx code inside an environment that sprays
~random IP addresses during DNS lookup, we're going to end up using
quite a large amount of statistics. While this amount of statistics
could potentially be great in general, we need to be mindful of memory
and disk occupation as well as of the overhead of serializing and
deserializing.

To address these concerns, add the arbitrary rule that we don't want
more than ten domain-endpoint entries and implement code to zap old
and/or always-failing entries. The code will run when loading the
entries, when serializing them, and periodically while the `*Network` is
up and running.

See ooni/probe#2531

internal/enginenetx/network.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"net/http/cookiejar"
 	"net/url"
+	"time"
 
 	"github.com/ooni/probe-cli/v3/internal/bytecounter"
 	"github.com/ooni/probe-cli/v3/internal/model"
@@ -54,7 +55,7 @@ func (n *Network) Close() error {
 	// same as above but for the resolver's connections
 	n.reso.CloseIdleConnections()
 
-	// make sure we sync stats to disk
+	// make sure we sync stats to disk and shutdown the background trimmer
 	return n.stats.Close()
 }
 
@@ -92,7 +93,8 @@ func NewNetwork(
 	dialer := netxlite.NewDialerWithResolver(logger, resolver)
 
 	// Create manager for keeping track of statistics
-	stats := newStatsManager(kvStore, logger)
+	const trimInterval = 30 * time.Second
+	stats := newStatsManager(kvStore, logger, trimInterval)
 
 	// Create a TLS dialer ONLY used for dialing TLS connections. This dialer will use
 	// happy-eyeballs and possibly custom policies for dialing TLS connections.

internal/enginenetx/network_internal_test.go

@@ -42,10 +42,14 @@ func TestNetworkUnit(t *testing.T) {
 				},
 			},
 			stats: &statsManager{
+				cancel:    func() { /* nothing */ },
+				closeOnce: sync.Once{},
 				container: &statsContainer{},
 				kvStore:   &kvstore.Memory{},
 				logger:    model.DiscardLogger,
 				mu:        sync.Mutex{},
+				pruned:    make(chan any),
+				wg:        &sync.WaitGroup{},
 			},
 			txp: expected,
 		}
@@ -67,10 +71,14 @@ func TestNetworkUnit(t *testing.T) {
 		netx := &Network{
 			reso: expected,
 			stats: &statsManager{
+				cancel:    func() { /* nothing */ },
+				closeOnce: sync.Once{},
 				container: &statsContainer{},
 				kvStore:   &kvstore.Memory{},
 				logger:    model.DiscardLogger,
 				mu:        sync.Mutex{},
+				pruned:    make(chan any),
+				wg:        &sync.WaitGroup{},
 			},
 			txp: &mocks.HTTPTransport{
 				MockCloseIdleConnections: func() {
@@ -82,7 +90,41 @@ func TestNetworkUnit(t *testing.T) {
 			t.Fatal(err)
 		}
 		if !called {
-			t.Fatal("did not call the transport's CloseIdleConnections")
+			t.Fatal("did not call the resolver's CloseIdleConnections")
+		}
+	})
+
+	t.Run("Close calls the .cancel field of the statsManager as a side effect", func(t *testing.T) {
+		var called bool
+		netx := &Network{
+			reso: &mocks.Resolver{
+				MockCloseIdleConnections: func() {
+					// nothing
+				},
+			},
+			stats: &statsManager{
+				cancel: func() {
+					called = true
+				},
+				closeOnce: sync.Once{},
+				container: &statsContainer{},
+				kvStore:   &kvstore.Memory{},
+				logger:    model.DiscardLogger,
+				mu:        sync.Mutex{},
+				pruned:    make(chan any),
+				wg:        &sync.WaitGroup{},
+			},
+			txp: &mocks.HTTPTransport{
+				MockCloseIdleConnections: func() {
+					// nothing
+				},
+			},
+		}
+		if err := netx.Close(); err != nil {
+			t.Fatal(err)
+		}
+		if !called {
+			t.Fatal("did not call the .cancel field of the statsManager")
 		}
 	})