7192721681

Committed 13 Dec 2023 08:41AM UTC coverage: 79.479% (-0.02%) from 79.502%

Build # 7192721681

Build Type

push

github

Committed by

web-flow

Commit Message

chore: update to quic-go@v0.40.1 (#1428)

Upgrade to quic-go@v0.40.1 and adapt closing policy after [changes in
quic-go@v0.40.0](https://github.com/quic-go/quic-go/releases/tag/v0.40.0)
and https://github.com/quic-go/quic-go/pull/4072 in particular. We're
creating an UDP conn and [passing it to
server.Serve](https://github.com/ooni/probe-cli/pull/1428/files#diff-efda3daa51fe894a3c850d533179c2425R105),
which [calls
serveConn](https://github.com/quic-go/quic-go/blob/v0.40.1/http3/server.go#L242),
which [calls
quicListen](https://github.com/quic-go/quic-go/blob/v0.40.1/http3/server.go#L316)
and then
[ServeListener](https://github.com/quic-go/quic-go/blob/v0.40.1/http3/server.go#L321).
In turn, ServeListener [is interrupted by
ErrServerClosed](https://github.com/quic-go/quic-go/blob/v0.40.1/http3/server.go#L268),
which seems to be generated by [server.Close calling Close for each
listener](https://github.com/quic-go/quic-go/blob/v0.40.1/http3/server.go#L657).
The following is what happened before updating the shutdown protocol:

```
goroutine 247 [sync.Mutex.Lock]:

[...]

github.com/quic-go/quic-go.(*Transport).closeServer(0x1400054a000?)
	/Users/sbs/go/pkg/mod/github.com/quic-go/quic-go@v0.40.0/transport.go:298 +0x90

github.com/quic-go/quic-go.(*baseServer).close.func1()
	/Users/sbs/go/pkg/mod/github.com/quic-go/quic-go@v0.40.0/server.go:344 +0x84

[...]

github.com/quic-go/quic-go.(*baseServer).close(0x14000cc1c10?, {0x102faa7a0?, 0x140002557b0?}, 0xec?)
	/Users/sbs/go/pkg/mod/github.com/quic-go/quic-go@v0.40.0/server.go:338 +0x64

github.com/quic-go/quic-go.(*baseServer).Close(...)
	/Users/sbs/go/pkg/mod/github.com/quic-go/quic-go@v0.40.0/server.go:333

github.com/quic-go/quic-go.(*EarlyListener).Close(0x14000120700?)
	/Users/sbs/go/pkg/mod/github.com/quic-go/quic-go@v0.40.0/server.go:165 +0x34

github.com/quic-go/quic-go/http3.(*Server).Close(0x140007344d0)
	/Users/sbs/go/pkg/mod... (continued)

Run Details

12 of 12 new or added lines in 2 files covered. (100.0%)

8 existing lines in 2 files now uncovered.

24783 of 31182 relevant lines covered (79.48%)

40.32 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.8

/internal/experiment/webconnectivitylte/priority.go

package webconnectivitylte

//
// Determine which connection(s) are allowed to fetch the webpage
// by giving higher priority to the system resolver, then to the
// UDP resolver, then to the DoH resolver, then to the TH.
//
// This sorting reflects the likelyhood that we will se a blockpage
// because the system resolver is the most likely to be blocked
// (e.g., in Italy). The UDP resolver is also blocked in countries
// with more censorship (e.g., in China). The DoH resolver and
// the TH have more or less the same priority here, but we needed
// to choose one of them to have higher priority.
//
// Note that this functionality is where Web Connectivity LTE
// diverges from websteps, which will instead fetch all the available
// webpages. To adhere to the Web Connectivity model, we need to
// have a single fetch per redirect. However, by allowing all the
// resolvers plus the TH to provide us with addresses, we increase
// our chances of detecting more kinds of censorship.
//
// Also note that this implementation basically makes the
// https://github.com/ooni/probe/issues/2258 issue obsolete,
// since now we're considering all resolvers.
//
// See https://github.com/ooni/probe/issues/2276.
//

import (
        "context"
        "errors"
        "fmt"
        "net"
        "time"

        "github.com/ooni/probe-cli/v3/internal/model"
        "github.com/ooni/probe-cli/v3/internal/runtimex"
)

// errNotPermittedToFetch indicates we're not permitted to fetch a webpage
// because another goroutine have already fetched that webpage.
var errNotPermittedToFetch = errors.New("webconnectivity: not permitted to fetch")

// prioritySelector selects the connection with the highest priority.
type prioritySelector struct {
        // ch is the channel used to ask for priority
        ch chan *priorityRequest

        // logger is the logger to use
        logger model.Logger

        // m contains a map from known addresses to their flags
        m map[string]int64

        // nhttps is the number of addrs resolved using DoH
        nhttps int

        // nsystem is the number of addrs resolved using the system resolver
        nsystem int

        // nudp is the nunber of addrs resolver using UDP
        nudp int

        // tk contains the TestKeys.
        tk *TestKeys

        // zeroTime is the zero time of the current measurement
        zeroTime time.Time
}

// priorityRequest is a request to get priority for fetching the webpage
// over other concurrent connections that are doing the same.
type priorityRequest struct {
        // addr is the address we're using
        addr string

        // resp is the buffered channel where the response will appear
        resp chan bool
}

// newPrioritySelector creates a new prioritySelector instance.
func newPrioritySelector(
        ctx context.Context,
        zeroTime time.Time,
        tk *TestKeys,
        logger model.Logger,
        addrs []DNSEntry,
) *prioritySelector {
        ps := &prioritySelector{
                ch:       make(chan *priorityRequest),
                logger:   logger,
                m:        map[string]int64{},
                nhttps:   0,
                nsystem:  0,
                nudp:     0,
                tk:       tk,
                zeroTime: zeroTime,
        }
        ps.log("create with %+v", addrs)
        for _, addr := range addrs {
                flags := addr.Flags
                ps.m[addr.Addr] = flags
                if (flags & DNSAddrFlagSystemResolver) != 0 {
                        ps.nsystem++
                }
                if (flags & DNSAddrFlagUDP) != 0 {
                        ps.nudp++
                }
                if (flags & DNSAddrFlagHTTPS) != 0 {
                        ps.nhttps++
                }
        }
        go ps.selector(ctx)
        return ps
}

// log formats and emits a ConnPriorityLogEntry
func (ps *prioritySelector) log(format string, v ...any) {
        ps.tk.AppendConnPriorityLogEntry(&ConnPriorityLogEntry{
                Msg: fmt.Sprintf(format, v...),
                T:   time.Since(ps.zeroTime).Seconds(),
        })
        ps.logger.Infof("prioritySelector: "+format, v...)
}

// permissionToFetch returns whether this ready-to-use connection
// is permitted to perform a round trip and fetch the webpage.
func (ps *prioritySelector) permissionToFetch(address string) bool {
        ipAddr, _, err := net.SplitHostPort(address)
        runtimex.PanicOnError(err, "net.SplitHostPort failed")
        r := &priorityRequest{
                addr: ipAddr,
                resp: make(chan bool, 1), // buffer to simplify selector() implementation
        }
        select {
        case <-time.After(10 * time.Millisecond):
                ps.log("conn %s: denied permission: timed out sending", address)
                return false
        case ps.ch <- r:
                select {
                case <-time.After(time.Second):
                        ps.log("conn %s: denied permission: timed out receiving", address)
                        return false
                case v := <-r.resp:
                        ps.log("conn %s: granted permission: %+v", address, v)
                        return v
                }
        }
}

// selector grants permission to the highest priority request that
// arrives within a reasonable time frame. This function runs into the
// background goroutine and terminates when [ctx] is done.
//
// This function implements https://github.com/ooni/probe/issues/2276.
func (ps *prioritySelector) selector(ctx context.Context) {
        // Implementation note: setting an arbitrary timeout here would
        // be ~an issue because we want this goroutine to be available in
        // case the only connections from which we could fetch a webpage
        // are the ones using TH addresses. However, we know the TH could
        // require a long time to complete due to timeouts caused by IP
        // addresses provided by the probe.
        //
        // See https://explorer.ooni.org/measurement/20220911T105037Z_webconnectivity_IT_30722_n1_ruzuQ219SmIO9SrT?input=http%3A%2F%2Festrenosli.org%2F
        // for a measurement where a too-short timeout prevented us from
        // attempting to fetch a webpage from TH-resolved addrs.
        //
        // See https://explorer.ooni.org/measurement/20220911T194527Z_webconnectivity_IT_30722_n1_jufRZGay0Db9Ge4v?input=http%3A%2F%2Festrenosli.org%2F
        // for a measurement where this issue was fixed.

        // await the first priority request
        var first *priorityRequest
        select {
        case <-ctx.Done():
                return
        case first = <-ps.ch:
        }

        // if this request is highest priority, grant permission
        if ps.isHighestPriority(first) {
                first.resp <- true // buffered channel
                return
        }

        // collect additional requests for up to extraTime, thus giving
        // a possibly higher priority connection time to establish
        const extraTime = 500 * time.Millisecond
        expired := time.NewTimer(extraTime)
        defer expired.Stop()
        requests := []*priorityRequest{first}
Loop:
        for {
                select {
                case <-expired.C:
                        break Loop
                case r := <-ps.ch:
                        requests = append(requests, r)
                case <-ctx.Done():
                        return
                }
        }

        // grant permission to the highest priority request
        highPrio := ps.findHighestPriority(requests)
        highPrio.resp <- true // buffered channel

        // deny permission to all the other inflight requests
        for _, r := range requests {
                if highPrio != r {
                        r.resp <- false // buffered channel
                }
        }
}

// findHighestPriority returns the highest priority request
func (ps *prioritySelector) findHighestPriority(reqs []*priorityRequest) *priorityRequest {
        runtimex.Assert(len(reqs) > 0, "findHighestPriority wants a non-empty reqs slice")
        for _, r := range reqs {
                if ps.isHighestPriority(r) {
                        return r
                }
        }
        return reqs[0]
}

// isHighestPriority returns whether this request is highest priority
func (ps *prioritySelector) isHighestPriority(r *priorityRequest) bool {
        // See https://github.com/ooni/probe/issues/2276
        flags := ps.m[r.addr]
        if ps.nsystem > 0 {
                if (flags & DNSAddrFlagSystemResolver) != 0 {
                        return true
                }
        } else if ps.nudp > 0 {
                if (flags & DNSAddrFlagUDP) != 0 {
                        return true
                }
        } else if ps.nhttps > 0 {
                if (flags & DNSAddrFlagHTTPS) != 0 {
                        return true
                }
        } else {
                // Happens when we only have addresses from the TH
                return true
        }
        return false
}

1	package webconnectivitylte
2
3	//
4	// Determine which connection(s) are allowed to fetch the webpage
5	// by giving higher priority to the system resolver, then to the
6	// UDP resolver, then to the DoH resolver, then to the TH.
7	//
8	// This sorting reflects the likelyhood that we will se a blockpage
9	// because the system resolver is the most likely to be blocked
10	// (e.g., in Italy). The UDP resolver is also blocked in countries
11	// with more censorship (e.g., in China). The DoH resolver and
12	// the TH have more or less the same priority here, but we needed
13	// to choose one of them to have higher priority.
14	//
15	// Note that this functionality is where Web Connectivity LTE
16	// diverges from websteps, which will instead fetch all the available
17	// webpages. To adhere to the Web Connectivity model, we need to
18	// have a single fetch per redirect. However, by allowing all the
19	// resolvers plus the TH to provide us with addresses, we increase
20	// our chances of detecting more kinds of censorship.
21	//
22	// Also note that this implementation basically makes the
23	// https://github.com/ooni/probe/issues/2258 issue obsolete,
24	// since now we're considering all resolvers.
25	//
26	// See https://github.com/ooni/probe/issues/2276.
27	//
28
29	import (
30	"context"
31	"errors"
32	"fmt"
33	"net"
34	"time"
35
36	"github.com/ooni/probe-cli/v3/internal/model"
37	"github.com/ooni/probe-cli/v3/internal/runtimex"
38	)
39
40	// errNotPermittedToFetch indicates we're not permitted to fetch a webpage
41	// because another goroutine have already fetched that webpage.
42	var errNotPermittedToFetch = errors.New("webconnectivity: not permitted to fetch")
43
44	// prioritySelector selects the connection with the highest priority.
45	type prioritySelector struct {
46	// ch is the channel used to ask for priority
47	ch chan *priorityRequest
48
49	// logger is the logger to use
50	logger model.Logger
51
52	// m contains a map from known addresses to their flags
53	m map[string]int64
54
55	// nhttps is the number of addrs resolved using DoH
56	nhttps int
57
58	// nsystem is the number of addrs resolved using the system resolver
59	nsystem int
60
61	// nudp is the nunber of addrs resolver using UDP
62	nudp int
63
64	// tk contains the TestKeys.
65	tk *TestKeys
66
67	// zeroTime is the zero time of the current measurement
68	zeroTime time.Time
69	}
70
71	// priorityRequest is a request to get priority for fetching the webpage
72	// over other concurrent connections that are doing the same.
73	type priorityRequest struct {
74	// addr is the address we're using
75	addr string
76
77	// resp is the buffered channel where the response will appear
78	resp chan bool
79	}
80
81	// newPrioritySelector creates a new prioritySelector instance.
82	func newPrioritySelector(
83	ctx context.Context,
84	zeroTime time.Time,
85	tk *TestKeys,
86	logger model.Logger,
87	addrs []DNSEntry,
88	) *prioritySelector {	35✔
89	ps := &prioritySelector{	35✔
90	ch: make(chan *priorityRequest),	35✔
91	logger: logger,	35✔
92	m: map[string]int64{},	35✔
93	nhttps: 0,	35✔
94	nsystem: 0,	35✔
95	nudp: 0,	35✔
96	tk: tk,	35✔
97	zeroTime: zeroTime,	35✔
98	}	35✔
99	ps.log("create with %+v", addrs)	35✔
100	for _, addr := range addrs {	69✔
101	flags := addr.Flags	34✔
102	ps.m[addr.Addr] = flags	34✔
103	if (flags & DNSAddrFlagSystemResolver) != 0 {	65✔
104	ps.nsystem++	31✔
105	}	31✔
106	if (flags & DNSAddrFlagUDP) != 0 {	67✔
107	ps.nudp++	33✔
108	}	33✔
109	if (flags & DNSAddrFlagHTTPS) != 0 {	35✔
110	ps.nhttps++	1✔
111	}	1✔
112	}
113	go ps.selector(ctx)	35✔
114	return ps	35✔
115	}
116
117	// log formats and emits a ConnPriorityLogEntry
118	func (ps *prioritySelector) log(format string, v ...any) {	61✔
119	ps.tk.AppendConnPriorityLogEntry(&ConnPriorityLogEntry{	61✔
120	Msg: fmt.Sprintf(format, v...),	61✔
121	T: time.Since(ps.zeroTime).Seconds(),	61✔
122	})	61✔
123	ps.logger.Infof("prioritySelector: "+format, v...)	61✔
124	}	61✔
125
126	// permissionToFetch returns whether this ready-to-use connection
127	// is permitted to perform a round trip and fetch the webpage.
128	func (ps *prioritySelector) permissionToFetch(address string) bool {	26✔
129	ipAddr, _, err := net.SplitHostPort(address)	26✔
130	runtimex.PanicOnError(err, "net.SplitHostPort failed")	26✔
131	r := &priorityRequest{	26✔
132	addr: ipAddr,	26✔
133	resp: make(chan bool, 1), // buffer to simplify selector() implementation	26✔
134	}	26✔
135	select {	26✔
136	case <-time.After(10 * time.Millisecond):	3✔
137	ps.log("conn %s: denied permission: timed out sending", address)	3✔
138	return false	3✔
139	case ps.ch <- r:	23✔
140	select {	23✔
141	case <-time.After(time.Second):	×
142	ps.log("conn %s: denied permission: timed out receiving", address)	×
143	return false	×
144	case v := <-r.resp:	23✔
145	ps.log("conn %s: granted permission: %+v", address, v)	23✔
146	return v	23✔
147	}
148	}
149	}
150
151	// selector grants permission to the highest priority request that
152	// arrives within a reasonable time frame. This function runs into the
153	// background goroutine and terminates when [ctx] is done.
154	//
155	// This function implements https://github.com/ooni/probe/issues/2276.
156	func (ps *prioritySelector) selector(ctx context.Context) {	35✔
157	// Implementation note: setting an arbitrary timeout here would	35✔
158	// be ~an issue because we want this goroutine to be available in	35✔
159	// case the only connections from which we could fetch a webpage	35✔
160	// are the ones using TH addresses. However, we know the TH could	35✔
161	// require a long time to complete due to timeouts caused by IP	35✔
162	// addresses provided by the probe.	35✔
163	//	35✔
164	// See https://explorer.ooni.org/measurement/20220911T105037Z_webconnectivity_IT_30722_n1_ruzuQ219SmIO9SrT?input=http%3A%2F%2Festrenosli.org%2F	35✔
165	// for a measurement where a too-short timeout prevented us from	35✔
166	// attempting to fetch a webpage from TH-resolved addrs.	35✔
167	//	35✔
168	// See https://explorer.ooni.org/measurement/20220911T194527Z_webconnectivity_IT_30722_n1_jufRZGay0Db9Ge4v?input=http%3A%2F%2Festrenosli.org%2F	35✔
169	// for a measurement where this issue was fixed.	35✔
170		35✔
171	// await the first priority request	35✔
172	var first *priorityRequest	35✔
173	select {	35✔
174	case <-ctx.Done():	12✔
175	return	12✔
176	case first = <-ps.ch:	23✔
177	}
178
179	// if this request is highest priority, grant permission
180	if ps.isHighestPriority(first) {	43✔
181	first.resp <- true // buffered channel	20✔
182	return	20✔
183	}	20✔
184
185	// collect additional requests for up to extraTime, thus giving
186	// a possibly higher priority connection time to establish
187	const extraTime = 500 * time.Millisecond	3✔
188	expired := time.NewTimer(extraTime)	3✔
189	defer expired.Stop()	3✔
190	requests := []*priorityRequest{first}	3✔
191	Loop:	3✔
192	for {	6✔
193	select {	3✔
194	case <-expired.C:	3✔
195	break Loop	3✔
UNCOV 196	case r := <-ps.ch:	×
UNCOV 197	requests = append(requests, r)	×
198	case <-ctx.Done():	×
199	return	×
200	}
201	}
202
203	// grant permission to the highest priority request
204	highPrio := ps.findHighestPriority(requests)	3✔
205	highPrio.resp <- true // buffered channel	3✔
206		3✔
207	// deny permission to all the other inflight requests	3✔
208	for _, r := range requests {	6✔
209	if highPrio != r {	3✔
UNCOV 210	r.resp <- false // buffered channel	×
UNCOV 211	}	×
212	}
213	}
214
215	// findHighestPriority returns the highest priority request
216	func (ps prioritySelector) findHighestPriority(reqs []priorityRequest) *priorityRequest {	3✔
217	runtimex.Assert(len(reqs) > 0, "findHighestPriority wants a non-empty reqs slice")	3✔
218	for _, r := range reqs {	6✔
219	if ps.isHighestPriority(r) {	3✔
UNCOV 220	return r	×
UNCOV 221	}	×
222	}
223	return reqs[0]	3✔
224	}
225
226	// isHighestPriority returns whether this request is highest priority
227	func (ps prioritySelector) isHighestPriority(r priorityRequest) bool {	26✔
228	// See https://github.com/ooni/probe/issues/2276	26✔
229	flags := ps.m[r.addr]	26✔
230	if ps.nsystem > 0 {	50✔
231	if (flags & DNSAddrFlagSystemResolver) != 0 {	42✔
232	return true	18✔
233	}	18✔
234	} else if ps.nudp > 0 {	4✔
235	if (flags & DNSAddrFlagUDP) != 0 {	4✔
236	return true	2✔
237	}	2✔
238	} else if ps.nhttps > 0 {	×
239	if (flags & DNSAddrFlagHTTPS) != 0 {	×
240	return true	×
241	}	×
242	} else {	×
243	// Happens when we only have addresses from the TH	×
244	return true	×
245	}	×
246	return false	6✔
247	}

ooni / probe-cli / 7192721681

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous