diff --git a/backend/radiance.go b/backend/radiance.go index 142feab5..581b42e5 100644 --- a/backend/radiance.go +++ b/backend/radiance.go @@ -2016,6 +2016,12 @@ func (r *LocalBackend) HelperActivity() helper.Activity { return r.helperActivity.Today() } +// HelperLive is whether each helper pairing's friend is connected to this +// volunteer now. +func (r *LocalBackend) HelperLive() helper.LiveSnapshot { + return helper.DefaultLive.Snapshot() +} + // SetHelperRelayForTest sets the relayer HelperRelay uses, for tests of the // IPC route, which cannot build a full backend. func (r *LocalBackend) SetHelperRelayForTest(rl *helper.Relayer) { diff --git a/go.mod b/go.mod index 781e7c6a..1e7b093b 100644 --- a/go.mod +++ b/go.mod @@ -36,11 +36,11 @@ require ( github.com/Microsoft/go-winio v0.6.2 github.com/alexflint/go-arg v1.6.1 github.com/getlantern/amp v0.0.0-20260606002220-a8629924577c - github.com/getlantern/broflake v0.0.0-20261030194216-3ae437b4c1a6 + github.com/getlantern/broflake v0.0.0-20261031023630-d3b5e0b34b68 github.com/getlantern/common v1.2.1-0.20261028190346-77afb53a5b63 github.com/getlantern/domainfront v0.0.0-20261026202828-9727b2f2a34a github.com/getlantern/kindling v0.0.0-20261026211804-dac6d561f52e - github.com/getlantern/lantern-box v0.0.182 + github.com/getlantern/lantern-box v0.0.184 github.com/getlantern/pluriconfig v0.0.0-20251126214241-8cc8bc561535 github.com/getlantern/publicip v0.0.0-20260328175246-2c460fe80c6b github.com/getlantern/semconv v0.0.0-20261028170058-373fcfda8972 @@ -127,7 +127,7 @@ require ( github.com/getlantern/algeneva v0.0.0-20260731172949-952f13725b7b // indirect github.com/getlantern/keepcurrent v0.0.0-20261009194109-22f33eaeac6e // indirect github.com/getlantern/lantern-water v0.0.0-20260520145825-958775d51395 // indirect - github.com/getlantern/samizdat v0.0.3-0.20261027215106-50f438518169 // indirect + github.com/getlantern/samizdat v0.0.3-0.20261031023545-1248df926631 // indirect github.com/getlantern/twiddle v0.0.0-20260914204124-f698a0d67d2f // indirect github.com/go-chi/chi/v5 v5.2.5 // indirect github.com/go-json-experiment/json v0.0.0-20250813024750-ebf49471dced // indirect diff --git a/go.sum b/go.sum index c564b931..65ffd167 100644 --- a/go.sum +++ b/go.sum @@ -229,8 +229,8 @@ github.com/getlantern/algeneva v0.0.0-20260731172949-952f13725b7b h1:3wQsYZSl/Uv github.com/getlantern/algeneva v0.0.0-20260731172949-952f13725b7b/go.mod h1:PrNR8tMXO26YNs8K9653XCUH7u2Kv4OdfFC3Ke1GsX0= github.com/getlantern/amp v0.0.0-20260606002220-a8629924577c h1:ZzxuhIWO295y4eE6wPYUq8Smntlgg+y39L5OOyMJY+k= github.com/getlantern/amp v0.0.0-20260606002220-a8629924577c/go.mod h1:b5teAOFT+vpBqc2CHoz74QTEM15iOv1PLdQogwXcfrM= -github.com/getlantern/broflake v0.0.0-20261030194216-3ae437b4c1a6 h1:506tYeA81D+BBSlpKyqVgpYkfoH7CBHUNT/ry7HlrTA= -github.com/getlantern/broflake v0.0.0-20261030194216-3ae437b4c1a6/go.mod h1:wIe3Pq0VMMADBiKdW2MUV0EW6veThHMeKuJFvmcGPc4= +github.com/getlantern/broflake v0.0.0-20261031023630-d3b5e0b34b68 h1:dB9/uchWa2TP0nymr67rY0bMUOx5tMh7ZZl7h4ge1ZY= +github.com/getlantern/broflake v0.0.0-20261031023630-d3b5e0b34b68/go.mod h1:wIe3Pq0VMMADBiKdW2MUV0EW6veThHMeKuJFvmcGPc4= github.com/getlantern/common v1.2.1-0.20261028190346-77afb53a5b63 h1:wGTbuemnpn7T39NJbRLV69OeaNb08G+4FedAKCD1Yec= github.com/getlantern/common v1.2.1-0.20261028190346-77afb53a5b63/go.mod h1:9iL78jR5kZqr0E7oPxi38o6gPMhVa5dJBC5ngUgHzdg= github.com/getlantern/context v0.0.0-20220418194847-3d5e7a086201 h1:oEZYEpZo28Wdx+5FZo4aU7JFXu0WG/4wJWese5reQSA= @@ -249,8 +249,8 @@ github.com/getlantern/keepcurrent v0.0.0-20261009194109-22f33eaeac6e h1:iJfNXnhg github.com/getlantern/keepcurrent v0.0.0-20261009194109-22f33eaeac6e/go.mod h1:ag5g9aWUw2FJcX5RVRpJ9EBQBy5yJuy2WXDouIn/m4w= github.com/getlantern/kindling v0.0.0-20261026211804-dac6d561f52e h1:gmpp6m6i/qxbOEiirSgeRYf4m3NfSZIlYfNY66MXddU= github.com/getlantern/kindling v0.0.0-20261026211804-dac6d561f52e/go.mod h1:AQdjYiOX4FQgHP6f6J/GUXgNb++U6ejXgt5JM3UZvaU= -github.com/getlantern/lantern-box v0.0.182 h1:pV6+mQjJT5xQelBOaZ93Jg26AcuOJYTtZfH5NaFx09U= -github.com/getlantern/lantern-box v0.0.182/go.mod h1:uACkGBbcGj9G8RXwKrFGl/kN6FV/ayzCX2nIjciT0oQ= +github.com/getlantern/lantern-box v0.0.184 h1:xslToE04RVt4Z6nohSveeteaV4FkpGl3Tny1XWY7n38= +github.com/getlantern/lantern-box v0.0.184/go.mod h1:zEuBNepQnfk1BTFT2l3pRSgwiOAnoNndXDRMqAdQCvo= github.com/getlantern/lantern-water v0.0.0-20260520145825-958775d51395 h1:grfGavAUp2E9w9ZoJuM3FyWyQ0sCJ64V4ZMKtZKRqTc= github.com/getlantern/lantern-water v0.0.0-20260520145825-958775d51395/go.mod h1:3JpJgwi4KEI6rS9loOAvcBp+F2jP65d0tTg2GQcTPBU= github.com/getlantern/ops v0.0.0-20231025133620-f368ab734534 h1:3BwvWj0JZzFEvNNiMhCu4bf60nqcIuQpTYb00Ezm1ag= @@ -261,8 +261,8 @@ github.com/getlantern/pluriconfig v0.0.0-20251126214241-8cc8bc561535 h1:rtDmW8YL github.com/getlantern/pluriconfig v0.0.0-20251126214241-8cc8bc561535/go.mod h1:WKJEdjMOD4IuTRYwjQHjT4bmqDl5J82RShMLxPAvi0Q= github.com/getlantern/publicip v0.0.0-20260328175246-2c460fe80c6b h1:gMYJzEhLrmIqQ+JnjiYNm+UyUDalK3WUmVyecFwmV5g= github.com/getlantern/publicip v0.0.0-20260328175246-2c460fe80c6b/go.mod h1:NpfXdK4ldEKkjQ4P1R+DBF4ua5VFOlxmgHROTnYrApg= -github.com/getlantern/samizdat v0.0.3-0.20261027215106-50f438518169 h1:HdyX8+/hCGO3zYsyN98Hw5I/8ne3c0WLQAFfTzPXtuc= -github.com/getlantern/samizdat v0.0.3-0.20261027215106-50f438518169/go.mod h1:uEeykQSW2/6rTjfPlj3MTTo59poSHXfAHTGgzYDkbr0= +github.com/getlantern/samizdat v0.0.3-0.20261031023545-1248df926631 h1:R26Ya7kJpA5LafZ72FsJGovwQGfzaOMRsk7gF6vy1H0= +github.com/getlantern/samizdat v0.0.3-0.20261031023545-1248df926631/go.mod h1:uEeykQSW2/6rTjfPlj3MTTo59poSHXfAHTGgzYDkbr0= github.com/getlantern/semconv v0.0.0-20261028170058-373fcfda8972 h1:DSaI15cQZ4MdMnTc70rnE1mxNdzatkJwSvkHuIV+mRs= github.com/getlantern/semconv v0.0.0-20261028170058-373fcfda8972/go.mod h1:GkPT5P9JoOTIRXRmFWxYgu1hhXgTFFTNc2hoG7WQc3g= github.com/getlantern/sing-box-minimal v1.13.20-0.20261019115333-624aa48e0252 h1:p6tbxiX1SHqo1FwO9l1uYerSSaMLt/R78QtLIScpP54= diff --git a/helper/live.go b/helper/live.go new file mode 100644 index 00000000..410bae19 --- /dev/null +++ b/helper/live.go @@ -0,0 +1,358 @@ +package helper + +import ( + "sort" + "sync" + "time" +) + +// Defaults for LiveTracker. A pairing is connected when it moved at least +// LiveThreshold bytes in the last LiveWindow. The threshold is far above what +// a url-test probe or an idle WebRTC slot moves in a minute, so neither shows +// a friend as connected. A connected pairing stays so until it has been below +// the threshold for LiveLinger, so traffic hovering at the threshold does not +// flap. +const ( + LiveThreshold = 32 << 10 + LiveWindow = 60 * time.Second + LiveLinger = 15 * time.Second + // LiveSampleInterval is how often open connections' byte totals are + // read while any is open, so traffic is stamped with when it moved + // rather than when someone next asked. + LiveSampleInterval = time.Second +) + +// DefaultLive is the process's LiveTracker. +var DefaultLive = NewLiveTracker(nil) + +// LivePairing is one pairing's live state on the volunteer. +type LivePairing struct { + PairingID string `json:"pairing_id"` + // Connected says the friend is relaying real traffic through this device + // now. + Connected bool `json:"connected"` + // Since is when Connected last changed, or when the pairing was first + // seen if it never has, in Unix milliseconds. + Since int64 `json:"since"` +} + +// LiveSnapshot is the live state of every pairing with an open connection, +// recent traffic, or a connected state. A pairing missing from it is not +// connected. +type LiveSnapshot struct { + Pairings []LivePairing `json:"pairings"` +} + +// LiveTracker works out, per helper pairing, whether the friend is connected +// to this volunteer right now. It is in memory only: pairing IDs and their +// timeline are never written to disk, logged, or sent anywhere but the local +// app. +type LiveTracker struct { + mu sync.Mutex + now func() time.Time + // wall reads the wall clock, which unlike now's monotonic reading keeps + // running while the device sleeps. + wall func() time.Time + // slept totals the wall clock's forward jumps past now, as while the + // device slept or after a forward wall-clock step. + slept time.Duration + lastNow time.Time + lastWall time.Time + threshold uint64 + window time.Duration + linger time.Duration + // interval is how often the sampler runs; zero disables it, for tests + // that drive sampling by hand. + interval time.Duration + // sampling says the sampler goroutine is running. It runs only while a + // connection is open, and exits once none is. + sampling bool + pairings map[string]*livePairing + flows map[any]*liveFlow +} + +type liveFlow struct { + owner string + pairing string + read func() uint64 + last uint64 + // lastAt is when the flow was last sampled, on clockLocked's clock. + lastAt time.Time +} + +type liveSample struct { + at time.Time + n uint64 +} + +type livePairing struct { + // samples are byte counts per second, oldest first, within the window. + samples []liveSample + flows int + connected bool + since time.Time + // quietSince is when a connected pairing fell below the threshold, or + // zero while it is above it. + quietSince time.Time +} + +// NewLiveTracker returns a tracker with the default threshold, window and +// linger. now nil means time.Now. +func NewLiveTracker(now func() time.Time) *LiveTracker { + if now == nil { + now = time.Now + } + return &LiveTracker{ + now: now, + wall: func() time.Time { return time.Now().Round(0) }, + threshold: LiveThreshold, + window: LiveWindow, + linger: LiveLinger, + interval: LiveSampleInterval, + pairings: make(map[string]*livePairing), + flows: make(map[any]*liveFlow), + } +} + +// Open starts following a connection that came through pairing. key +// identifies it for Close, and owner groups it for CloseOwner; opening a key +// already open closes the earlier one. read, if not nil, returns its +// cumulative byte count, sampled every LiveSampleInterval while any +// connection is open, at Snapshot and at Close. An empty pairing is ignored. +func (t *LiveTracker) Open(owner string, key any, pairing string, read func() uint64) { + if t == nil || pairing == "" || key == nil { + return + } + t.mu.Lock() + defer t.mu.Unlock() + now := t.clockLocked() + if old, ok := t.flows[key]; ok { + t.closeLocked(key, old, now) + } + f := &liveFlow{owner: owner, pairing: pairing, read: read, lastAt: now} + if read != nil { + f.last = read() + } + t.flows[key] = f + p := t.pairingLocked(pairing, now) + p.flows++ + t.evaluateLocked(p, now) + if !t.sampling && t.interval > 0 { + t.sampling = true + go t.sampleLoop(t.interval) + } +} + +func (t *LiveTracker) sampleLoop(interval time.Duration) { + tk := time.NewTicker(interval) + defer tk.Stop() + for range tk.C { + t.mu.Lock() + if len(t.flows) == 0 { + t.sampling = false + t.mu.Unlock() + return + } + t.sampleAllLocked(t.clockLocked()) + t.mu.Unlock() + } +} + +func (t *LiveTracker) sampleAllLocked(now time.Time) { + for _, f := range t.flows { + t.sampleLocked(f, now) + } + for _, p := range t.pairings { + t.evaluateLocked(p, now) + } +} + +// Close stops following the connection key, counting its last bytes. +func (t *LiveTracker) Close(key any) { + if t == nil || key == nil { + return + } + t.mu.Lock() + defer t.mu.Unlock() + if f, ok := t.flows[key]; ok { + t.closeLocked(key, f, t.clockLocked()) + } +} + +// CloseOwner closes every connection opened with owner, for a source that +// stopped and will report no closes. +func (t *LiveTracker) CloseOwner(owner string) { + if t == nil { + return + } + t.mu.Lock() + defer t.mu.Unlock() + now := t.clockLocked() + for key, f := range t.flows { + if f.owner == owner { + t.closeLocked(key, f, now) + } + } +} + +// Add counts n bytes relayed for pairing now, for a source that reports +// bytes as they move rather than as a running total. +func (t *LiveTracker) Add(pairing string, n uint64) { + if t == nil || pairing == "" || n == 0 { + return + } + t.mu.Lock() + defer t.mu.Unlock() + now := t.clockLocked() + t.addLocked(t.pairingLocked(pairing, now), now, n) +} + +// Snapshot samples every open connection and returns each pairing's state, +// sorted by pairing ID. +func (t *LiveTracker) Snapshot() LiveSnapshot { + s := LiveSnapshot{Pairings: []LivePairing{}} + if t == nil { + return s + } + t.mu.Lock() + defer t.mu.Unlock() + now := t.clockLocked() + t.sampleAllLocked(now) + for id, p := range t.pairings { + if !p.connected && p.flows == 0 && len(p.samples) == 0 { + delete(t.pairings, id) + continue + } + since := t.lastWall.Add(-now.Sub(p.since)) + s.Pairings = append(s.Pairings, LivePairing{PairingID: id, Connected: p.connected, Since: since.UnixMilli()}) + } + sort.Slice(s.Pairings, func(i, j int) bool { return s.Pairings[i].PairingID < s.Pairings[j].PairingID }) + return s +} + +// Reset forgets everything, for tests. +func (t *LiveTracker) Reset() { + t.mu.Lock() + defer t.mu.Unlock() + clear(t.pairings) + clear(t.flows) +} + +// clockLocked returns now plus slept, so windows and lingers age through a +// sleep. +func (t *LiveTracker) clockLocked() time.Time { + now, wall := t.now(), t.wall() + if !t.lastNow.IsZero() { + // A backward wall-clock step gives a negative skew, which is ignored. + if skew := wall.Sub(t.lastWall) - now.Sub(t.lastNow); skew > liveSleepTolerance { + t.slept += skew + } + } + t.lastNow, t.lastWall = now, wall + return now.Add(t.slept) +} + +// liveSleepTolerance is how far the wall clock may outrun now between reads +// before the difference counts as sleep. Smaller would count ordinary skew +// between the two clocks as sleep; a shorter sleep barely moves the window. +const liveSleepTolerance = time.Second + +func (t *LiveTracker) pairingLocked(id string, now time.Time) *livePairing { + p := t.pairings[id] + if p == nil { + p = &livePairing{since: now} + t.pairings[id] = p + } + return p +} + +func (t *LiveTracker) closeLocked(key any, f *liveFlow, now time.Time) { + t.sampleLocked(f, now) + delete(t.flows, key) + if p := t.pairings[f.pairing]; p != nil { + p.flows-- + t.evaluateLocked(p, now) + } +} + +func (t *LiveTracker) sampleLocked(f *liveFlow, now time.Time) { + if f.read == nil { + return + } + cur := f.read() + gap := now.Sub(f.lastAt) + f.lastAt = now + if cur <= f.last { + return + } + n := cur - f.last + f.last = cur + // After a gap longer than the window, as when the device slept, the + // bytes may all have moved before the window, so they are not counted. + if gap > t.window { + return + } + t.addLocked(t.pairingLocked(f.pairing, now), now, n) +} + +// addLocked counts n bytes for p at now, after settling p's state up to +// now, so bytes arriving after its linger ended reconnect it rather than +// extend the old connection. +func (t *LiveTracker) addLocked(p *livePairing, now time.Time, n uint64) { + t.evaluateLocked(p, now) + p.add(now, n) + t.evaluateLocked(p, now) +} + +func (p *livePairing) add(now time.Time, n uint64) { + if last := len(p.samples) - 1; last >= 0 && now.Sub(p.samples[last].at) < time.Second { + p.samples[last].n += n + return + } + p.samples = append(p.samples, liveSample{at: now, n: n}) +} + +func (t *LiveTracker) evaluateLocked(p *livePairing, now time.Time) { + cutoff := now.Add(-t.window) + var total uint64 + for _, s := range p.samples { + total += s.n + } + // If the window was at the threshold and an expiry took it below, the + // pairing went quiet when that sample expired, not at this later + // evaluation. + var crossedAt time.Time + drop := 0 + for drop < len(p.samples) && !p.samples[drop].at.After(cutoff) { + before := total + total -= p.samples[drop].n + if before >= t.threshold && total < t.threshold { + crossedAt = p.samples[drop].at.Add(t.window) + } + drop++ + } + if drop > 0 { + p.samples = append(p.samples[:0], p.samples[drop:]...) + } + switch { + case total >= t.threshold: + p.quietSince = time.Time{} + if !p.connected { + p.connected = true + p.since = now + } + case p.connected: + if p.quietSince.IsZero() { + p.quietSince = now + if !crossedAt.IsZero() && crossedAt.Before(now) { + p.quietSince = crossedAt + } + } + if now.Sub(p.quietSince) >= t.linger { + p.connected = false + // The linger's end, not this possibly later evaluation. + p.since = p.quietSince.Add(t.linger) + p.quietSince = time.Time{} + } + } +} diff --git a/helper/live_test.go b/helper/live_test.go new file mode 100644 index 00000000..83e3ca03 --- /dev/null +++ b/helper/live_test.go @@ -0,0 +1,387 @@ +package helper + +import ( + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type liveClock struct{ t time.Time } + +func (c *liveClock) now() time.Time { return c.t } +func (c *liveClock) advance(d time.Duration) { c.t = c.t.Add(d) } +func newLiveClock() *liveClock { return &liveClock{t: time.Unix(1_800_000_000, 0)} } +func liveOf(s LiveSnapshot, id string) *LivePairing { + for i := range s.Pairings { + if s.Pairings[i].PairingID == id { + return &s.Pairings[i] + } + } + return nil +} + +// newManualTracker is a tracker that also uses now as its wall clock and has +// no sampler goroutine, so a test's fake clock is only read from the test's +// goroutine. +func newManualTracker(now func() time.Time) *LiveTracker { + lt := NewLiveTracker(now) + lt.interval = 0 + lt.wall = now + return lt +} + +func (lt *LiveTracker) sampleNow() { + lt.mu.Lock() + defer lt.mu.Unlock() + lt.sampleAllLocked(lt.clockLocked()) +} + +// Traffic that moved minutes ago, with nobody reading the tracker since, is +// not "connected now": the sampler stamped it when it moved. +func TestLive_SampledTrafficKeepsItsTime(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + n.Add(LiveThreshold) + clk.advance(time.Second) + lt.sampleNow() // what the sampler goroutine does each second + clk.advance(5 * time.Minute) + p := liveOf(lt.Snapshot(), "pA") + require.NotNil(t, p, "the connection is still open") + assert.False(t, p.Connected, "the burst was five minutes ago") +} + +// A pairing first evaluated long after its traffic left the window went quiet +// when it left, so its linger is long over too. +func TestLive_QuietFromWhenTrafficLeftTheWindow(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + lt.Add("pA", LiveThreshold) + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected) + clk.advance(5 * time.Minute) + assert.Nil(t, liveOf(lt.Snapshot(), "pA"), "no linger five minutes on") +} + +// The linger starts when the expiry that took the window below the threshold +// happened, not when the last expired sample did. +func TestLive_QuietFromTheCrossingNotTheLastExpiry(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + n.Add(LiveThreshold) + lt.sampleNow() // t=0 + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected) + clk.advance(20 * time.Second) + n.Add(1) + lt.Close("c1") // t=20: one more byte, then closed + // The window fell below the threshold at t=60, so the linger ended at + // t=75. + clk.advance(60 * time.Second) // t=80 + assert.Nil(t, liveOf(lt.Snapshot(), "pA"), "not connected at t=80") +} + +// A pairing evaluated only after its linger ended reports the linger's end +// as when it disconnected, not the evaluation. +func TestLive_DisconnectedSinceTheLingerEnded(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + start := clk.t + lt.Open("peer", "c1", "pA", nil) + lt.Add("pA", LiveThreshold) + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected) + clk.advance(90 * time.Second) + p := liveOf(lt.Snapshot(), "pA") + require.NotNil(t, p) + assert.False(t, p.Connected) + assert.Equal(t, start.Add(LiveWindow+LiveLinger).UnixMilli(), p.Since) +} + +// A burst sampled after the linger ended starts a new connection rather +// than extending the old one. +func TestLive_BurstAfterTheLingerReconnects(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + n.Add(LiveThreshold) + lt.sampleNow() + require.True(t, liveOf(lt.Snapshot(), "pA").Connected) + // The sampler ran each second until 74s; the linger ends at 75s, and + // the next sample, at 80s, carries a new burst. + for range 74 { + clk.advance(time.Second) + lt.sampleNow() + } + clk.advance(6 * time.Second) + n.Add(LiveThreshold) + p := liveOf(lt.Snapshot(), "pA") + assert.True(t, p.Connected) + assert.Equal(t, clk.t.UnixMilli(), p.Since, "sampled") + + clk.advance(80 * time.Second) + lt.Add("pA", LiveThreshold) + p = liveOf(lt.Snapshot(), "pA") + assert.True(t, p.Connected) + assert.Equal(t, clk.t.UnixMilli(), p.Since, "added") +} + +// Bytes read after a wall-clock sampling gap longer than the window, as +// after the device slept, may have moved before the window, so they do not +// count. +func TestLive_BytesAfterALongGapDoNotCount(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + lt.sampleNow() + clk.advance(5 * time.Minute) + n.Add(LiveThreshold) + assert.False(t, liveOf(lt.Snapshot(), "pA").Connected, "after a long gap") + + clk.advance(30 * time.Second) + n.Add(LiveThreshold) + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected, "after an ordinary gap") +} + +// After a sleep the monotonic clock has moved one tick and the wall clock +// minutes: the bytes read then do not count. +func TestLive_BytesAfterASleepDoNotCount(t *testing.T) { + mono, wall := newLiveClock(), newLiveClock() + lt := newManualTracker(mono.now) + lt.wall = wall.now + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + lt.sampleNow() + + mono.advance(time.Second) + wall.advance(5 * time.Minute) + n.Add(LiveThreshold) + assert.False(t, liveOf(lt.Snapshot(), "pA").Connected) +} + +// A pairing connected just before the device slept is disconnected on waking, +// dated when its linger ended during the sleep, though the monotonic clock +// barely moved; one with no connection open is dropped. +func TestLive_SleepExpiresTheWindowAndLinger(t *testing.T) { + for _, open := range []bool{true, false} { + mono, wall := newLiveClock(), newLiveClock() + lt := newManualTracker(mono.now) + lt.wall = wall.now + start := wall.t + if open { + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + n.Add(LiveThreshold) + lt.sampleNow() + } else { + lt.Add("pA", LiveThreshold) + } + require.True(t, liveOf(lt.Snapshot(), "pA").Connected) + + mono.advance(time.Second) + wall.advance(time.Hour) + s := lt.Snapshot() + if !open { + assert.Nil(t, liveOf(s, "pA"), "nothing open") + continue + } + p := liveOf(s, "pA") + require.NotNil(t, p) + assert.False(t, p.Connected) + assert.Equal(t, start.Add(LiveWindow+LiveLinger).UnixMilli(), p.Since) + } +} + +// Samples keep the clock's monotonic reading, so a backward wall-clock step +// does not move them. +func TestLive_SamplesKeepTheMonotonicClock(t *testing.T) { + lt := newManualTracker(time.Now) + lt.Add("pA", 1) + lt.mu.Lock() + defer lt.mu.Unlock() + require.Len(t, lt.pairings["pA"].samples, 1) + assert.Contains(t, lt.pairings["pA"].samples[0].at.String(), "m=", "monotonic reading kept") +} + +// The sampler runs by itself while a connection is open, with no Snapshot +// calls, and stops once none is. +func TestLive_SamplerRunsOnlyWhileConnectionsAreOpen(t *testing.T) { + lt := NewLiveTracker(nil) + lt.interval = 5 * time.Millisecond + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + n.Add(100) + require.Eventually(t, func() bool { + lt.mu.Lock() + defer lt.mu.Unlock() + p := lt.pairings["pA"] + return p != nil && len(p.samples) > 0 + }, 2*time.Second, time.Millisecond, "bytes sampled without a Snapshot") + lt.Close("c1") + require.Eventually(t, func() bool { + lt.mu.Lock() + defer lt.mu.Unlock() + return !lt.sampling + }, 2*time.Second, time.Millisecond, "the sampler exits with no connection open") + lt.Open("peer", "c2", "pA", n.Load) + lt.mu.Lock() + assert.True(t, lt.sampling, "and restarts with the next") + lt.mu.Unlock() + lt.Close("c2") +} + +func TestLive_BelowThresholdIsNotConnected(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + // A url-test probe's worth of bytes every few seconds stays under 32 KiB + // in any minute. + for range 20 { + n.Add(1000) + clk.advance(3 * time.Second) + p := liveOf(lt.Snapshot(), "pA") + require.NotNil(t, p, "an open connection is listed") + assert.False(t, p.Connected) + } +} + +func TestLive_ThresholdWindowAndLinger(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + + n.Add(LiveThreshold - 1) + clk.advance(time.Second) + assert.False(t, liveOf(lt.Snapshot(), "pA").Connected, "one byte short") + + n.Add(1) + clk.advance(time.Second) + start := clk.t + p := liveOf(lt.Snapshot(), "pA") + assert.True(t, p.Connected, "at the threshold") + assert.Equal(t, start.UnixMilli(), p.Since) + + // The bytes stay in the window for 60s, so it stays connected without + // more traffic. + clk.advance(57 * time.Second) + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected) + + // Out of the window: quiet, but lingers. It went below the threshold at + // T+61, when the first sample expired, so the linger ends at T+76. + clk.advance(3 * time.Second) // T+62 + p = liveOf(lt.Snapshot(), "pA") + assert.True(t, p.Connected, "lingers") + assert.Equal(t, start.UnixMilli(), p.Since) + clk.advance(LiveLinger - 2*time.Second) // T+75 + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected, "still lingering") + clk.advance(time.Second) + p = liveOf(lt.Snapshot(), "pA") + assert.False(t, p.Connected, "linger over") + assert.Equal(t, clk.t.UnixMilli(), p.Since) +} + +func TestLive_LingerAbsorbsAShortDip(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + lt.Add("pA", LiveThreshold) + since := clk.t.UnixMilli() + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected) + clk.advance(61 * time.Second) + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected, "dip below") + clk.advance(5 * time.Second) + lt.Add("pA", LiveThreshold) + clk.advance(LiveLinger) + p := liveOf(lt.Snapshot(), "pA") + assert.True(t, p.Connected, "traffic back within the linger") + assert.Equal(t, since, p.Since, "no flap") +} + +func TestLive_MultiplePairingsAreIndependent(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var a, b atomic.Uint64 + lt.Open("peer", "a", "pA", a.Load) + lt.Open("unbounded", "b", "pB", b.Load) + lt.Add("pC", 10) + a.Add(LiveThreshold) + b.Add(100) + clk.advance(time.Second) + s := lt.Snapshot() + require.Len(t, s.Pairings, 3) + assert.Equal(t, []string{"pA", "pB", "pC"}, []string{s.Pairings[0].PairingID, s.Pairings[1].PairingID, s.Pairings[2].PairingID}) + assert.True(t, liveOf(s, "pA").Connected) + assert.False(t, liveOf(s, "pB").Connected) + assert.False(t, liveOf(s, "pC").Connected) +} + +func TestLive_BytesFromSeveralConnectionsAndAddSum(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var a, b atomic.Uint64 + lt.Open("peer", "a", "pA", a.Load) + lt.Open("peer", "b", "pA", b.Load) + a.Add(LiveThreshold / 4) + b.Add(LiveThreshold / 4) + lt.Add("pA", LiveThreshold/4) + clk.advance(time.Second) + assert.False(t, liveOf(lt.Snapshot(), "pA").Connected) + lt.Add("pA", LiveThreshold/4) + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected) +} + +func TestLive_CloseCountsFinalBytesAndPrunes(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + var n atomic.Uint64 + lt.Open("peer", "c1", "pA", n.Load) + n.Add(LiveThreshold) + lt.Close("c1") + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected, "the closed connection's bytes count") + clk.advance(LiveWindow + time.Second) + assert.True(t, liveOf(lt.Snapshot(), "pA").Connected, "lingers") + clk.advance(LiveLinger) + assert.Nil(t, liveOf(lt.Snapshot(), "pA"), "a disconnected pairing with nothing open is dropped") +} + +func TestLive_CloseOwnerOnlyClosesThatOwner(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + lt.Open("peer", "a", "pA", nil) + lt.Open("unbounded", 1, "pB", nil) + lt.CloseOwner("peer") + s := lt.Snapshot() + assert.Nil(t, liveOf(s, "pA")) + assert.NotNil(t, liveOf(s, "pB")) + lt.Close(1) + assert.Empty(t, lt.Snapshot().Pairings) +} + +func TestLive_IgnoresPoolAndReopen(t *testing.T) { + clk := newLiveClock() + lt := newManualTracker(clk.now) + lt.Open("peer", "a", "", nil) + lt.Add("", LiveThreshold) + assert.Empty(t, lt.Snapshot().Pairings, "pool connections are not tracked") + + lt.Open("peer", "a", "pA", nil) + lt.Open("peer", "a", "pB", nil) // same key reused: the first is closed + s := lt.Snapshot() + assert.Nil(t, liveOf(s, "pA")) + assert.NotNil(t, liveOf(s, "pB")) +} + +func TestLive_NilTrackerIsSafe(t *testing.T) { + var lt *LiveTracker + lt.Open("peer", "a", "pA", nil) + lt.Add("pA", 1) + lt.Close("a") + lt.CloseOwner("peer") + assert.Empty(t, lt.Snapshot().Pairings) +} diff --git a/internal/testutil/logargs.go b/internal/testutil/logargs.go new file mode 100644 index 00000000..5b9bffe1 --- /dev/null +++ b/internal/testutil/logargs.go @@ -0,0 +1,72 @@ +package testutil + +import ( + "go/ast" + "go/parser" + "go/token" + "os" + "path/filepath" + "strings" + "testing" +) + +// LoggedIdentifiers fails t for every slog call in the non-test Go files of +// dir whose arguments use an identifier forbidden reports true for. It is how +// a package shows that a value, such as a helper pairing ID, never reaches its +// logs, without replacing the default logger in a test. +func LoggedIdentifiers(t *testing.T, dir string, forbidden func(name string) bool) { + t.Helper() + files, err := filepath.Glob(filepath.Join(dir, "*.go")) + if err != nil { + t.Fatal(err) + } + fset := token.NewFileSet() + calls := 0 + for _, path := range files { + if strings.HasSuffix(path, "_test.go") { + continue + } + src, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + f, err := parser.ParseFile(fset, path, src, 0) + if err != nil { + t.Fatal(err) + } + ast.Inspect(f, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok { + return true + } + if pkg, ok := sel.X.(*ast.Ident); !ok || pkg.Name != "slog" || !isLogFunc(sel.Sel.Name) { + return true + } + calls++ + for _, arg := range call.Args { + ast.Inspect(arg, func(n ast.Node) bool { + if id, ok := n.(*ast.Ident); ok && forbidden(id.Name) { + t.Errorf("%s: slog.%s logs %s", fset.Position(id.Pos()), sel.Sel.Name, id.Name) + } + return true + }) + } + return true + }) + } + if calls == 0 { + t.Fatalf("no slog calls found in %s", dir) + } +} + +func isLogFunc(name string) bool { + switch name { + case "Debug", "Info", "Warn", "Error", "Log", "DebugContext", "InfoContext", "WarnContext", "ErrorContext", "LogAttrs": + return true + } + return false +} diff --git a/ipc/client.go b/ipc/client.go index 2aa1039e..13e25a68 100644 --- a/ipc/client.go +++ b/ipc/client.go @@ -914,3 +914,11 @@ func (c *Client) HelperActivity(ctx context.Context) (helper.Activity, error) { err := c.doJSON(ctx, http.MethodGet, helperActivityEndpoint, nil, &a) return a, err } + +// HelperLive reads whether each helper pairing's friend is connected to this +// volunteer now (see helper.LiveSnapshot). +func (c *Client) HelperLive(ctx context.Context) (helper.LiveSnapshot, error) { + var s helper.LiveSnapshot + err := c.doJSON(ctx, http.MethodGet, helperLiveEndpoint, nil, &s) + return s, err +} diff --git a/ipc/helper_test.go b/ipc/helper_test.go index 8538c749..b0e36a25 100644 --- a/ipc/helper_test.go +++ b/ipc/helper_test.go @@ -129,6 +129,43 @@ func TestHelperActivityClientRoundTrip(t *testing.T) { assert.Zero(t, a.RelayedBytes) } +func TestHelperLiveClientRoundTrip(t *testing.T) { + helper.DefaultLive.Reset() + t.Cleanup(helper.DefaultLive.Reset) + c := newClient() + c.http.Transport = localTransport{api: newLocalAPI(&backend.LocalBackend{}, false)} + + s, err := c.HelperLive(context.Background()) + require.NoError(t, err) + assert.Empty(t, s.Pairings) + + helper.DefaultLive.Open("test", "k", "pairing-1", nil) + helper.DefaultLive.Add("pairing-1", helper.LiveThreshold) + s, err = c.HelperLive(context.Background()) + require.NoError(t, err) + require.Len(t, s.Pairings, 1) + assert.Equal(t, "pairing-1", s.Pairings[0].PairingID) + assert.True(t, s.Pairings[0].Connected) + assert.NotZero(t, s.Pairings[0].Since) + + w := serveHelper(t, newLocalAPI(&backend.LocalBackend{}, false), http.MethodGet, helperLiveEndpoint, nil) + require.Equal(t, http.StatusOK, w.Code) + var raw struct { + Pairings []map[string]any `json:"pairings"` + } + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &raw)) + require.Len(t, raw.Pairings, 1) + assert.ElementsMatch(t, []string{"pairing_id", "connected", "since"}, mapKeys(raw.Pairings[0])) +} + +func mapKeys(m map[string]any) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +} + // A body that is not JSON is refused before marshalling, with fixed text // rather than an encoder error that could quote it. func TestHelperRelayClientRefusesANonJSONBody(t *testing.T) { @@ -165,6 +202,8 @@ func TestLoggerSkipsHelperRoutes(t *testing.T) { assert.False(t, ok, "relay") _, ok = routeOf(http.MethodGet, helperActivityEndpoint) assert.False(t, ok, "activity") + _, ok = routeOf(http.MethodGet, helperLiveEndpoint) + assert.False(t, ok, "live") route, ok := routeOf(http.MethodGet, "/not-a-route") assert.True(t, ok) assert.Equal(t, "/not-a-route", route) diff --git a/ipc/server.go b/ipc/server.go index c64bcee9..6eecfd92 100644 --- a/ipc/server.go +++ b/ipc/server.go @@ -93,6 +93,7 @@ const ( // Helper invites helperRelayEndpoint = "/helper/relay" helperActivityEndpoint = "/helper/activity" + helperLiveEndpoint = "/helper/live" // Split tunnel endpoint splitTunnelEndpoint = "/split-tunnel" @@ -298,6 +299,8 @@ func newLocalAPI(b *backend.LocalBackend, withAuth bool) *localapi { // Not traced either: a traced poll would mark the device as a helper // volunteer in exported telemetry. mux.HandleFunc("GET "+helperActivityEndpoint, s.helperActivityHandler) + // Not traced either, for the same reason. + mux.HandleFunc("GET "+helperLiveEndpoint, s.helperLiveHandler) // Split tunnel mux.HandleFunc(splitTunnelEndpoint, traced(s.splitTunnelHandler)) @@ -1591,3 +1594,7 @@ func (s *localapi) helperRelayHandler(w http.ResponseWriter, r *http.Request) { func (s *localapi) helperActivityHandler(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, s.backend(r.Context()).HelperActivity()) } + +func (s *localapi) helperLiveHandler(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.backend(r.Context()).HelperLive()) +} diff --git a/peer/api.go b/peer/api.go index 11ccffd8..c22dd992 100644 --- a/peer/api.go +++ b/peer/api.go @@ -43,6 +43,18 @@ type RegisterResponse struct { // in Unix milliseconds, so the peer never replaces a set with an older // one. Non-zero exactly when PairingShortIDsKnown is true. PairingShortIDsRevision Revision `json:"pairing_short_ids_revision"` + // PairingShortIDSets is PairingShortIDs grouped by pairing, so the peer + // can tell which pairing a connection came through. Joined in order, the + // groups' IDs are PairingShortIDs. Covered by the same Known and + // Revision. Absent from servers that predate it. + PairingShortIDSets []PairingShortIDSet `json:"pairing_short_id_sets"` +} + +// PairingShortIDSet is one pairing's short IDs. PairingID identifies a +// friend relationship and must never be logged or persisted. +type PairingShortIDSet struct { + PairingID string `json:"pairing_id"` + ShortIDs []string `json:"short_ids"` } // Revision is an int64 that decodes from a JSON number or, as lantern-cloud's @@ -73,6 +85,8 @@ type HeartbeatResponse struct { PairingShortIDs []string `json:"pairing_short_ids"` PairingShortIDsKnown bool `json:"pairing_short_ids_known"` PairingShortIDsRevision Revision `json:"pairing_short_ids_revision"` + // PairingShortIDSets is as for RegisterResponse. + PairingShortIDSets []PairingShortIDSet `json:"pairing_short_id_sets"` } type LifecycleRequest struct { diff --git a/peer/pairing_test.go b/peer/pairing_test.go new file mode 100644 index 00000000..0dbf5f41 --- /dev/null +++ b/peer/pairing_test.go @@ -0,0 +1,556 @@ +package peer + +import ( + "context" + "encoding/hex" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/getlantern/lantern-box/tracker/peerconn" + + "github.com/getlantern/radiance/events" + "github.com/getlantern/radiance/helper" + "github.com/getlantern/radiance/internal/testutil" +) + +const ( + pairingA = "a1b2c3d4-0000-4000-8000-00000000000a" + pairingB = "a1b2c3d4-0000-4000-8000-00000000000b" + sidA1 = "1111111111111111" + sidA2 = "1111111111112222" + sidB1 = "2222222222222222" +) + +func sid(t *testing.T, s string) [8]byte { + t.Helper() + raw, err := hex.DecodeString(s) + require.NoError(t, err) + require.Len(t, raw, 8) + return [8]byte(raw) +} + +func groupsAB() []PairingShortIDSet { + return []PairingShortIDSet{ + {PairingID: pairingA, ShortIDs: []string{sidA1, sidA2}}, + {PairingID: pairingB, ShortIDs: []string{sidB1}}, + } +} + +func TestAPI_DecodesPairingShortIDSets(t *testing.T) { + const body = `{"route_id":"r","pairing_short_ids":["1111111111111111","2222222222222222"],` + + `"pairing_short_ids_known":true,"pairing_short_ids_revision":"1700000000000",` + + `"pairing_short_id_sets":[{"pairing_id":"` + pairingA + `","short_ids":["1111111111111111"]},` + + `{"pairing_id":"` + pairingB + `","short_ids":["2222222222222222"]}]}` + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(body)) + })) + defer srv.Close() + api := NewAPI(srv.Client(), srv.URL, "dev") + want := []PairingShortIDSet{ + {PairingID: pairingA, ShortIDs: []string{sidA1}}, + {PairingID: pairingB, ShortIDs: []string{sidB1}}, + } + + reg, err := api.Register(context.Background(), RegisterRequest{}) + require.NoError(t, err) + assert.Equal(t, want, reg.PairingShortIDSets) + assert.Equal(t, Revision(1700000000000), reg.PairingShortIDsRevision) + + hb, err := api.Heartbeat(context.Background(), "r", 0) + require.NoError(t, err) + assert.Equal(t, want, hb.PairingShortIDSets) + assert.True(t, hb.PairingShortIDsKnown) + + // An older server sends no groups. + var old HeartbeatResponse + require.NoError(t, json.Unmarshal([]byte(`{"pairing_short_ids":["1111111111111111"],"pairing_short_ids_known":true,"pairing_short_ids_revision":5}`), &old)) + assert.Nil(t, old.PairingShortIDSets) +} + +func TestGroupPairingShortIDs(t *testing.T) { + flat := []string{sidA1, sidA2, sidB1} + + m, ok := groupPairingShortIDs(flat, groupsAB(), []string{testBaseShortID}) + require.True(t, ok) + assert.Equal(t, map[[8]byte]string{sid(t, sidA1): pairingA, sid(t, sidA2): pairingA, sid(t, sidB1): pairingB}, m) + + upper := []string{strings.ToUpper(sidA1), sidA2, sidB1} + _, ok = groupPairingShortIDs(upper, groupsAB(), nil) + assert.True(t, ok, "hex case is ignored") + + for name, f := range map[string][]string{ + "reordered": {sidB1, sidA1, sidA2}, + "shorter": {sidA1, sidA2}, + "longer": {sidA1, sidA2, sidB1, "3333333333333333"}, + "different": {sidA1, sidA2, "3333333333333333"}, + "empty": nil, + } { + _, ok := groupPairingShortIDs(f, groupsAB(), nil) + assert.False(t, ok, name) + } + _, ok = groupPairingShortIDs(flat, nil, nil) + assert.False(t, ok, "no groups") + + // The launch config's own ID never names a pairing, nor does a bad entry + // or a group without a pairing ID. + groups := []PairingShortIDSet{ + {PairingID: pairingA, ShortIDs: []string{testBaseShortID, "zz"}}, + {PairingID: "", ShortIDs: []string{sidB1}}, + } + m, ok = groupPairingShortIDs([]string{testBaseShortID, "zz", sidB1}, groups, []string{testBaseShortID}) + require.True(t, ok) + assert.Empty(t, m) +} + +// pairingClient is a started client whose register answered flat at revision +// 100 with no groups, as an older server replica does. +func pairingClient(t *testing.T, opts ...func(*Config)) *Client { + t.Helper() + srv := newStubServer(t) + srv.registerResp.ServerConfig = launchCfgWithShortID() + srv.registerResp.PairingShortIDs = []string{sidA1, sidA2, sidB1} + srv.registerResp.PairingShortIDsKnown = true + srv.registerResp.PairingShortIDsRevision = 100 + opts = append([]func(*Config){func(cfg *Config) { cfg.HeartbeatInterval = time.Hour }}, opts...) + c := newTestClient(t, &fakeForwarder{}, &fakeBoxService{}, srv, opts...) + require.NoError(t, c.Start(context.Background())) + t.Cleanup(func() { _ = c.Stop(context.Background()) }) + return c +} + +func TestClient_GroupsAcceptedAtTheAppliedRevisionWhenTheyMatch(t *testing.T) { + c := pairingClient(t) + assert.Equal(t, "", c.pairingFor(sid(t, sidA1)), "no groups yet") + + // A newer replica answers with groups at the revision already applied. + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 100) + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA1))) + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA2))) + assert.Equal(t, pairingB, c.pairingFor(sid(t, sidB1))) + + // And an older replica answering flat at the same revision afterwards + // leaves them alone. + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, nil, true, 100) + assert.Equal(t, pairingB, c.pairingFor(sid(t, sidB1))) + ids, _ := pairingState(c) + assert.Equal(t, []string{sidA1, sidA2, sidB1}, ids) +} + +func TestClient_GroupsRejectedAtTheAppliedRevisionWhenTheyDiffer(t *testing.T) { + c := pairingClient(t) + mismatched := []PairingShortIDSet{ + {PairingID: pairingA, ShortIDs: []string{sidA1}}, + {PairingID: pairingB, ShortIDs: []string{sidB1}}, + } + // The flat list in the response is ignored at this revision; the groups + // are checked against the set applied, which they do not describe. + c.offerPairingShortIDs([]string{sidA1, sidB1}, mismatched, true, 100) + assert.Equal(t, "", c.pairingFor(sid(t, sidA1))) + assert.Equal(t, "", c.pairingFor(sid(t, sidB1))) + ids, _ := pairingState(c) + assert.Equal(t, []string{sidA1, sidA2, sidB1}, ids, "the flat set is unchanged") +} + +func TestClient_GroupsFollowTheStrictlyNewerRuleOtherwise(t *testing.T) { + c := pairingClient(t) + + // Older than the applied revision: neither the set nor its groups. + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 99) + assert.Equal(t, "", c.pairingFor(sid(t, sidA1))) + + // Newer, with groups that do not match their own set: the set is taken, + // the groups are not. + c.offerPairingShortIDs([]string{sidA1, sidB1}, groupsAB(), true, 200) + ids, _ := pairingState(c) + assert.Equal(t, []string{sidA1, sidB1}, ids) + assert.Equal(t, "", c.pairingFor(sid(t, sidA1))) + + // Newer with matching groups: both. + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 300) + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA2))) + + // Newer without groups: the set is taken, and the newest grouped set's + // mapping stays. + c.offerPairingShortIDs([]string{sidB1}, nil, true, 400) + ids, _ = pairingState(c) + assert.Equal(t, []string{sidB1}, ids) + assert.Equal(t, pairingB, c.pairingFor(sid(t, sidB1))) + + // A newer grouped set replaces the mapping, keeping the groups it + // replaced until a different grouped set replaces it. + onlyB := []PairingShortIDSet{{PairingID: pairingB, ShortIDs: []string{sidB1}}} + c.offerPairingShortIDs([]string{sidB1}, onlyB, true, 500) + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA1)), "the replaced groups are kept") + assert.Equal(t, pairingB, c.pairingFor(sid(t, sidB1))) + c.offerPairingShortIDs([]string{sidB1}, onlyB, true, 600) + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA1)), "an unchanged set keeps them") + onlyA2 := []PairingShortIDSet{{PairingID: pairingA, ShortIDs: []string{sidA2}}} + c.offerPairingShortIDs([]string{sidA2}, onlyA2, true, 700) + assert.Equal(t, "", c.pairingFor(sid(t, sidA1)), "and a different one drops them") + assert.Equal(t, pairingB, c.pairingFor(sid(t, sidB1))) + + // An unknown response changes nothing. + c.offerPairingShortIDs(nil, groupsAB(), false, 0) + assert.Equal(t, "", c.pairingFor(sid(t, sidA1))) +} + +// A repeated response with the held groups keeps the groups they replaced, +// which a rotation's box may still be serving. +func TestClient_RepeatedGroupsKeepThePreviousOnes(t *testing.T) { + c := pairingClient(t) + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 100) + onlyB := []PairingShortIDSet{{PairingID: pairingB, ShortIDs: []string{sidB1}}} + c.offerPairingShortIDs([]string{sidB1}, onlyB, true, 200) + c.offerPairingShortIDs([]string{sidB1}, onlyB, true, 200) + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA1))) + assert.Equal(t, pairingB, c.pairingFor(sid(t, sidB1))) +} + +func TestClient_HeartbeatGroupsAtTheSameRevision(t *testing.T) { + srv := newStubServer(t) + srv.registerResp.ServerConfig = launchCfgWithShortID() + srv.registerResp.PairingShortIDs = []string{sidA1, sidA2, sidB1} + srv.registerResp.PairingShortIDsKnown = true + srv.registerResp.PairingShortIDsRevision = 100 + srv.heartbeatRevision.Store(100) + srv.heartbeatShortIDs.Store(&[]string{sidA1, sidA2, sidB1}) + sets := groupsAB() + srv.heartbeatShortIDSets.Store(&sets) + c := newTestClient(t, &fakeForwarder{}, &fakeBoxService{}, srv, func(cfg *Config) { cfg.HeartbeatInterval = 20 * time.Millisecond }) + require.NoError(t, c.Start(context.Background())) + t.Cleanup(func() { _ = c.Stop(context.Background()) }) + require.Eventually(t, func() bool { return c.pairingFor(sid(t, sidB1)) == pairingB }, 2*time.Second, 10*time.Millisecond) +} + +func TestClient_StartTakesRegisterGroups(t *testing.T) { + srv := newStubServer(t) + srv.registerResp.ServerConfig = launchCfgWithShortID() + srv.registerResp.PairingShortIDs = []string{sidA1, sidA2, sidB1} + srv.registerResp.PairingShortIDsKnown = true + srv.registerResp.PairingShortIDSets = groupsAB() + c := newTestClient(t, &fakeForwarder{}, &fakeBoxService{}, srv, func(cfg *Config) { cfg.HeartbeatInterval = time.Hour }) + require.NoError(t, c.Start(context.Background())) + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA1))) + require.NoError(t, c.Stop(context.Background())) + assert.Equal(t, "", c.pairingFor(sid(t, sidA1)), "stop forgets the mapping") +} + +func TestClient_ConnectionEventsCarryThePairing(t *testing.T) { + live := helper.NewLiveTracker(nil) + c := pairingClient(t, func(cfg *Config) { cfg.Live = live }) + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 100) + + got := make(chan ConnectionEvent, 16) + sub := events.Subscribe(func(evt ConnectionEvent) { got <- evt }) + defer sub.Unsubscribe() + next := func() ConnectionEvent { + t.Helper() + select { + case evt := <-got: + return evt + case <-time.After(2 * time.Second): + t.Fatal("no connection event") + return ConnectionEvent{} + } + } + + friend := counter(0, 0) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: friend, ShortID: sid(t, sidA2)}) + assert.Equal(t, pairingA, next().Pairing, "a pairing's short ID") + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.2:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, testBaseShortID)}) + assert.Equal(t, "", next().Pairing, "the launch config's own short ID is the pool") + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.3:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, "3333333333333333")}) + assert.Equal(t, "", next().Pairing, "an unknown short ID") + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.4:1000", Destination: "example.com:443", Bytes: counter(0, 0)}) + assert.Equal(t, "", next().Pairing, "no short ID") + + // The snapshot labels the friend's IP for the globe. + for _, pb := range c.RelaySnapshot().PeerBytes { + if pb.IP == "198.51.100.1" { + assert.Equal(t, pairingA, pb.Pairing) + } else { + assert.Equal(t, "", pb.Pairing, pb.IP) + } + } + + // And the LiveTracker sees the friend once real traffic moves. + friend.AddDown(helper.LiveThreshold) + s := live.Snapshot() + require.Len(t, s.Pairings, 1) + assert.Equal(t, pairingA, s.Pairings[0].PairingID) + assert.True(t, s.Pairings[0].Connected) + + peerconn.Notify(peerconn.Event{State: -1, Source: "198.51.100.1:1000", Bytes: friend, ShortID: sid(t, sidA2)}) + evt := next() + assert.Equal(t, -1, evt.State) + assert.Equal(t, pairingA, evt.Pairing) + + data, err := json.Marshal(evt) + require.NoError(t, err) + assert.Contains(t, string(data), `"pairing":"`+pairingA+`"`) +} + +func TestClient_StopClosesLiveConnections(t *testing.T) { + live := helper.NewLiveTracker(nil) + c := pairingClient(t, func(cfg *Config) { cfg.Live = live }) + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 100) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, sidB1)}) + require.Len(t, live.Snapshot().Pairings, 1) + require.NoError(t, c.Stop(context.Background())) + assert.Empty(t, live.Snapshot().Pairings, "an idle pairing whose connection Stop dropped is gone") +} + +// Pairing IDs and short IDs never reach the logs: no slog call in the package +// is passed a pairing, a short ID, or a whole event that carries one. +func TestPairingIsNeverLogged(t *testing.T) { + testutil.LoggedIdentifiers(t, ".", func(name string) bool { + l := strings.ToLower(name) + return strings.Contains(l, "pairing") || strings.Contains(l, "shortid") || name == "evt" + }) +} + +// An accept callback that pauses after resolving its pairing, while Stop +// ends the session, must not add a connection once it resumes: Stop would +// have nothing left to close it, and its close is dropped while draining. +func TestClient_AcceptRacingStopLeavesNoLiveConnection(t *testing.T) { + live := helper.NewLiveTracker(nil) + c := pairingClient(t, func(cfg *Config) { cfg.Live = live }) + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 100) + + paused, resume := make(chan struct{}), make(chan struct{}) + c.beforeTrackLive = func() { + close(paused) + <-resume + } + notified := make(chan struct{}) + go func() { + defer close(notified) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, sidA1)}) + }() + <-paused + require.NoError(t, c.Stop(context.Background())) + close(resume) + <-notified + + assert.Empty(t, live.Snapshot().Pairings, "the stopped session's late accept was not tracked") +} + +// A friend connected before the server's grouped set arrived (an older +// replica answered register flat) is attributed once it does, for the rest +// of the stream's life. +func TestClient_StreamAcceptedBeforeGroupsIsAttributedLater(t *testing.T) { + live := helper.NewLiveTracker(nil) + c := pairingClient(t, func(cfg *Config) { cfg.Live = live }) + + friend, pool := counter(0, 0), counter(0, 0) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: friend, ShortID: sid(t, sidA1)}) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.2:1000", Destination: "example.com:443", Bytes: pool, ShortID: sid(t, testBaseShortID)}) + assert.Empty(t, live.Snapshot().Pairings, "no groups yet") + + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 100) + friend.AddDown(helper.LiveThreshold) + pool.AddDown(helper.LiveThreshold) + s := live.Snapshot() + require.Len(t, s.Pairings, 1) + assert.Equal(t, pairingA, s.Pairings[0].PairingID) + assert.True(t, s.Pairings[0].Connected) + got := map[string]string{} + for _, pb := range c.RelaySnapshot().PeerBytes { + got[pb.IP] = pb.Pairing + } + assert.Equal(t, map[string]string{"198.51.100.1": pairingA, "198.51.100.2": ""}, got) + + peerconn.Notify(peerconn.Event{State: -1, Source: "198.51.100.1:1000", Bytes: friend, ShortID: sid(t, sidA1)}) + peerconn.Notify(peerconn.Event{State: -1, Source: "198.51.100.2:1000", Bytes: pool, ShortID: sid(t, testBaseShortID)}) + c.liveMu.Lock() + assert.Empty(t, c.conns, "closed streams are forgotten") + c.liveMu.Unlock() +} + +// Unresolved streams are dropped with their session, and a grouped set +// arriving afterwards opens nothing. +func TestClient_StopForgetsUnresolvedStreams(t *testing.T) { + live := helper.NewLiveTracker(nil) + c := pairingClient(t, func(cfg *Config) { cfg.Live = live }) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, sidA1)}) + c.liveMu.Lock() + require.Len(t, c.conns, 1) + c.liveMu.Unlock() + require.NoError(t, c.Stop(context.Background())) + c.liveMu.Lock() + assert.Empty(t, c.conns) + c.liveMu.Unlock() + c.setPairingByShortID(map[[8]byte]string{sid(t, sidA1): pairingA}) + assert.Empty(t, live.Snapshot().Pairings) +} + +// Attributing an older stream late does not relabel its IP when a newer +// stream from that IP came after it. +func TestClient_LateAttributionKeepsTheLatestStreamsLabel(t *testing.T) { + c := pairingClient(t, func(cfg *Config) { cfg.Live = helper.NewLiveTracker(nil) }) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, sidA1)}) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1001", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, testBaseShortID)}) + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 100) + snap := c.RelaySnapshot() + require.Len(t, snap.PeerBytes, 1) + assert.Equal(t, "", snap.PeerBytes[0].Pairing, "the IP's latest stream is the pool one") + + // A stream resolved at accept still labels its IP. + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.2:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, sidB1)}) + for _, pb := range c.RelaySnapshot().PeerBytes { + if pb.IP == "198.51.100.2" { + assert.Equal(t, pairingB, pb.Pairing) + } + } +} + +// A close reports the pairing its accept, or a later grouped set, attributed +// the stream to, even after a newer set dropped its short ID. +func TestClient_CloseKeepsTheStreamsPairing(t *testing.T) { + c := pairingClient(t, func(cfg *Config) { cfg.Live = helper.NewLiveTracker(nil) }) + c.offerPairingShortIDs([]string{sidA1, sidA2, sidB1}, groupsAB(), true, 100) + + got := make(chan ConnectionEvent, 16) + sub := events.Subscribe(func(evt ConnectionEvent) { got <- evt }) + defer sub.Unsubscribe() + closeOf := func(src string) string { + t.Helper() + for { + select { + case evt := <-got: + if evt.State < 0 && evt.Source == src { + return evt.Pairing + } + case <-time.After(2 * time.Second): + t.Fatal("no close event") + return "" + } + } + } + + accepted := counter(0, 0) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: accepted, ShortID: sid(t, sidA1)}) + // sidA1 leaves the newest set, as it does when the epoch rotates. + c.offerPairingShortIDs([]string{sidB1}, []PairingShortIDSet{{PairingID: pairingB, ShortIDs: []string{sidB1}}}, true, 200) + peerconn.Notify(peerconn.Event{State: -1, Source: "198.51.100.1:1000", Bytes: accepted, ShortID: sid(t, sidA1)}) + assert.Equal(t, pairingA, closeOf("198.51.100.1:1000"), "attributed at accept") + + late := counter(0, 0) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.2:1000", Destination: "example.com:443", Bytes: late, ShortID: sid(t, sidA2)}) + c.offerPairingShortIDs([]string{sidA2}, []PairingShortIDSet{{PairingID: pairingA, ShortIDs: []string{sidA2}}}, true, 300) + c.offerPairingShortIDs([]string{sidB1}, []PairingShortIDSet{{PairingID: pairingB, ShortIDs: []string{sidB1}}}, true, 400) + peerconn.Notify(peerconn.Event{State: -1, Source: "198.51.100.2:1000", Bytes: late, ShortID: sid(t, sidA2)}) + assert.Equal(t, pairingA, closeOf("198.51.100.2:1000"), "attributed late") +} + +// A grouped set the inbound refused does not replace the map: the inbound +// still accepts the previous IDs, so they keep their pairings until a retry +// pushes the new set, which then also attributes streams waiting for it. +func TestClient_GroupsReplaceTheMapOnlyOnceTheSetIsPushed(t *testing.T) { + srv := newStubServer(t) + srv.registerResp.ServerConfig = launchCfgWithShortID() + srv.registerResp.PairingShortIDs = []string{sidA1} + srv.registerResp.PairingShortIDsKnown = true + srv.registerResp.PairingShortIDsRevision = 100 + srv.registerResp.PairingShortIDSets = []PairingShortIDSet{{PairingID: pairingA, ShortIDs: []string{sidA1}}} + b := &fakeBoxService{} + live := helper.NewLiveTracker(nil) + c := newTestClient(t, &fakeForwarder{}, b, srv, func(cfg *Config) { + cfg.HeartbeatInterval = time.Hour + cfg.Live = live + }) + require.NoError(t, c.Start(context.Background())) + t.Cleanup(func() { _ = c.Stop(context.Background()) }) + require.Equal(t, pairingA, c.pairingFor(sid(t, sidA1))) + + b.setUpdateErr(errors.New("refused")) + c.offerPairingShortIDs([]string{sidB1}, []PairingShortIDSet{{PairingID: pairingB, ShortIDs: []string{sidB1}}}, true, 200) + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA1)), "the inbound still accepts it") + assert.Equal(t, "", c.pairingFor(sid(t, sidB1)), "not pushed yet") + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.2:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, sidB1)}) + + b.setUpdateErr(nil) + c.offerPairingShortIDs(nil, nil, false, 0) // a heartbeat's retry + assert.Equal(t, pairingA, c.pairingFor(sid(t, sidA1)), "the replaced groups are kept until the next set") + assert.Equal(t, pairingB, c.pairingFor(sid(t, sidB1))) + s := live.Snapshot() + require.Len(t, s.Pairings, 1, "the stream waiting for the set is attributed") + assert.Equal(t, pairingB, s.Pairings[0].PairingID) +} + +// A stream the inbound accepts on a new ID as soon as the set is pushed, and +// that ends before the push returns, is attributed: the groups are published +// before the push. +func TestClient_StreamRightAfterThePushIsAttributed(t *testing.T) { + srv := newStubServer(t) + srv.registerResp.ServerConfig = launchCfgWithShortID() + b := &fakeBoxService{} + live := helper.NewLiveTracker(nil) + c := newTestClient(t, &fakeForwarder{}, b, srv, func(cfg *Config) { + cfg.HeartbeatInterval = time.Hour + cfg.Live = live + }) + require.NoError(t, c.Start(context.Background())) + t.Cleanup(func() { _ = c.Stop(context.Background()) }) + + var once sync.Once + b.afterUpdate = func() { + once.Do(func() { + friend := counter(0, 0) + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: friend, ShortID: sid(t, sidB1)}) + friend.AddDown(helper.LiveThreshold) + peerconn.Notify(peerconn.Event{State: -1, Source: "198.51.100.1:1000", Bytes: friend, ShortID: sid(t, sidB1)}) + }) + } + c.offerPairingShortIDs([]string{sidB1}, []PairingShortIDSet{{PairingID: pairingB, ShortIDs: []string{sidB1}}}, true, 100) + s := live.Snapshot() + require.Len(t, s.Pairings, 1) + assert.Equal(t, pairingB, s.Pairings[0].PairingID) + assert.True(t, s.Pairings[0].Connected) +} + +// A friend who connects while the server verifies the route, before Start +// returns, is reported and tracked. +func TestClient_ConnectionDuringVerifyIsTracked(t *testing.T) { + srv := newStubServer(t) + srv.registerResp.ServerConfig = launchCfgWithShortID() + srv.registerResp.PairingShortIDs = []string{sidA1} + srv.registerResp.PairingShortIDsKnown = true + srv.registerResp.PairingShortIDSets = []PairingShortIDSet{{PairingID: pairingA, ShortIDs: []string{sidA1}}} + srv.onVerify = func(LifecycleRequest) int { + peerconn.Notify(peerconn.Event{State: +1, Source: "198.51.100.1:1000", Destination: "example.com:443", Bytes: counter(0, 0), ShortID: sid(t, sidA1)}) + return 0 + } + live := helper.NewLiveTracker(nil) + c := newTestClient(t, &fakeForwarder{}, &fakeBoxService{}, srv, func(cfg *Config) { + cfg.HeartbeatInterval = time.Hour + cfg.Live = live + }) + require.NoError(t, c.Start(context.Background())) + t.Cleanup(func() { _ = c.Stop(context.Background()) }) + s := live.Snapshot() + require.Len(t, s.Pairings, 1) + assert.Equal(t, pairingA, s.Pairings[0].PairingID) +} + +// A failed push publishes none of its groups. +func TestClient_FailedPushUnstagesItsGroups(t *testing.T) { + srv := newStubServer(t) + srv.registerResp.ServerConfig = launchCfgWithShortID() + b := &fakeBoxService{} + c := newTestClient(t, &fakeForwarder{}, b, srv, func(cfg *Config) { cfg.HeartbeatInterval = time.Hour }) + require.NoError(t, c.Start(context.Background())) + t.Cleanup(func() { _ = c.Stop(context.Background()) }) + b.setUpdateErr(errors.New("refused")) + c.offerPairingShortIDs([]string{sidB1}, []PairingShortIDSet{{PairingID: pairingB, ShortIDs: []string{sidB1}}}, true, 100) + assert.Equal(t, "", c.pairingFor(sid(t, sidB1))) +} diff --git a/peer/peer.go b/peer/peer.go index 0645e622..31ccd270 100644 --- a/peer/peer.go +++ b/peer/peer.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log/slog" + "maps" "math/rand/v2" "net" "runtime/debug" @@ -30,6 +31,7 @@ import ( "github.com/getlantern/radiance/common/env" "github.com/getlantern/radiance/common/settings" "github.com/getlantern/radiance/events" + "github.com/getlantern/radiance/helper" "github.com/getlantern/radiance/portforward" ) @@ -48,11 +50,16 @@ type StatusEvent struct { // State +1 on accept, -1 on close // Source remote peer "ip:port" // Timestamp emit time in Unix milliseconds +// Pairing the helper pairing the client came through, "" for the pool or not yet known +// +// Pairing identifies a friend relationship. It goes to the local app only: +// it must never be logged, traced or persisted. type ConnectionEvent struct { events.Event State int `json:"state"` Source string `json:"source"` Timestamp int64 `json:"timestamp"` + Pairing string `json:"pairing,omitempty"` } // Port range chosen to minimize collision risk on the typical home network, @@ -143,8 +150,14 @@ type Config struct { // left nothing serving what the server advertises. It runs on its own // goroutine after Stop returns, with the reason. OnSelfStop func(reason error) + // Live receives the connections that came through a helper pairing, for + // the volunteer's live per-friend status. Nil means helper.DefaultLive. + Live *helper.LiveTracker } +// liveOwner groups this package's connections in the LiveTracker. +const liveOwner = "peer" + // defaultBoxCloseTimeout: samizdat's Close releases the listening port at once // but then waits for every open client connection to end, which with // long-lived clients has taken 18 minutes. Nothing teardown does next depends @@ -252,6 +265,32 @@ type Client struct { // pairingConfirmedAt is when a known set was last accepted. A rotation // that reuses the record keeps it. pairingConfirmedAt time.Time + // pairingByShortID maps a short ID to its pairing ID for every grouped + // set a serving box may accept: heldGroups, prevGroups, and + // pendingGroups while a push of them is in flight. Written under c.mu, + // read lock-free by the connection listener. + pairingByShortID atomic.Pointer[map[[8]byte]string] + // heldGroups are the groups of the newest grouped set pushed to the + // inbound, and prevGroups those it replaced, which a box built during a + // rotation may still hold. pendingGroups are the recorded set's groups + // until a push of that set succeeds. Guarded by c.mu. + heldGroups map[[8]byte]string + prevGroups map[[8]byte]string + pendingGroups *map[[8]byte]string + // liveMu orders the listener's LiveTracker updates against a session's + // end. liveGen identifies the running session and changes at every + // begin and end, so a listener callback that resumes after its session + // ended adds nothing that the end would have had to close. + liveMu sync.Mutex + liveGen uint64 + // conns holds this session's open connections by byte counter, so a + // close reports the pairing the connection was attributed to, at accept + // or by a grouped set that arrived later. Cleared at the session's begin + // and end. Guarded by liveMu, which is taken after c.mu. + conns map[*peerconn.Counter]liveConn + // beforeTrackLive, if set, runs in the listener just before its + // LiveTracker update, so tests can interleave it with Stop. + beforeTrackLive func() // shortIDsMu serializes recording a set with pushing it onto the box, // so the box and pairingApplied agree. It is held across those local // pushes only, never across network I/O or a box's Start or Close. @@ -379,6 +418,7 @@ func (c *Client) Start(ctx context.Context) (retErr error) { // here, a Stop→Start cycle would leave the wrapper permanently muted. c.listenerDraining.Store(false) relayGen := c.relay.begin() + liveGen := c.beginLive() var ( success bool @@ -403,15 +443,18 @@ func (c *Client) Start(ctx context.Context) (retErr error) { // registered route + router rule. cleanupCtx, cancel := context.WithTimeout(context.Background(), peerCleanupTimeout) defer cancel() - // Always clear the connection listener on rollback. The listener is - // only Set on the success path, so this is a no-op if Start failed - // before reaching it — but cheap insurance against a future re-order - // that registers earlier. Drain-flag first so any in-flight Notify - // callbacks short-circuit even if SetListener races (see Stop). + // Always clear the connection listener on rollback: it is registered + // before box.Start and Verify, either of which can fail. Drain-flag + // first so any in-flight Notify callbacks short-circuit even if + // SetListener races (see Stop). c.listenerDraining.Store(true) peerconn.SetListener(nil) c.resetConnTracking() c.relay.end() + c.endLive() + c.mu.Lock() + c.clearGroupsLocked() + c.mu.Unlock() if box != nil { _ = c.closeBox(box) } @@ -542,24 +585,23 @@ func (c *Client) Start(ctx context.Context) (retErr error) { startRevision = int64(regResp.PairingShortIDsRevision) startPairingApplied = c.applyShortIDs(box, baseShortIDs, startPairing) } - - if err := box.Start(); err != nil { - cancelRun() - return fmt.Errorf("start sing-box: %w", err) + var startGroups *map[[8]byte]string + if startPairingKnown { + if g, ok := groupPairingShortIDs(startPairing, regResp.PairingShortIDSets, baseShortIDs); ok { + startGroups = &g + } } - c.emitPhase(PhaseVerifying, "") - // Now that sing-box is listening with the just-built creds, ask the - // server to dial back through them. Splitting verify out of Register - // into this explicit follow-up avoids the chicken-and-egg where the - // server tried to verify before the peer could possibly be listening - // (the cert/key only arrive in the Register response). Failure here - // is fatal — the server has already deprecated the row, so the - // deferred cleanup tears the rest of the session down. - if err := c.cfg.API.Verify(ctx, regResp.RouteID); err != nil { - cancelRun() - return fmt.Errorf("verify with lantern-cloud: %w", err) + // Published before serving, so connections accepted from the start + // resolve their pairing. + c.mu.Lock() + c.heldGroups, c.prevGroups, c.pendingGroups = nil, nil, startGroups + if startPairingApplied { + c.commitPendingGroupsLocked() + } else { + c.publishGroupsLocked(false) } + c.mu.Unlock() // Forward inbound accept/close events from lantern-box's samizdat // inbound to the radiance event bus. Consumers (lantern-core's @@ -567,8 +609,10 @@ func (c *Client) Start(ctx context.Context) (retErr error) { // events.Subscribe[ConnectionEvent]. Listener is process-wide // single-active; cleared on Stop and in the rollback defer so // post-teardown accept-loop callbacks land on a no-op rather than - // emit events to a torn-down consumer. Must run AFTER box.Start so - // the accept loop is serving when notifications start flowing. + // emit events to a torn-down consumer. + // Registered before box.Start: samizdat binds each connection to the + // listener at accept, so registering any later would never report a + // connection accepted in between, such as one during Verify. peerconn.SetListener(func(evt peerconn.Event) { state, source := evt.State, evt.Source if c.listenerDraining.Load() { @@ -596,14 +640,36 @@ func (c *Client) Start(ctx context.Context) (retErr error) { "state", state, "source", source) return } + // The pairing is deliberately absent from this line and every other. slog.Debug("peer listener: forwarding connection event", "state", state, "source", source) c.trackConn(state, source) - c.relay.recordFor(relayGen, state, source, evt.Bytes) - events.Emit(ConnectionEvent{State: state, Source: source, Timestamp: time.Now().UnixMilli()}) + if c.beforeTrackLive != nil { + c.beforeTrackLive() + } + pairing := c.trackPairing(liveGen, relayGen, state, source, evt.ShortID, evt.Bytes) + events.Emit(ConnectionEvent{State: state, Source: source, Timestamp: time.Now().UnixMilli(), Pairing: pairing}) }) slog.Info("peer listener: registered with peerconn", "route_id", regResp.RouteID) + if err := box.Start(); err != nil { + cancelRun() + return fmt.Errorf("start sing-box: %w", err) + } + + c.emitPhase(PhaseVerifying, "") + // Now that sing-box is listening with the just-built creds, ask the + // server to dial back through them. Splitting verify out of Register + // into this explicit follow-up avoids the chicken-and-egg where the + // server tried to verify before the peer could possibly be listening + // (the cert/key only arrive in the Register response). Failure here + // is fatal — the server has already deprecated the row, so the + // deferred cleanup tears the rest of the session down. + if err := c.cfg.API.Verify(ctx, regResp.RouteID); err != nil { + cancelRun() + return fmt.Errorf("verify with lantern-cloud: %w", err) + } + // HeartbeatIntervalSeconds is server-driven so lantern-cloud can dial up // the cadence on registrations it wants to expire faster. Honor any // positive value verbatim — clamping short intervals up would defeat @@ -770,6 +836,7 @@ func (c *Client) stopHeld(ctx context.Context, target stopTarget) (bool, error) c.pairingApplied = false c.pairingRevision = 0 c.pairingConfirmedAt = time.Time{} + c.clearGroupsLocked() c.runCtx = nil c.status = Status{Phase: PhaseStopping} stoppingSnapshot := c.status @@ -788,6 +855,8 @@ func (c *Client) stopHeld(ctx context.Context, target stopTarget) (bool, error) peerconn.SetListener(nil) c.resetConnTracking() c.relay.end() + // Stop drops the closes of connections still open, so end them here. + c.endLive() cancel() <-done @@ -1014,7 +1083,7 @@ func (c *Client) heartbeatLoop(ctx context.Context, interval time.Duration, done // Replaces the pairing set, so an ended pairing stops working // here, unless a newer set was accepted while it was in // flight. - c.applyPairingShortIDs(resp.PairingShortIDs, true, int64(resp.PairingShortIDsRevision)) + c.offerPairingShortIDs(resp.PairingShortIDs, resp.PairingShortIDSets, true, int64(resp.PairingShortIDsRevision)) } else { // A failed heartbeat, or one that does not vouch for a set, // says nothing about pairings, so the last set stays, and is @@ -1211,7 +1280,7 @@ func (c *Client) rotateCreds(ctx context.Context) error { if regResp.PairingShortIDsKnown { // Offered like a heartbeat's set: taken only if newer than the // one held, and pushed onto the box serving now. - c.applyPairingShortIDs(regResp.PairingShortIDs, true, int64(regResp.PairingShortIDsRevision)) + c.offerPairingShortIDs(regResp.PairingShortIDs, regResp.PairingShortIDSets, true, int64(regResp.PairingShortIDsRevision)) } sameRoute := regResp.RouteID == oldRouteID // abandon undoes a registration this rotation can no longer use. A new @@ -1296,6 +1365,9 @@ func (c *Client) rotateCreds(ctx context.Context) error { } c.mu.Lock() rotPairing, rotKnown = slices.Clone(c.pairingShortIDs), c.pairingKnown + if rotKnown { + c.publishGroupsLocked(true) + } c.mu.Unlock() rotApplied = rotKnown && c.applyShortIDs(b, newBase, rotPairing) return b, nil @@ -1349,13 +1421,24 @@ func (c *Client) rotateCreds(ctx context.Context) error { } held := slices.Equal(pushed, want) c.pairingApplied = held + if held { + c.commitPendingGroupsLocked() + } else { + c.publishGroupsLocked(true) + } c.status.RouteID = regResp.RouteID c.status.ExternalIP = externalIP c.mu.Unlock() - if !held && c.applyShortIDs(newBox, newBase, want) { + if !held { + ok := c.applyShortIDs(newBox, newBase, want) c.mu.Lock() if c.box == newBox { - c.pairingApplied = true + if ok { + c.pairingApplied = true + c.commitPendingGroupsLocked() + } else { + c.publishGroupsLocked(false) + } } c.mu.Unlock() } @@ -1670,6 +1753,15 @@ func (c *Client) acceptsPairingRevision(rev int64) bool { // none is known) because a push failed, it is pushed again; with known // false that is all it does. func (c *Client) applyPairingShortIDs(pairing []string, known bool, rev int64) { + c.offerPairingShortIDs(pairing, nil, known, rev) +} + +// offerPairingShortIDs is applyPairingShortIDs with the set's per-pairing +// groups, taken only when their IDs, joined in order, equal the set held at +// that revision. Groups are also taken at the revision already held, since a +// replica that predates groups answers that revision without them. A newer +// set without matching groups keeps the previous mapping. +func (c *Client) offerPairingShortIDs(pairing []string, groups []PairingShortIDSet, known bool, rev int64) { c.shortIDsMu.Lock() defer c.shortIDsMu.Unlock() c.mu.Lock() @@ -1684,6 +1776,17 @@ func (c *Client) applyPairingShortIDs(pairing []string, known bool, rev int64) { c.pairingRevision = rev c.pairingConfirmedAt = c.cfg.Now() c.pairingApplied = false + c.pendingGroups = nil + if g, ok := groupPairingShortIDs(pairing, groups, base); ok { + c.pendingGroups = &g + } + } else if known && c.pairingKnown && rev == c.pairingRevision { + if g, ok := groupPairingShortIDs(c.pairingShortIDs, groups, base); ok { + c.pendingGroups = &g + if c.pairingApplied { + c.commitPendingGroupsLocked() + } + } } if c.pairingApplied { c.mu.Unlock() @@ -1695,17 +1798,59 @@ func (c *Client) applyPairingShortIDs(pairing []string, known bool, rev int64) { if c.pairingKnown { record = slices.Clone(c.pairingShortIDs) } + // Staged before the push, so a stream the inbound accepts on a new ID + // resolves its pairing at once. + c.publishGroupsLocked(true) c.mu.Unlock() - if !c.applyShortIDs(b, base, record) { - return - } + ok := c.applyShortIDs(b, base, record) c.mu.Lock() if c.box == b { - c.pairingApplied = true + if ok { + c.pairingApplied = true + c.commitPendingGroupsLocked() + } else { + c.publishGroupsLocked(false) + } } c.mu.Unlock() } +// commitPendingGroupsLocked makes pendingGroups the held groups, keeping +// the groups they replace, and publishes the map. Pending groups equal to the +// held ones replace nothing, so a repeated response keeps prevGroups. With +// none pending it only publishes. The box must already hold the recorded +// set, and c.mu must be held. +func (c *Client) commitPendingGroupsLocked() { + if c.pendingGroups != nil { + if !maps.Equal(*c.pendingGroups, c.heldGroups) { + c.prevGroups, c.heldGroups = c.heldGroups, *c.pendingGroups + } + c.pendingGroups = nil + } + c.publishGroupsLocked(false) +} + +// publishGroupsLocked sets the pairing map to the held and previous groups, +// and the pending ones too when staged is true. Called with c.mu held. +func (c *Client) publishGroupsLocked(staged bool) { + sets := []map[[8]byte]string{c.prevGroups, c.heldGroups} + if staged && c.pendingGroups != nil { + sets = append(sets, *c.pendingGroups) + } + m := make(map[[8]byte]string) + for _, set := range sets { + maps.Copy(m, set) + } + c.setPairingByShortID(m) +} + +// clearGroupsLocked forgets every group and empties the pairing map. +// Called with c.mu held. +func (c *Client) clearGroupsLocked() { + c.heldGroups, c.prevGroups, c.pendingGroups = nil, nil, nil + c.setPairingByShortID(nil) +} + // pairingMaxUnconfirmed is how long a pairing set is kept without the server // vouching for it again. A pairing short ID is valid for epochs e-1..e+1 of // 24 hours, so none in a set is still valid this long after it was issued. @@ -1741,11 +1886,157 @@ func (c *Client) expireUnconfirmedPairing() { c.pairingKnown = false c.pairingApplied = true c.pairingConfirmedAt = time.Time{} + c.clearGroupsLocked() } c.mu.Unlock() slog.Info("peer: pairing short IDs unconfirmed for too long; dropped") } +// groupPairingShortIDs maps each short ID in groups to its pairing, and +// reports whether groups describe flat: their IDs, joined in order, equal +// flat's, ignoring case. A short ID in base is the launch config's own, which +// pool clients and the verifier use, so it never maps to a pairing. +func groupPairingShortIDs(flat []string, groups []PairingShortIDSet, base []string) (map[[8]byte]string, bool) { + if len(groups) == 0 { + return nil, false + } + var joined []string + for _, g := range groups { + for _, id := range g.ShortIDs { + joined = append(joined, strings.ToLower(id)) + } + } + if len(joined) != len(flat) { + return nil, false + } + for i, id := range flat { + if strings.ToLower(id) != joined[i] { + return nil, false + } + } + isBase := make(map[string]bool, len(base)) + for _, id := range base { + isBase[strings.ToLower(id)] = true + } + out := make(map[[8]byte]string, len(joined)) + for _, g := range groups { + if g.PairingID == "" { + continue + } + for _, id := range g.ShortIDs { + id = strings.ToLower(id) + raw, err := hex.DecodeString(id) + if err != nil || len(raw) != 8 || isBase[id] { + continue + } + out[[8]byte(raw)] = g.PairingID + } + } + return out, true +} + +type liveConn struct { + shortID [8]byte + pairing string +} + +// setPairingByShortID replaces the short ID to pairing map; nil clears it. +// Open connections the new map attributes are then followed in the +// LiveTracker and labelled in the relay snapshot. +func (c *Client) setPairingByShortID(m map[[8]byte]string) { + if len(m) == 0 { + c.pairingByShortID.Store(nil) + return + } + c.pairingByShortID.Store(&m) + // After the store, so a connection the listener records without a + // pairing either sees the new map itself or is here to be resolved. + c.liveMu.Lock() + defer c.liveMu.Unlock() + for bytes, lc := range c.conns { + if lc.pairing != "" { + continue + } + pairing := m[lc.shortID] + if pairing == "" { + continue + } + c.conns[bytes] = liveConn{shortID: lc.shortID, pairing: pairing} + c.relay.setPairing(bytes, pairing) + c.live().Open(liveOwner, bytes, pairing, func() uint64 { return bytes.Down() + bytes.Up() }) + } +} + +// pairingFor is the pairing a connection with short ID id came through, or "" +// if none. +func (c *Client) pairingFor(id [8]byte) string { + if id == ([8]byte{}) { + return "" + } + m := c.pairingByShortID.Load() + if m == nil { + return "" + } + return (*m)[id] +} + +func (c *Client) live() *helper.LiveTracker { + if c.cfg.Live != nil { + return c.cfg.Live + } + return helper.DefaultLive +} + +// beginLive starts a session's live tracking and returns its generation. +func (c *Client) beginLive() uint64 { + c.liveMu.Lock() + defer c.liveMu.Unlock() + c.liveGen++ + clear(c.conns) + return c.liveGen +} + +// endLive ends the session's live tracking: every connection it opened is +// closed, and its late callbacks are ignored. +func (c *Client) endLive() { + c.liveMu.Lock() + defer c.liveMu.Unlock() + c.liveGen++ + clear(c.conns) + c.live().CloseOwner(liveOwner) +} + +// trackPairing records a connection event in the relay snapshot and the +// LiveTracker and returns its pairing, "" for the pool or not yet known. +// Events from an ended session are not live-tracked. +func (c *Client) trackPairing(liveGen, relayGen uint64, state int, source string, shortID [8]byte, bytes *peerconn.Counter) string { + c.liveMu.Lock() + defer c.liveMu.Unlock() + pairing := c.pairingFor(shortID) + lc, open := c.conns[bytes] + if state < 0 && open { + pairing = lc.pairing + } + c.relay.recordPairingFor(relayGen, state, source, pairing, bytes) + if bytes == nil || liveGen != c.liveGen { + return pairing + } + switch { + case state > 0: + if c.conns == nil { + c.conns = make(map[*peerconn.Counter]liveConn) + } + c.conns[bytes] = liveConn{shortID: shortID, pairing: pairing} + if pairing != "" { + c.live().Open(liveOwner, bytes, pairing, func() uint64 { return bytes.Down() + bytes.Up() }) + } + case state < 0: + delete(c.conns, bytes) + c.live().Close(bytes) + } + return pairing +} + // defaultBuildBoxService builds the sing-box service for one peer box. // // A peer-proxy box needs no TUN / platform-VPN integration the way the main diff --git a/peer/peer_test.go b/peer/peer_test.go index 9e66f7db..e820c340 100644 --- a/peer/peer_test.go +++ b/peer/peer_test.go @@ -136,6 +136,9 @@ type fakeBoxService struct { // idsAtStart is the set the inbound held when Start was called, nil for // its base ID only. idsAtStart []string + // afterUpdate, if set, runs after a set is applied, as connections on + // its IDs could arrive then. Set before the box is in use. + afterUpdate func() } func (b *fakeBoxService) Start() error { @@ -172,6 +175,11 @@ func (b *fakeBoxService) UpdateShortIDs(ids []string) error { return b.updateErr } b.shortIDs = append(b.shortIDs, append([]string(nil), ids...)) + if b.afterUpdate != nil { + b.idsMu.Unlock() + b.afterUpdate() + b.idsMu.Lock() + } return nil } @@ -238,6 +246,10 @@ type stubServer struct { deregisterDeviceID atomic.Value // string lastRegisterReq atomic.Value // RegisterRequest lastVerifyReq atomic.Value // LifecycleRequest + + // heartbeatShortIDSets, when set, is sent as pairing_short_id_sets with + // heartbeatShortIDs. + heartbeatShortIDSets atomic.Pointer[[]PairingShortIDSet] } func newStubServer(t *testing.T) *stubServer { @@ -313,7 +325,11 @@ func newStubServer(t *testing.T) *stubServer { return } if ids != nil { - _ = json.NewEncoder(w).Encode(HeartbeatResponse{PairingShortIDs: *ids, PairingShortIDsKnown: true, PairingShortIDsRevision: Revision(rev)}) + resp := HeartbeatResponse{PairingShortIDs: *ids, PairingShortIDsKnown: true, PairingShortIDsRevision: Revision(rev)} + if sets := s.heartbeatShortIDSets.Load(); sets != nil { + resp.PairingShortIDSets = *sets + } + _ = json.NewEncoder(w).Encode(resp) return } w.WriteHeader(http.StatusOK) diff --git a/peer/snapshot.go b/peer/snapshot.go index d7fe3b90..5fa84657 100644 --- a/peer/snapshot.go +++ b/peer/snapshot.go @@ -34,6 +34,10 @@ type PeerBytes struct { IP string `json:"ip"` Down uint64 `json:"down"` Up uint64 `json:"up"` + // Pairing is the helper pairing of this client's latest connection, "" + // for the pool or not yet attributed. It identifies a friend + // relationship and must never be logged. + Pairing string `json:"pairing,omitempty"` } // relayTally is one client IP's bytes: closed holds its finished @@ -42,6 +46,11 @@ type relayTally struct { closedDown, closedUp uint64 open int endedAt time.Time + // pairing is the pairing of the client's latest connection, and + // latest that connection's counter, so a connection attributed late + // relabels the client only if it is still the latest. + pairing string + latest *peerconn.Counter } // relayConn is an open connection's source and, once a later session has @@ -186,6 +195,12 @@ func (r *relayStats) record(state int, source string, bytes *peerconn.Counter) { // recordFor is record for the session with generation gen; an event from any // other session is ignored. func (r *relayStats) recordFor(gen uint64, state int, source string, bytes *peerconn.Counter) { + r.recordPairingFor(gen, state, source, "", bytes) +} + +// recordPairingFor is recordFor for a connection that came through pairing, +// "" for the pool. +func (r *relayStats) recordPairingFor(gen uint64, state int, source, pairing string, bytes *peerconn.Counter) { if bytes == nil { return } @@ -219,6 +234,8 @@ func (r *relayStats) recordFor(gen uint64, state int, source string, bytes *peer r.ips[ip] = t } t.open++ + t.pairing = pairing + t.latest = bytes case state < 0: rc, ok := r.conns[bytes] if !ok || !rc.retiredAt.IsZero() { @@ -232,6 +249,20 @@ func (r *relayStats) recordFor(gen uint64, state int, source string, bytes *peer } } +// setPairing labels an open connection's client with pairing, for a +// connection attributed after it was accepted. +func (r *relayStats) setPairing(bytes *peerconn.Counter, pairing string) { + r.mu.Lock() + defer r.mu.Unlock() + rc, ok := r.conns[bytes] + if !ok || !rc.retiredAt.IsZero() { + return + } + if t := r.ips[ipOf(rc.source)]; t != nil && t.latest == bytes { + t.pairing = pairing + } +} + func (r *relayStats) snapshot(running bool) Snapshot { r.mu.Lock() defer r.mu.Unlock() @@ -260,7 +291,7 @@ func (r *relayStats) snapshot(running bool) Snapshot { } for ip, t := range r.ips { pb := live[ip] - s.PeerBytes = append(s.PeerBytes, PeerBytes{IP: ip, Down: t.closedDown + pb.Down, Up: t.closedUp + pb.Up}) + s.PeerBytes = append(s.PeerBytes, PeerBytes{IP: ip, Down: t.closedDown + pb.Down, Up: t.closedUp + pb.Up, Pairing: t.pairing}) } sort.Strings(s.Peers) sort.Slice(s.PeerBytes, func(i, j int) bool { return s.PeerBytes[i].IP < s.PeerBytes[j].IP }) diff --git a/unbounded/pairing_test.go b/unbounded/pairing_test.go new file mode 100644 index 00000000..f5d8bdc3 --- /dev/null +++ b/unbounded/pairing_test.go @@ -0,0 +1,119 @@ +package unbounded + +import ( + "context" + "net" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/getlantern/broflake/clientcore" + "github.com/getlantern/broflake/common" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/getlantern/radiance/helper" + "github.com/getlantern/radiance/internal/testutil" +) + +const testPairing = "b2c3d4e5-0000-4000-8000-00000000000c" + +// startCapturing starts the manager with a fake widget and returns the +// options start installed on it. +func startCapturing(t *testing.T, live *helper.LiveTracker) *clientcore.BroflakeOptions { + t.Helper() + var captured atomic.Pointer[clientcore.BroflakeOptions] + resetManager(t, func(bfOpt *clientcore.BroflakeOptions, _ *clientcore.WebRTCOptions, _ *clientcore.EgressOptions) (widget, error) { + captured.Store(bfOpt) + return &fakeWidget{}, nil + }) + manager.live = live + primeManager(t, nil) + manager.start() + require.Eventually(t, func() bool { return captured.Load() != nil }, time.Second, 5*time.Millisecond) + require.Eventually(t, func() bool { return CurrentSnapshot().Running }, time.Second, time.Millisecond) + return captured.Load() +} + +// connect delivers a connection change the way broflake does: the plain +// callback, then the info one. +func connect(bf *clientcore.BroflakeOptions, state, idx int, ip, pairing string) { + var addr net.IP + info := common.ConsumerInfo{} + if state > 0 { + addr = net.ParseIP(ip) + info = common.ConsumerInfo{Addr: addr, Pairing: pairing} + } + bf.OnConnectionChangeFunc(state, idx, addr) + bf.OnConnectionChangeInfoFunc(state, idx, info) +} + +func TestInfoCallbackCarriesPairingToTheTracker(t *testing.T) { + live := helper.NewLiveTracker(nil) + bf := startCapturing(t, live) + require.NotNil(t, bf.OnConnectionChangeInfoFunc, "start installs the info callback") + + connect(bf, 1, 0, "198.51.100.7", testPairing) + connect(bf, 1, 1, "198.51.100.8", "") + s := live.Snapshot() + require.Len(t, s.Pairings, 1, "the pool slot is not tracked") + assert.Equal(t, testPairing, s.Pairings[0].PairingID) + assert.False(t, s.Pairings[0].Connected, "an idle slot is not connected") + + // Bytes on the pool slot do not count toward the friend. + bf.OnDownstreamChunkFunc(helper.LiveThreshold, 1) + assert.False(t, live.Snapshot().Pairings[0].Connected) + + bf.OnDownstreamChunkFunc(helper.LiveThreshold/2, 0) + bf.OnUpstreamChunkFunc(helper.LiveThreshold/2, 0) + assert.True(t, live.Snapshot().Pairings[0].Connected) + + got := peerBytesByIP(CurrentSnapshot()) + assert.Equal(t, testPairing, got["198.51.100.7"].Pairing, "the snapshot labels the friend's IP") + assert.Equal(t, "", got["198.51.100.8"].Pairing) + + // The slot closes; the friend's bytes still count for the window. + connect(bf, -1, 0, "", "") + assert.True(t, live.Snapshot().Pairings[0].Connected) + + // A new pool occupant of the same slot is not the friend. + connect(bf, 1, 0, "198.51.100.9", "") + bf.OnDownstreamChunkFunc(10*helper.LiveThreshold, 0) + live.Snapshot() + require.NoError(t, Stop(context.Background())) +} + +func TestBytesBeforeTheFriendAreNotTheFriends(t *testing.T) { + live := helper.NewLiveTracker(nil) + bf := startCapturing(t, live) + connect(bf, 1, 0, "198.51.100.9", "") + bf.OnDownstreamChunkFunc(10*helper.LiveThreshold, 0) + connect(bf, -1, 0, "", "") + connect(bf, 1, 0, "198.51.100.7", testPairing) + s := live.Snapshot() + require.Len(t, s.Pairings, 1) + assert.False(t, s.Pairings[0].Connected, "the slot's earlier pool traffic is not the friend's") + require.NoError(t, Stop(context.Background())) +} + +func TestStopEndsTrackedSlots(t *testing.T) { + live := helper.NewLiveTracker(nil) + bf := startCapturing(t, live) + connect(bf, 1, 0, "198.51.100.7", testPairing) + require.Len(t, live.Snapshot().Pairings, 1) + require.NoError(t, Stop(context.Background())) + assert.Empty(t, live.Snapshot().Pairings) + + // A callback broflake delivers after the stop is ignored. + bf.OnConnectionChangeInfoFunc(1, 2, common.ConsumerInfo{Pairing: testPairing}) + assert.Empty(t, live.Snapshot().Pairings) +} + +// The pairing never reaches the logs: no slog call in the package is passed +// a pairing or the ConsumerInfo that carries one. +func TestPairingIsNotLogged(t *testing.T) { + testutil.LoggedIdentifiers(t, ".", func(name string) bool { + return strings.Contains(strings.ToLower(name), "pairing") || name == "info" + }) +} diff --git a/unbounded/snapshot.go b/unbounded/snapshot.go index a9964955..446775cd 100644 --- a/unbounded/snapshot.go +++ b/unbounded/snapshot.go @@ -9,6 +9,8 @@ import ( "time" "github.com/getlantern/broflake/clientcore" + + "github.com/getlantern/radiance/helper" ) var snapshotEpoch = strconv.FormatInt(time.Now().UnixNano(), 10) @@ -37,6 +39,10 @@ type PeerBytes struct { IP string `json:"ip"` Down uint64 `json:"down"` Up uint64 `json:"up"` + // Pairing is the helper pairing this consumer most recently came + // through, "" for the pool or one broflake could not attribute. It + // identifies a friend relationship and must never be logged. + Pairing string `json:"pairing,omitempty"` } // slotBytes holds bytes relayed through one broflake worker slot since they @@ -44,6 +50,9 @@ type PeerBytes struct { // atomics, so the per-chunk data path never contends on m.mu. type slotBytes struct { down, up atomic.Uint64 + // total is every byte the slot has relayed, never reset, which the + // LiveTracker samples for a slot serving a helper pairing. + total atomic.Uint64 } // slotTable is created per worker run so chunks a stopped worker delivers late @@ -63,6 +72,7 @@ func (t *slotTable) add(idx int, size uint64, down bool) { return } s := t.slot(idx) + s.total.Add(size) if down { s.down.Add(size) } else { @@ -91,6 +101,8 @@ type peerTally struct { // active counts slots currently held by this IP; endedAt is set when it drops to zero. active int endedAt time.Time + // pairing is the helper pairing of the IP's latest slot, "" for the pool. + pairing string } // chunkCounter returns a broflake ChunkFunc that adds relayed bytes to the @@ -134,7 +146,7 @@ func (m *unboundedManager) snapshot() Snapshot { } sort.Strings(s.Peers) for ip, t := range m.peerBytes { - s.PeerBytes = append(s.PeerBytes, PeerBytes{IP: ip, Down: t.down, Up: t.up}) + s.PeerBytes = append(s.PeerBytes, PeerBytes{IP: ip, Down: t.down, Up: t.up, Pairing: t.pairing}) } sort.Slice(s.PeerBytes, func(i, j int) bool { return s.PeerBytes[i].IP < s.PeerBytes[j].IP }) return s @@ -189,6 +201,50 @@ func (m *unboundedManager) recordConnection(ctx context.Context, state, slot int } } +// liveOwner groups this package's slots in the LiveTracker. +const liveOwner = "unbounded" + +// liveSlot identifies one slot of one run in the LiveTracker. +type liveSlot struct { + slots *slotTable + idx int +} + +// recordPairing follows broflake's pairing for a slot's connection change, +// which arrives just after recordConnection's. +func (m *unboundedManager) recordPairing(ctx context.Context, slots *slotTable, state, slot int, pairing string) { + m.mu.Lock() + defer m.mu.Unlock() + if ctx.Err() != nil || slots == nil { + return + } + key := liveSlot{slots: slots, idx: slot} + if state < 0 { + m.liveTracker().Close(key) + return + } + if state == 0 { + return + } + if source := m.peers[slot]; source != "" { + if t := m.peerBytes[source]; t != nil { + t.pairing = pairing + } + } + if pairing == "" { + m.liveTracker().Close(key) + return + } + m.liveTracker().Open(liveOwner, key, pairing, slots.slot(slot).total.Load) +} + +func (m *unboundedManager) liveTracker() *helper.LiveTracker { + if m.live != nil { + return m.live + } + return helper.DefaultLive +} + // retiredRun keeps a stopped run's slot table and its final slot owners, so // bytes broflake delivers during its asynchronous teardown are still counted // and credited to the consumers that relayed them. @@ -203,6 +259,7 @@ type retiredRun struct { // table for the same period. Caller must hold m.mu. func (m *unboundedManager) releaseAllPeers() { m.prunePeerBytes() + m.liveTracker().CloseOwner(liveOwner) for _, source := range m.peers { m.releasePeer(source) } diff --git a/unbounded/unbounded.go b/unbounded/unbounded.go index 6d619e0a..275e60d1 100644 --- a/unbounded/unbounded.go +++ b/unbounded/unbounded.go @@ -38,11 +38,13 @@ import ( C "github.com/getlantern/common" "github.com/getlantern/broflake/clientcore" + "github.com/getlantern/broflake/common" "github.com/getlantern/broflake/common/helperticket" "github.com/getlantern/radiance/common/settings" "github.com/getlantern/radiance/config" "github.com/getlantern/radiance/events" + "github.com/getlantern/radiance/helper" ) // ConnectionEvent fires every time a consumer (i.e. a censored client @@ -206,6 +208,9 @@ type unboundedManager struct { retired []retiredRun // now overrides time.Now for peer retention tests. now func() time.Time + // live receives slots serving helper pairings. Nil means + // helper.DefaultLive. + live *helper.LiveTracker } // shouldStart reports whether all three start conditions hold. Caller @@ -782,6 +787,13 @@ func (m *unboundedManager) startLocked() { }) } + // broflake calls this just after OnConnectionChangeFunc, with the + // helper pairing the consumer came through, which the producer has + // checked against its own tickets. The pairing is never logged. + bfOpt.OnConnectionChangeInfoFunc = func(state int, workerIdx int, info common.ConsumerInfo) { + m.recordPairing(ctx, slots, state, workerIdx, info.Pairing) + } + // Downstream chunks fire per chunk on broflake's data path, so the // counters stay lock-free and attribution to an IP happens later // under m.mu. Chunks arriving after stop land in the retired run,