Merge branch 'master' into issue/2545-small-6

internal/dslx/dns.go

@@ -107,8 +107,9 @@ func DNSLookupGetaddrinfo(rt Runtime) Func[*DomainToResolve, *ResolvedAddresses]
 		}
 
 		return &Maybe[*ResolvedAddresses]{
-			Error: err,
-			State: state,
+			Error:        err,
+			Observations: maybeTraceToObservations(trace),
+			State:        state,
 		}
 	})
 }
@@ -157,8 +158,9 @@ func DNSLookupUDP(rt Runtime, endpoint string) Func[*DomainToResolve, *ResolvedA
 		}
 
 		return &Maybe[*ResolvedAddresses]{
-			Error: err,
-			State: state,
+			Error:        err,
+			Observations: maybeTraceToObservations(trace),
+			State:        state,
 		}
 	})
 }

internal/dslx/fxcore.go

@@ -6,6 +6,8 @@ package dslx
 
 import (
 	"context"
+
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 // Func is a function f: (context.Context, A) -> B.
@@ -19,7 +21,11 @@ type Operation[A, B any] func(ctx context.Context, a A) *Maybe[B]
 // Apply implements Func.
 func (op Operation[A, B]) Apply(ctx context.Context, a *Maybe[A]) *Maybe[B] {
 	if a.Error != nil {
-		return NewMaybeWithError[B](a.Error)
+		return &Maybe[B]{
+			Error:        a.Error,
+			Observations: a.Observations,
+			State:        *new(B), // zero value
+		}
 	}
 	return op(ctx, a.State)
 }
@@ -30,6 +36,9 @@ type Maybe[State any] struct {
 	// Error is either the error that occurred or nil.
 	Error error
 
+	// Observations contains the collected observations.
+	Observations []*Observations
+
 	// State contains state passed between function calls. You should
 	// only access State when Error is nil and Skipped is false.
 	State State
@@ -38,16 +47,9 @@ type Maybe[State any] struct {
 // NewMaybeWithValue constructs a Maybe containing the given value.
 func NewMaybeWithValue[State any](value State) *Maybe[State] {
 	return &Maybe[State]{
-		Error: nil,
-		State: value,
-	}
-}
-
-// NewMaybeWithError constructs a Maybe containing the given error.
-func NewMaybeWithError[State any](err error) *Maybe[State] {
-	return &Maybe[State]{
-		Error: err,
-		State: *new(State), // zero value
+		Error:        nil,
+		Observations: []*Observations{},
+		State:        value,
 	}
 }
 
@@ -67,5 +69,23 @@ type compose2Func[A, B, C any] struct {
 
 // Apply implements Func
 func (h *compose2Func[A, B, C]) Apply(ctx context.Context, a *Maybe[A]) *Maybe[C] {
-	return h.g.Apply(ctx, h.f.Apply(ctx, a))
+	mb := h.f.Apply(ctx, a)
+	runtimex.Assert(mb != nil, "h.f.Apply returned a nil pointer")
+
+	if mb.Error != nil {
+		return &Maybe[C]{
+			Error:        mb.Error,
+			Observations: mb.Observations,
+			State:        *new(C), // zero value
+		}
+	}
+
+	mc := h.g.Apply(ctx, mb)
+	runtimex.Assert(mc != nil, "h.g.Apply returned a nil pointer")
+
+	return &Maybe[C]{
+		Error:        mc.Error,
+		Observations: append(mb.Observations, mc.Observations...), // merge observations
+		State:        mc.State,
+	}
 }

internal/dslx/fxcore_test.go

@@ -27,6 +27,11 @@ func (f *fn) Apply(ctx context.Context, i *Maybe[int]) *Maybe[int] {
 	return &Maybe[int]{
 		Error: f.err,
 		State: i.State + 1,
+		Observations: []*Observations{
+			{
+				NetworkEvents: []*model.ArchivalNetworkEvent{{Tags: []string{"apply"}}},
+			},
+		},
 	}
 }
 
@@ -45,8 +50,9 @@ func TestStageAdapter(t *testing.T) {
 
 		// create input that contains an error
 		input := &Maybe[*DomainToResolve]{
-			Error: errors.New("mocked error"),
-			State: nil,
+			Error:        errors.New("mocked error"),
+			Observations: []*Observations{},
+			State:        nil,
 		}
 
 		// run the pipeline
@@ -85,6 +91,9 @@ func TestCompose2(t *testing.T) {
 				if r.Error != tt.err {
 					t.Fatalf("unexpected error")
 				}
+				if len(r.Observations) != tt.numObs {
+					t.Fatalf("unexpected number of (merged) observations")
+				}
 			})
 		}
 	})
@@ -104,3 +113,20 @@ func TestGen(t *testing.T) {
 		}
 	})
 }
