fix(enginenetx): periodically trim statistics

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")
 		}
 	})