x

internal/httpclientx/overlapped.go

@@ -8,14 +8,16 @@ import (
 	"context"
 	"errors"
 	"time"
-
-	"github.com/ooni/probe-cli/v3/internal/erroror"
 )
 
 // OverlappedDefaultScheduleInterval is the default schedule interval. After this interval
 // has elapsed for a URL without seeing a success, we will schedule the next URL.
 const OverlappedDefaultScheduleInterval = 15 * time.Second
 
+// OverlappedDefaultWatchdogTimeout is the timeout after which we assume all the API calls
+// have gone rogue and forcibly interrupt all of them.
+const OverlappedDefaultWatchdogTimeout = 5 * time.Minute
+
 // Overlapped represents the possibility of overlapping HTTP calls for a set of
 // functionally equivalent URLs, such that we start a new call if the previous one
 // has failed to produce a result within the configured ScheduleInterval.
@@ -24,7 +26,7 @@ const OverlappedDefaultScheduleInterval = 15 * time.Second
 //
 // Under very bad networking conditions, [*Overlapped] would cause a new network
 // call to start while the previous one is still in progress and very slowly downloading
-// a response. A future implementation SHOULD probably account for this possibility.
+// a response. A future implementation MIGHT want to account for this possibility.
 type Overlapped[Output any] struct {
 	// RunFunc is the MANDATORY function that fetches the given [*Endpoint].
 	//
@@ -42,12 +44,22 @@ type Overlapped[Output any] struct {
 	//
 	// If you set it manually, you MUST modify it before calling [*Overlapped.Run].
 	ScheduleInterval time.Duration
+
+	// WatchdogTimeout is the MANDATORY timeout after which the code assumes
+	// that all API calls must be aborted and give up.
+	//
+	// This field is typically initialized by [NewOverlappedGetJSON], [NewOverlappedGetRaw],
+	// [NewOverlappedGetXML], or [NewOverlappedPostJSON] to be [OverlappedDefaultWatchdogTimeout].
+	//
+	// If you set it manually, you MUST modify it before calling [*Overlapped.Run].
+	WatchdogTimeout time.Duration
 }
 
 func newOverlappedWithFunc[Output any](fx func(context.Context, *Endpoint) (Output, error)) *Overlapped[Output] {
 	return &Overlapped[Output]{
 		RunFunc:          fx,
 		ScheduleInterval: OverlappedDefaultScheduleInterval,
+		WatchdogTimeout:  OverlappedDefaultWatchdogTimeout,
 	}
 }
 
@@ -82,12 +94,22 @@ func NewOverlappedPostJSON[Input, Output any](input Input, config *Config) *Over
 // ErrGenericOverlappedFailure indicates that a generic [*Overlapped] failure occurred.
 var ErrGenericOverlappedFailure = errors.New("overlapped: generic failure")
 
+// TODO(bassosimone): to use this API with test helpers, we also need to return
+// the successful index to the caller.
+
 // Run runs the overlapped operations, returning the result of the first operation
 // that succeeds and otherwise returning an error describing what happened.
 func (ovx *Overlapped[Output]) Run(ctx context.Context, epnts ...*Endpoint) (Output, error) {
 	return ovx.Reduce(ovx.Map(ctx, epnts...))
 }
 
+// OverlappedErrorOr combines error information, result information and the endpoint index.
+type OverlappedErrorOr[Output any] struct {
+	Err   error
+	Index int
+	Value Output
+}
+
 // Map applies the [*Overlapped.RunFunc] function to each epnts entry, thus producing
 // a result for each entry. This function will cancel subsequent operations until there
 // is a success: subsequent results will be [context.Canceled] errors.
@@ -96,19 +118,20 @@ func (ovx *Overlapped[Output]) Run(ctx context.Context, epnts ...*Endpoint) (Out
 // of each operation, which is mostly useful when running unit tests.
 //
 // Note that this function will return a zero length slice of epnts lenth is also zero.
-func (ovx *Overlapped[Output]) Map(ctx context.Context, epnts ...*Endpoint) []*erroror.Value[Output] {
-	// create cancellable context for early cancellation
+func (ovx *Overlapped[Output]) Map(ctx context.Context, epnts ...*Endpoint) []*OverlappedErrorOr[Output] {
+	// create cancellable context for early cancellation and also apply the
+	// watchdog timeout so that eventually this code returns.
 	//
 	// we are going to cancel this context as soon as we have a successful response so
 	// that we do not waste network resources by performing other attempts.
-	ctx, cancel := context.WithCancel(ctx)
+	ctx, cancel := context.WithTimeout(ctx, ovx.WatchdogTimeout)
 	defer cancel()
 
 	// construct channel for collecting the results
 	//
 	// we're using this channel to communicate results back from goroutines running
 	// in the background and performing the real API call
-	output := make(chan *erroror.Value[Output])
+	output := make(chan *OverlappedErrorOr[Output])
 
 	// create ticker for scheduling subsequent attempts
 	//
@@ -125,7 +148,7 @@ func (ovx *Overlapped[Output]) Map(ctx context.Context, epnts ...*Endpoint) []*e
 	// for simplicity, we're going to collect results from every goroutine
 	// including the ones cancelled by context after the previous success and
 	// then we're going to filter the results and produce a final result
-	results := []*erroror.Value[Output]{}
+	results := []*OverlappedErrorOr[Output]{}
 
 	// keep looping until we have results for each endpoints
 	for len(results) < len(epnts) {
@@ -165,7 +188,7 @@ func (ovx *Overlapped[Output]) Map(ctx context.Context, epnts ...*Endpoint) []*e
 }
 
 // Reduce takes the results of [*Overlapped.Map] and returns either an Output or an error.
-func (ovx *Overlapped[Output]) Reduce(results []*erroror.Value[Output]) (Output, error) {
+func (ovx *Overlapped[Output]) Reduce(results []*OverlappedErrorOr[Output]) (Output, error) {
 	// postprocess the results to check for success and
 	// aggregate all the errors that occurred
 	errorv := []error{}
@@ -191,15 +214,18 @@ func (ovx *Overlapped[Output]) Reduce(results []*erroror.Value[Output]) (Output,
 }
 
 // transact performs an HTTP transaction with the given URL and writes results to the output channel.
-func (ovx *Overlapped[Output]) transact(ctx context.Context, _ int, epnt *Endpoint, output chan<- *erroror.Value[Output]) {
-	// TODO(bassosimone): the index is currently unused but we need to use it
-	// soon to return back which endpoint actually succeded
+func (ovx *Overlapped[Output]) transact(
+	ctx context.Context, idx int, epnt *Endpoint, output chan<- *OverlappedErrorOr[Output]) {
 	// obtain the results
 	value, err := ovx.RunFunc(ctx, epnt)
 
 	// emit the results
 	//
 	// note that this unconditional channel write REQUIRES that we keep reading from
 	// the results channel in Run until we have a result per input endpoint
-	output <- &erroror.Value[Output]{Err: err, Value: value}
+	output <- &OverlappedErrorOr[Output]{
+		Err:   err,
+		Index: idx,
+		Value: value,
+	}
 }

internal/httpclientx/overlapped_test.go

@@ -11,6 +11,8 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/ooni/probe-cli/v3/internal/model"
 	"github.com/ooni/probe-cli/v3/internal/must"
+	"github.com/ooni/probe-cli/v3/internal/netxlite"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 	"github.com/ooni/probe-cli/v3/internal/testingx"
 )
 
@@ -73,15 +75,46 @@ func TestNewOverlappedPostJSONFastRecoverFromEarlyErrors(t *testing.T) {
 	})
 
 	// Now we issue the requests and check we're getting the correct response.
+	//
+	// We're splitting the algorithm into its Map step and its Reduce step because
+	// this allows us to clearly observe what happened.
 
-	apiResp, err := overlapped.Run(
+	results := overlapped.Map(
 		context.Background(),
 		NewEndpoint(zeroTh.URL),
 		NewEndpoint(oneTh.URL),
 		NewEndpoint(twoTh.URL),
 		NewEndpoint(threeTh.URL),
 	)
 
+	runtimex.Assert(len(results) == 4, "unexpected number of results")
+
+	// the first three attempts should have failed with connection reset
+	// while the fourth result should be successful
+	for _, entry := range results {
+		t.Log(entry.Index, string(must.MarshalJSON(entry)))
+		switch entry.Index {
+		case 0, 1, 2:
+			if err := entry.Err; !errors.Is(err, netxlite.ECONNRESET) {
+				t.Fatal("unexpected error", err)
+			}
+		case 3:
+			if err := entry.Err; err != nil {
+				t.Fatal("unexpected error", err)
+			}
+			if diff := cmp.Diff(expectedResponse, entry.Value); diff != "" {
+				t.Fatal(diff)
+			}
+		default:
+			t.Fatal("unexpected index", entry.Index)
+		}
+	}
+
+	// Now run the reduce step of the algorithm and make sure we correctly
+	// return the first success and the nil error
+
+	apiResp, err := overlapped.Reduce(results)
+
 	// we do not expect to see a failure because threeTh is WAI
 	if err != nil {
 		t.Fatal(err)
@@ -104,39 +137,30 @@ func TestNewOverlappedPostJSONFirstCallSucceeds(t *testing.T) {
 	// - 3.th.ooni.org is WAI but slow
 	//
 	// We expect to get a response from the first TH because it's the first goroutine
-	// that we schedule and, even if the wakeup signals for THs are random, the schedule
-	// interval is 15 seconds while we emit a wakeup signal every 0.25 seconds.
-	//
-	// What we're testing here, therefore, is that subsequent calls w
+	// that we schedule. Subsequent calls should be canceled.
 	//
 
 	expectedResponse := &apiResponse{
 		Age:  41,
 		Name: "sbs",
 	}
 
-	slowwakeup := make(chan any)
-
 	zeroTh := testingx.MustNewHTTPServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		<-slowwakeup
 		w.Write(must.MarshalJSON(expectedResponse))
 	}))
 	defer zeroTh.Close()
 
 	oneTh := testingx.MustNewHTTPServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		<-slowwakeup
 		w.Write(must.MarshalJSON(expectedResponse))
 	}))
 	defer oneTh.Close()
 
 	twoTh := testingx.MustNewHTTPServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		<-slowwakeup
 		w.Write(must.MarshalJSON(expectedResponse))
 	}))
 	defer twoTh.Close()
 
 	threeTh := testingx.MustNewHTTPServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		<-slowwakeup
 		w.Write(must.MarshalJSON(expectedResponse))
 	}))
 	defer threeTh.Close()