+
+func TestObservations(t *testing.T) {
+	t.Run("Extract observations", func(t *testing.T) {
+		fn1 := getFn(nil, "succeed")
+		fn2 := getFn(nil, "succeed")
+		composit := Compose2(fn1, fn2)
+		r1 := composit.Apply(context.Background(), NewMaybeWithValue(3))
+		r2 := composit.Apply(context.Background(), NewMaybeWithValue(42))
+		if len(r1.Observations) != 2 || len(r2.Observations) != 2 {
+			t.Fatalf("unexpected number of observations")
+		}
+		mergedObservations := ExtractObservations(r1, r2)
+		if len(mergedObservations) != 4 {
+			t.Fatalf("unexpected number of merged observations")
+		}
+	})
+}

internal/dslx/httpcore.go

@@ -151,8 +151,9 @@ func HTTPRequest(rt Runtime, options ...HTTPRequestOption) Func[*HTTPConnection,
 		}
 
 		return &Maybe[*HTTPResponse]{
-			Error: err,
-			State: state,
+			Error:        err,
+			Observations: observations,
+			State:        state,
 		}
 	})
 }

internal/dslx/httpquic.go

@@ -35,8 +35,9 @@ func HTTPConnectionQUIC(rt Runtime) Func[*QUICConnection, *HTTPConnection] {
 			Transport:             httpTransport,
 		}
 		return &Maybe[*HTTPConnection]{
-			Error: nil,
-			State: state,
+			Error:        nil,
+			Observations: nil,
+			State:        state,
 		}
 	})
 }

internal/dslx/httptcp.go

@@ -36,8 +36,9 @@ func HTTPConnectionTCP(rt Runtime) Func[*TCPConnection, *HTTPConnection] {
 			Transport:             httpTransport,
 		}
 		return &Maybe[*HTTPConnection]{
-			Error: nil,
-			State: state,
+			Error:        nil,
+			Observations: nil,
+			State:        state,
 		}
 	})
 }

internal/dslx/httptls.go

@@ -36,8 +36,9 @@ func HTTPConnectionTLS(rt Runtime) Func[*TLSConnection, *HTTPConnection] {
 			Transport:             httpTransport,
 		}
 		return &Maybe[*HTTPConnection]{
-			Error: nil,
-			State: state,
+			Error:        nil,
+			Observations: nil,
+			State:        state,
 		}
 	})
 }

internal/dslx/quic.go

@@ -71,8 +71,9 @@ func QUICHandshake(rt Runtime, options ...TLSHandshakeOption) Func[*Endpoint, *Q
 		}
 
 		return &Maybe[*QUICConnection]{
-			Error: err,
-			State: state,
+			Error:        err,
+			Observations: maybeTraceToObservations(trace),
+			State:        state,
 		}
 	})
 }

internal/dslx/tcp.go

@@ -55,8 +55,9 @@ func TCPConnect(rt Runtime) Func[*Endpoint, *TCPConnection] {
 		}
 
 		return &Maybe[*TCPConnection]{
-			Error: err,
-			State: state,
+			Error:        err,
+			Observations: maybeTraceToObservations(trace),
+			State:        state,
 		}
 	})
 }

internal/dslx/tls.go

@@ -96,8 +96,9 @@ func TLSHandshake(rt Runtime, options ...TLSHandshakeOption) Func[*TCPConnection
 		}
 
 		return &Maybe[*TLSConnection]{
-			Error: err,
-			State: state,
+			Error:        err,
+			Observations: maybeTraceToObservations(trace),
+			State:        state,
 		}
 	})
 }