@@ -160,28 +184,47 @@ func TestNewOverlappedPostJSONFirstCallSucceeds(t *testing.T) {
 	// to fetch from their respective URLs.
 	overlapped.ScheduleInterval = 15 * time.Second
 
-	// In the background we're going to emit slow wakeup signals at fixed intervals
-	// after an initial waiting interval, such that goroutines unblock in order
-
-	go func() {
-		time.Sleep(250 * time.Millisecond)
-		for idx := 0; idx < 4; idx++ {
-			slowwakeup <- true
-			time.Sleep(250 * time.Millisecond)
-		}
-		close(slowwakeup)
-	}()
-
 	// Now we issue the requests and check we're getting the correct response.
+	//
+	// We're splitting the algorithm into its Map step and its Reduce step because
+	// this allows us to clearly observe what happened.
 
-	apiResp, err := overlapped.Run(
+	results := overlapped.Map(
 		context.Background(),
 		NewEndpoint(zeroTh.URL),
 		NewEndpoint(oneTh.URL),
 		NewEndpoint(twoTh.URL),
 		NewEndpoint(threeTh.URL),
 	)
 
+	runtimex.Assert(len(results) == 4, "unexpected number of results")
+
+	// the first three attempts should succeed and subsequent ones should
+	// have failed with the context.Canceled error
+	for _, entry := range results {
+		t.Log(entry.Index, string(must.MarshalJSON(entry)))
+		switch entry.Index {
+		case 1, 2, 3:
+			if err := entry.Err; !errors.Is(err, context.Canceled) {
+				t.Fatal("unexpected error", err)
+			}
+		case 0:
+			if err := entry.Err; err != nil {
+				t.Fatal("unexpected error", err)
+			}
+			if diff := cmp.Diff(expectedResponse, entry.Value); diff != "" {
+				t.Fatal(diff)
+			}
+		default:
+			t.Fatal("unexpected index", entry.Index)
+		}
+	}
+
+	// Now run the reduce step of the algorithm and make sure we correctly
+	// return the first success and the nil error
+
+	apiResp, err := overlapped.Reduce(results)
+
 	// we do not expect to see a failure because all the THs are WAI
 	if err != nil {
 		t.Fatal(err)
@@ -252,30 +295,169 @@ func TestNewOverlappedPostJSONHandlesAllTimeouts(t *testing.T) {
 
 	// Now we issue the requests and check we're getting the correct response.
 	//
-	// IMPORTANT: here we need a context with timeout to ensure that we
-	// eventually stop trying with the blocked-forever servers. In a more
-	// real scenario, even without a context timeout, we have other
-	// safeguards to unblock stuck readers in netxlite code.
+	// We're splitting the algorithm into its Map step and its Reduce step because
+	// this allows us to clearly observe what happened.
 
-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
+	// modify the watchdog timeout be much smaller than usual
+	overlapped.WatchdogTimeout = 2 * time.Second
 
-	apiResp, err := overlapped.Run(
-		ctx,
+	results := overlapped.Map(
+		context.Background(),
 		NewEndpoint(zeroTh.URL),
 		NewEndpoint(oneTh.URL),
 		NewEndpoint(twoTh.URL),
 		NewEndpoint(threeTh.URL),
 	)
 
-	// we do not expect to see a failure because all the THs are WAI
+	runtimex.Assert(len(results) == 4, "unexpected number of results")
+
+	// all the attempts should have failed with context deadline exceeded
+	for _, entry := range results {
+		t.Log(entry.Index, string(must.MarshalJSON(entry)))
+		switch entry.Index {
+		case 0, 1, 2, 3:
+			if err := entry.Err; !errors.Is(err, context.DeadlineExceeded) {
+				t.Fatal("unexpected error", err)
+			}
+		default:
+			t.Fatal("unexpected index", entry.Index)
+		}
+	}
+
+	// Now run the reduce step of the algorithm and make sure we correctly
+	// return the first success and the nil error
+
+	apiResp, err := overlapped.Reduce(results)
+
+	// we expect to see a failure because the watchdog timeout should have fired
 	if !errors.Is(err, context.DeadlineExceeded) {
 		t.Fatal("unexpected error", err)
 	}
 
 	// we expect the api response to be nil
 	if apiResp != nil {
-		t.Fatal("expected non-nil resp")
+		t.Fatal("expected nil resp")
+	}
+
+	// now unblock the blocked goroutines
+	close(blockforever)
+}
+
+func TestNewOverlappedPostJSONResetTimeoutSuccessCanceled(t *testing.T) {
+
+	//
+	// Scenario:
+	//
+	// - 0.th.ooni.org resets the connection
+	// - 1.th.ooni.org causes timeout
+	// - 2.th.ooni.org is WAI
+	// - 3.th.ooni.org causes timeout
+	//
+	// We expect to see a success and to never attempt with 3.th.ooni.org.
+	//
+
+	blockforever := make(chan any)
+
+	zeroTh := testingx.MustNewHTTPServer(testingx.HTTPHandlerReset())
+	defer zeroTh.Close()
+
+	oneTh := testingx.MustNewHTTPServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-blockforever
+		w.WriteHeader(http.StatusBadGateway)
+	}))
+	defer oneTh.Close()
+
+	expectedResponse := &apiResponse{
+		Age:  41,
+		Name: "sbs",
+	}
+
+	twoTh := testingx.MustNewHTTPServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write(must.MarshalJSON(expectedResponse))
+	}))
+	defer twoTh.Close()
+
+	threeTh := testingx.MustNewHTTPServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-blockforever
+		w.WriteHeader(http.StatusBadGateway)
+	}))
+	defer threeTh.Close()
+
+	// Create client configuration. We don't care much about the
+	// JSON requests and reponses being aligned to reality.
+
+	apiReq := &apiRequest{
+		UserID: 117,
+	}
+
+	overlapped := NewOverlappedPostJSON[*apiRequest, *apiResponse](apiReq, &Config{
+		Authorization: "", // not relevant for this test
+		Client:        http.DefaultClient,
+		Logger:        log.Log,
+		UserAgent:     model.HTTPHeaderUserAgent,
+	})
+
+	// make sure the schedule interval is low to make this test run faster.
+	overlapped.ScheduleInterval = 250 * time.Millisecond
+
+	// Now we issue the requests and check we're getting the correct response.
+	//
+	// We're splitting the algorithm into its Map step and its Reduce step because
+	// this allows us to clearly observe what happened.
+
+	// modify the watchdog timeout be much smaller than usual
+	overlapped.WatchdogTimeout = 2 * time.Second
+
+	results := overlapped.Map(
+		context.Background(),
+		NewEndpoint(zeroTh.URL),
+		NewEndpoint(oneTh.URL),
+		NewEndpoint(twoTh.URL),
+		NewEndpoint(threeTh.URL),
+	)
+
+	runtimex.Assert(len(results) == 4, "unexpected number of results")
+
+	// attempt 0: should have seen connection reset
+	// attempt 1: should have seen the context canceled
+	// attempt 2: should be successful
+	// attempt 3: should have seen the context canceled
+	for _, entry := range results {
+		t.Log(entry.Index, string(must.MarshalJSON(entry)))
+		switch entry.Index {
+		case 0:
+			if err := entry.Err; !errors.Is(err, netxlite.ECONNRESET) {
+				t.Fatal("unexpected error", err)
+			}
+		case 1, 3:
+			if err := entry.Err; !errors.Is(err, context.Canceled) {
+				t.Fatal("unexpected error", err)
+			}
+		case 2:
+			if err := entry.Err; err != nil {
+				t.Fatal("unexpected error", err)
+			}
+			if diff := cmp.Diff(expectedResponse, entry.Value); diff != "" {
+				t.Fatal(diff)
+			}
+		default:
+			t.Fatal("unexpected index", entry.Index)
+		}
+	}
+
+	// Now run the reduce step of the algorithm and make sure we correctly
+	// return the first success and the nil error
+
+	apiResp, err := overlapped.Reduce(results)
+
+	// we do not expect to see a failure because all the THs are WAI
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// compare response to expectation
+	if diff := cmp.Diff(expectedResponse, apiResp); diff != "" {
+		t.Fatal(diff)
 	}
 
 	// now unblock the blocked goroutines