diff --git a/backend/peer_share.go b/backend/peer_share.go index 22a32b14..acddffa7 100644 --- a/backend/peer_share.go +++ b/backend/peer_share.go @@ -22,6 +22,7 @@ type peerController interface { Stop(ctx context.Context) error IsActive() bool CurrentStatus() peer.Status + RelaySnapshot() peer.Snapshot } // peerToggleTimeout caps blocking time on a slow router; UPnP M-SEARCH + @@ -197,3 +198,13 @@ func (r *LocalBackend) PeerStatus() peer.Status { } return r.peerClient.CurrentStatus() } + +// PeerRelaySnapshot returns the bytes the current or most recent direct-sharing +// session has relayed, for the IPC /peer/snapshot endpoint. A nil peerClient +// reports an empty snapshot, as PeerStatus does. +func (r *LocalBackend) PeerRelaySnapshot() peer.Snapshot { + if r.peerClient == nil { + return peer.Snapshot{Peers: []string{}, PeerBytes: []peer.PeerBytes{}} + } + return r.peerClient.RelaySnapshot() +} diff --git a/backend/radiance_test.go b/backend/radiance_test.go index 183b4cba..005eb0d0 100644 --- a/backend/radiance_test.go +++ b/backend/radiance_test.go @@ -398,8 +398,9 @@ func (f *fakePeerController) Stop(_ context.Context) error { return nil } -func (f *fakePeerController) IsActive() bool { return f.active.Load() } -func (f *fakePeerController) CurrentStatus() peer.Status { return peer.Status{Active: f.active.Load()} } +func (f *fakePeerController) IsActive() bool { return f.active.Load() } +func (f *fakePeerController) CurrentStatus() peer.Status { return peer.Status{Active: f.active.Load()} } +func (f *fakePeerController) RelaySnapshot() peer.Snapshot { return peer.Snapshot{} } // newPeerTestBackend wires a minimal LocalBackend with only the fields // applyPeerShare touches. settings is initialized to a fresh tempdir per test @@ -699,8 +700,9 @@ func (f *slowStartFake) Stop(_ context.Context) error { f.active.Store(false) return nil } -func (f *slowStartFake) IsActive() bool { return f.active.Load() } -func (f *slowStartFake) CurrentStatus() peer.Status { return peer.Status{Active: f.active.Load()} } +func (f *slowStartFake) IsActive() bool { return f.active.Load() } +func (f *slowStartFake) CurrentStatus() peer.Status { return peer.Status{Active: f.active.Load()} } +func (f *slowStartFake) RelaySnapshot() peer.Snapshot { return peer.Snapshot{} } // newCloseableTestBackend mirrors newPeerTestBackend but provides the fields // Close needs (closeOnce, cancel) so we can exercise the shutdown path diff --git a/go.mod b/go.mod index 641e28b8..207224d7 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.26.9 // replace github.com/sagernet/sing => github.com/getlantern/sing v0.7.18-lantern -replace github.com/sagernet/sing-box => github.com/getlantern/sing-box-minimal v1.13.20-0.20261017021932-5a256d757969 +replace github.com/sagernet/sing-box => github.com/getlantern/sing-box-minimal v1.13.20-0.20261019115333-624aa48e0252 // replace github.com/sagernet/wireguard-go => github.com/getlantern/wireguard-go v0.0.1-beta.7.0.20251208214020-d78e69f1eff4 @@ -40,7 +40,7 @@ require ( github.com/getlantern/common v1.2.1-0.20260910154003-08e030318f24 github.com/getlantern/domainfront v0.0.0-20260915172349-4984689a8667 github.com/getlantern/kindling v0.0.0-20261014210827-0f4a89dbe1ee - github.com/getlantern/lantern-box v0.0.154 + github.com/getlantern/lantern-box v0.0.166 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-20261011133221-f5e57ec9c1c5 @@ -126,7 +126,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.20260819153658-ebc74116a064 // indirect + github.com/getlantern/samizdat v0.0.3-0.20261019142234-18bd41c612f1 // 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 b53f73f3..bddd898d 100644 --- a/go.sum +++ b/go.sum @@ -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-20261014210827-0f4a89dbe1ee h1:FXod6NjAD/KDl6Aj5mP1MVCLf2AAJv8qU4kmUrYcMZk= github.com/getlantern/kindling v0.0.0-20261014210827-0f4a89dbe1ee/go.mod h1:98jp1v8UyXJRGn8slpvzNz2tKnyzvSqtlVYkQ+kYapQ= -github.com/getlantern/lantern-box v0.0.154 h1:EPQjjBqNC1gbDb2SlpbWg1wo7pNM/TccZ/UV+yhVQjI= -github.com/getlantern/lantern-box v0.0.154/go.mod h1:gy3dGHMTQXaFi5DYIJJrMNT3WS2E7GOxuZl2jebcbuw= +github.com/getlantern/lantern-box v0.0.166 h1:Z2qy+5Yvd6s0pQBGlqP/bwRllkynK5Clf94HcHzy13Q= +github.com/getlantern/lantern-box v0.0.166/go.mod h1:RK7qWraZxaDGC6iJYJV6KEUyQvOy2njF5iEENZ1bl/k= 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,12 +261,12 @@ 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.20260819153658-ebc74116a064 h1:NhCyQAWGEWjok/jg4gRD6fzCZvyOQy/4dVSFYhzU5xk= -github.com/getlantern/samizdat v0.0.3-0.20260819153658-ebc74116a064/go.mod h1:uEeykQSW2/6rTjfPlj3MTTo59poSHXfAHTGgzYDkbr0= +github.com/getlantern/samizdat v0.0.3-0.20261019142234-18bd41c612f1 h1:Q1AHHHTd3eiHXn0HR0mltq6/4cGnv9cQFKYmoRuXWQo= +github.com/getlantern/samizdat v0.0.3-0.20261019142234-18bd41c612f1/go.mod h1:uEeykQSW2/6rTjfPlj3MTTo59poSHXfAHTGgzYDkbr0= github.com/getlantern/semconv v0.0.0-20261011133221-f5e57ec9c1c5 h1:oF5kQTI7RwMvjMnr4WhmQ1beAjSMl1zoQ4I7Cx5nm6I= github.com/getlantern/semconv v0.0.0-20261011133221-f5e57ec9c1c5/go.mod h1:GkPT5P9JoOTIRXRmFWxYgu1hhXgTFFTNc2hoG7WQc3g= -github.com/getlantern/sing-box-minimal v1.13.20-0.20261017021932-5a256d757969 h1:xDlY1aEy5FILPxlncY5ciHQQY5FdyiNNZ2RC86ACfsw= -github.com/getlantern/sing-box-minimal v1.13.20-0.20261017021932-5a256d757969/go.mod h1:4E8WvF7Vi60N1NtPw1CeG994J/yiJZZkty5dKKwG+V0= +github.com/getlantern/sing-box-minimal v1.13.20-0.20261019115333-624aa48e0252 h1:p6tbxiX1SHqo1FwO9l1uYerSSaMLt/R78QtLIScpP54= +github.com/getlantern/sing-box-minimal v1.13.20-0.20261019115333-624aa48e0252/go.mod h1:4E8WvF7Vi60N1NtPw1CeG994J/yiJZZkty5dKKwG+V0= github.com/getlantern/soar v0.2.3 h1:BAahWy3b4J6wtrAJD7WgoO2O+x2Q3T6KN2r0E3ogNgs= github.com/getlantern/soar v0.2.3/go.mod h1:6T2+Kbx0GVqYIHqfEaUqlYI5ZRRVZYYhuoMWcxHC+xA= github.com/getlantern/twiddle v0.0.0-20260914204124-f698a0d67d2f h1:8WL/o8hKl27jpHunB4nO5dJZKSTfyMwvS7J72ix6vqQ= diff --git a/ipc/client.go b/ipc/client.go index 3555c7dc..c9972f7c 100644 --- a/ipc/client.go +++ b/ipc/client.go @@ -111,6 +111,14 @@ func (c *Client) PeerStatus(ctx context.Context) (peer.Status, error) { return status, err } +// PeerRelaySnapshot reads the bytes the current or most recent direct-sharing +// session has relayed, in the shape of UnboundedSnapshot. +func (c *Client) PeerRelaySnapshot(ctx context.Context) (peer.Snapshot, error) { + var s peer.Snapshot + err := c.doJSON(ctx, http.MethodGet, peerSnapshotEndpoint, nil, &s) + return s, err +} + ///////////// // VPN // ///////////// diff --git a/ipc/server.go b/ipc/server.go index e4bc047a..71537696 100644 --- a/ipc/server.go +++ b/ipc/server.go @@ -81,6 +81,7 @@ const ( peerStatusEndpoint = "/peer/status" peerStatusEventsEndpoint = "/peer/status/events" peerConnectionEventsEndpoint = "/peer/connection/events" + peerSnapshotEndpoint = "/peer/snapshot" // Unbounded ("Basic mode") endpoints. Connection events have the same // JSON shape as /peer/connection/events but are emitted from a @@ -276,6 +277,9 @@ func newLocalAPI(b *backend.LocalBackend, withAuth bool) *localapi { // SSE skips the tracer middleware since it buffers the entire response body. mux.HandleFunc("GET "+peerStatusEventsEndpoint, s.peerStatusEventsHandler) mux.HandleFunc("GET "+peerConnectionEventsEndpoint, s.peerConnectionEventsHandler) + mux.HandleFunc("GET "+peerSnapshotEndpoint, traced(func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.backend(r.Context()).PeerRelaySnapshot()) + })) mux.HandleFunc("GET "+unboundedConnectionEventsEndpoint, s.unboundedConnectionEventsHandler) mux.HandleFunc("GET "+unboundedSnapshotEndpoint, traced(func(w http.ResponseWriter, r *http.Request) { diff --git a/peer/peer.go b/peer/peer.go index fd1a9e85..a1d0c920 100644 --- a/peer/peer.go +++ b/peer/peer.go @@ -203,6 +203,14 @@ type Client struct { // streams over one TCP conn, so several can share a source. Guarded by // connsMu. verifierConns map[string]int + // verifierCounters holds the byte counters of open verifier streams. When + // events carry a counter it identifies the stream, so a close is hidden + // exactly when its own accept was; verifierConns is the fallback for + // events without one. Guarded by connsMu. + verifierCounters map[*peerconn.Counter]struct{} + + // relay totals the bytes each session relays, for RelaySnapshot. + relay relayStats // rejections tallies destinations refused by reject rules, across // sessions, so allowlist gaps are visible. Flushed on each heartbeat and @@ -330,6 +338,7 @@ func (c *Client) Start(ctx context.Context) (retErr error) { // silence the disconnect cascade during box.Close; if we don't reset // here, a Stop→Start cycle would leave the wrapper permanently muted. c.listenerDraining.Store(false) + relayGen := c.relay.begin() var ( success bool @@ -362,6 +371,7 @@ func (c *Client) Start(ctx context.Context) (retErr error) { c.listenerDraining.Store(true) peerconn.SetListener(nil) c.resetConnTracking() + c.relay.end() if box != nil { _ = c.closeBox(box) } @@ -520,7 +530,7 @@ func (c *Client) Start(ctx context.Context) (retErr error) { if evt.Rejected { c.rejections.record(evt.Destination) } - if c.isVerifierConn(state, source, evt.Destination) { + if c.isVerifierConn(state, source, evt.Destination, evt.Bytes) { slog.Debug("peer listener: hiding verifier connection", "state", state, "source", source) return @@ -528,6 +538,7 @@ func (c *Client) Start(ctx context.Context) (retErr error) { 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()}) }) slog.Info("peer listener: registered with peerconn", "route_id", regResp.RouteID) @@ -700,6 +711,7 @@ func (c *Client) stopHeld(ctx context.Context, target stopTarget) (bool, error) c.listenerDraining.Store(true) peerconn.SetListener(nil) c.resetConnTracking() + c.relay.end() cancel() <-done @@ -833,6 +845,7 @@ func (c *Client) resetConnTracking() { defer c.connsMu.Unlock() c.connsByIP = nil c.verifierConns = nil + c.verifierCounters = nil } // verifierCallbackHosts are the hosts lantern-cloud's verifier fetches through @@ -844,11 +857,12 @@ var verifierCallbackHosts = []string{"api.iantem.io", "api.staging.iantem.io"} // isVerifierConn reports whether a connection event belongs to lantern-cloud's // verifier, which re-verifies serving peers too, so that it is neither shown // as a person helped nor reported to the server as a client. An accept is -// matched by destination. A close carries no destination, so each hidden -// accept hides one later close from the same source. When ordinary and -// verifier streams share a source, which close is hidden may not be the -// verifier's own, but the per-IP totals still balance. -func (c *Client) isVerifierConn(state int, source, destination string) bool { +// matched by destination. A close carries no destination, so it is matched by +// its byte counter, which identifies the stream. Without a counter, each +// hidden accept hides one later close from the same source; when ordinary and +// verifier streams share a source that may not be the verifier's own close, +// but the per-IP totals still balance. +func (c *Client) isVerifierConn(state int, source, destination string, bytes *peerconn.Counter) bool { c.connsMu.Lock() defer c.connsMu.Unlock() // Same reason as trackConn: a late accept must not repopulate the set @@ -857,6 +871,13 @@ func (c *Client) isVerifierConn(state int, source, destination string) bool { return false } if state < 0 { + if bytes != nil { + if _, ok := c.verifierCounters[bytes]; !ok { + return false + } + delete(c.verifierCounters, bytes) + return true + } n := c.verifierConns[source] if n == 0 { return false @@ -877,6 +898,13 @@ func (c *Client) isVerifierConn(state int, source, destination string) bool { }) { return false } + if bytes != nil { + if c.verifierCounters == nil { + c.verifierCounters = make(map[*peerconn.Counter]struct{}) + } + c.verifierCounters[bytes] = struct{}{} + return true + } if c.verifierConns == nil { c.verifierConns = make(map[string]int) } diff --git a/peer/peer_test.go b/peer/peer_test.go index 95cff6f7..cba923f8 100644 --- a/peer/peer_test.go +++ b/peer/peer_test.go @@ -1236,17 +1236,17 @@ func TestTrackConn_IgnoresUnmatchedClose(t *testing.T) { func TestIsVerifierConn_PairsCloseWithAccept(t *testing.T) { c := &Client{} - assert.True(t, c.isVerifierConn(1, "35.233.177.182:33792", "api.iantem.io:443")) - assert.True(t, c.isVerifierConn(1, "35.233.177.183:1000", "API.staging.iantem.io:443")) - assert.False(t, c.isVerifierConn(1, "35.233.177.182:33793", "google.com:443")) + assert.True(t, c.isVerifierConn(1, "35.233.177.182:33792", "api.iantem.io:443", nil)) + assert.True(t, c.isVerifierConn(1, "35.233.177.183:1000", "API.staging.iantem.io:443", nil)) + assert.False(t, c.isVerifierConn(1, "35.233.177.182:33793", "google.com:443", nil)) - assert.True(t, c.isVerifierConn(-1, "35.233.177.182:33792", "")) - assert.False(t, c.isVerifierConn(-1, "35.233.177.182:33792", ""), + assert.True(t, c.isVerifierConn(-1, "35.233.177.182:33792", "", nil)) + assert.False(t, c.isVerifierConn(-1, "35.233.177.182:33792", "", nil), "a second close for the same source is not the verifier's") - assert.False(t, c.isVerifierConn(-1, "35.233.177.182:33793", "")) + assert.False(t, c.isVerifierConn(-1, "35.233.177.182:33793", "", nil)) c.resetConnTracking() - assert.False(t, c.isVerifierConn(-1, "35.233.177.183:1000", ""), + assert.False(t, c.isVerifierConn(-1, "35.233.177.183:1000", "", nil), "a reset forgets verifier connections from the old session") } diff --git a/peer/snapshot.go b/peer/snapshot.go new file mode 100644 index 00000000..d7fe3b90 --- /dev/null +++ b/peer/snapshot.go @@ -0,0 +1,274 @@ +package peer + +import ( + "net" + "sort" + "strconv" + "sync" + "time" + + "github.com/getlantern/lantern-box/tracker/peerconn" +) + +// relayRetention keeps a departed client's byte totals in the snapshot long +// enough for the app's 1s poller to observe its final count. +const relayRetention = 60 * time.Second + +// Snapshot is the data direct sharing has relayed, in the shape of +// unbounded.Snapshot so the app can fold both with the same code. Like +// Unbounded's, Epoch is fixed for the process and byte counts are cumulative +// across sessions, so a Stop and Start between two polls loses nothing. +type Snapshot struct { + Epoch string `json:"epoch"` + Running bool `json:"running"` + Peers []string `json:"peers"` + BytesDown uint64 `json:"bytesDown"` + BytesUp uint64 `json:"bytesUp"` + PeerBytes []PeerBytes `json:"peerBytes"` +} + +// PeerBytes is the cumulative traffic relayed for one client IP. An entry +// pruned after relayRetention restarts at zero if the IP returns, which +// consumers treat as a reset. +type PeerBytes struct { + IP string `json:"ip"` + Down uint64 `json:"down"` + Up uint64 `json:"up"` +} + +// relayTally is one client IP's bytes: closed holds its finished +// connections' totals; its open connections are read live from relay.conns. +type relayTally struct { + closedDown, closedUp uint64 + open int + endedAt time.Time +} + +// relayConn is an open connection's source and, once a later session has +// begun, when its own session was retired. A retired connection may still be +// draining in a box that has not finished closing, so its counter is read +// live for relayRetention before being folded into the totals. +type relayConn struct { + source string + retiredAt time.Time +} + +// relayStats accumulates relayed bytes from the counters lantern-box +// attaches to each connection. It has its own lock because +// Snapshot is polled every second and must not wait on Start or Stop. +type relayStats struct { + mu sync.Mutex + epoch string + // conns holds each open connection's counter. The counter, not the + // source, is the key: samizdat multiplexes streams over one TCP conn, so + // several open streams can share a source, and the same counter comes + // back on each stream's close. + conns map[*peerconn.Counter]relayConn + ips map[string]*relayTally + closedDown, closedUp uint64 + // endedAt is when the session stopped, or zero while it runs. Stop drops + // the close events of connections still open, so those are retired by + // time instead: relayRetention after endedAt, or at the next begin. + endedAt time.Time + // gen identifies the running session and changes at every begin and end. + // A listener callback that resumes after Stop, or after Stop and Start, + // carries an old generation, so its event is ignored rather than counted + // after the session it belongs to has ended. + gen uint64 + now func() time.Time +} + +func (r *relayStats) clock() time.Time { + if r.now != nil { + return r.now() + } + return time.Now() +} + +// begin starts a session and returns its generation. Totals carry on from +// earlier sessions under the same epoch. Connections the previous session left +// open are retired: no longer listed as peers, but still read live until +// relayRetention has passed, since the old box may still be draining them. +func (r *relayStats) begin() uint64 { + r.mu.Lock() + defer r.mu.Unlock() + if r.epoch == "" { + r.epoch = strconv.FormatInt(r.clock().UnixNano(), 10) + } + now := r.clock() + for c, rc := range r.conns { + if !rc.retiredAt.IsZero() { + continue + } + rc.retiredAt = now + r.conns[c] = rc + if t := r.ips[ipOf(rc.source)]; t != nil { + t.release(now) + } + } + r.endedAt = time.Time{} + r.gen++ + return r.gen +} + +// end marks the session stopped and ends its generation, so the listener's +// late events are ignored. Its open connections' counters stay readable for +// relayRetention, long enough for the app's poller to see the bytes relayed +// while the box closed, and are then folded into the totals and dropped. +func (r *relayStats) end() { + r.mu.Lock() + defer r.mu.Unlock() + r.endedAt = r.clock() + r.gen++ +} + +// foldLocked moves a finished connection's count into the totals and its +// client's tally. Caller must hold r.mu and remove c from r.conns. +func (r *relayStats) foldLocked(c *peerconn.Counter, source string) { + down, up := c.Down(), c.Up() + r.closedDown += down + r.closedUp += up + if t := r.ips[ipOf(source)]; t != nil { + t.closedDown += down + t.closedUp += up + } +} + +// release ends one of a client's open connections. +func (t *relayTally) release(now time.Time) { + t.open-- + if t.open <= 0 { + t.open = 0 + t.endedAt = now + } +} + +func ipOf(source string) string { + ip, _, err := net.SplitHostPort(source) + if err != nil { + return source + } + return ip +} + +// pruneLocked folds connections retired more than relayRetention ago into the +// totals, does the same for every connection once a stopped session is past +// it, and drops clients that left more than relayRetention ago. Caller must +// hold r.mu. +func (r *relayStats) pruneLocked(now time.Time) { + ended := !r.endedAt.IsZero() && now.Sub(r.endedAt) > relayRetention + for c, rc := range r.conns { + if ended || (!rc.retiredAt.IsZero() && now.Sub(rc.retiredAt) > relayRetention) { + r.foldLocked(c, rc.source) + delete(r.conns, c) + } + } + if ended { + r.ips = nil + return + } + for ip, t := range r.ips { + if t.open == 0 && now.Sub(t.endedAt) > relayRetention { + delete(r.ips, ip) + } + } +} + +// record follows one connection's accept or close. A close folds the +// connection's final count, so totals keep it after the connection is gone. +func (r *relayStats) record(state int, source string, bytes *peerconn.Counter) { + r.mu.Lock() + gen := r.gen + r.mu.Unlock() + r.recordFor(gen, state, source, bytes) +} + +// 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) { + if bytes == nil { + return + } + ip, _, err := net.SplitHostPort(source) + if err != nil { + ip = source + } + r.mu.Lock() + defer r.mu.Unlock() + if gen != r.gen { + return + } + if state > 0 { + // Pruned here too, so a session no one polls does not keep every + // client it has ever seen. Before acquiring, so a returning IP's + // fresh tally does not inherit its pruned totals. + r.pruneLocked(r.clock()) + } + if r.ips == nil { + r.ips = make(map[string]*relayTally) + } + if r.conns == nil { + r.conns = make(map[*peerconn.Counter]relayConn) + } + switch { + case state > 0: + r.conns[bytes] = relayConn{source: source} + t := r.ips[ip] + if t == nil { + t = &relayTally{} + r.ips[ip] = t + } + t.open++ + case state < 0: + rc, ok := r.conns[bytes] + if !ok || !rc.retiredAt.IsZero() { + return + } + delete(r.conns, bytes) + r.foldLocked(bytes, rc.source) + if t := r.ips[ip]; t != nil { + t.release(r.clock()) + } + } +} + +func (r *relayStats) snapshot(running bool) Snapshot { + r.mu.Lock() + defer r.mu.Unlock() + r.pruneLocked(r.clock()) + s := Snapshot{ + Epoch: r.epoch, + Running: running, + Peers: []string{}, + BytesDown: r.closedDown, + BytesUp: r.closedUp, + PeerBytes: []PeerBytes{}, + } + live := make(map[string]PeerBytes, len(r.ips)) + for c, rc := range r.conns { + down, up := c.Down(), c.Up() + s.BytesDown += down + s.BytesUp += up + if running && rc.retiredAt.IsZero() { + s.Peers = append(s.Peers, rc.source) + } + ip := ipOf(rc.source) + pb := live[ip] + pb.Down += down + pb.Up += up + live[ip] = pb + } + 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}) + } + sort.Strings(s.Peers) + sort.Slice(s.PeerBytes, func(i, j int) bool { return s.PeerBytes[i].IP < s.PeerBytes[j].IP }) + return s +} + +// RelaySnapshot reports what the current or most recent direct-sharing +// session has relayed. +func (c *Client) RelaySnapshot() Snapshot { + return c.relay.snapshot(c.CurrentStatus().Phase == PhaseServing) +} diff --git a/peer/snapshot_test.go b/peer/snapshot_test.go new file mode 100644 index 00000000..5419391e --- /dev/null +++ b/peer/snapshot_test.go @@ -0,0 +1,283 @@ +package peer + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/getlantern/lantern-box/tracker/peerconn" +) + +func counter(down, up int) *peerconn.Counter { + c := &peerconn.Counter{} + c.AddDown(down) + c.AddUp(up) + return c +} + +// Open connections are read live; a close folds its final count so the totals +// keep it, and a client's connections add up under its IP. +func TestRelayStats_CountsLiveAndClosedConnections(t *testing.T) { + var r relayStats + r.begin() + a, b := counter(100, 10), counter(50, 5) + r.record(1, "203.0.113.7:1000", a) + r.record(1, "203.0.113.7:1001", b) + other := counter(7, 3) + r.record(1, "198.51.100.9:2000", other) + + a.AddDown(900) + s := r.snapshot(true) + assert.Equal(t, uint64(1057), s.BytesDown) + assert.Equal(t, uint64(18), s.BytesUp) + assert.Equal(t, []string{"198.51.100.9:2000", "203.0.113.7:1000", "203.0.113.7:1001"}, s.Peers) + assert.Equal(t, []PeerBytes{ + {IP: "198.51.100.9", Down: 7, Up: 3}, + {IP: "203.0.113.7", Down: 1050, Up: 15}, + }, s.PeerBytes) + + r.record(-1, "203.0.113.7:1000", a) + a.AddDown(1) // the counter is no longer read once its close is folded + s = r.snapshot(true) + assert.Equal(t, uint64(1057), s.BytesDown, "a closed connection's bytes stay counted") + assert.Equal(t, []PeerBytes{ + {IP: "198.51.100.9", Down: 7, Up: 3}, + {IP: "203.0.113.7", Down: 1050, Up: 15}, + }, s.PeerBytes) + assert.NotContains(t, s.Peers, "203.0.113.7:1000") +} + +// samizdat multiplexes streams over one TCP conn, so two open streams can share +// a source; each keeps its own count, and closing one leaves the other live. +func TestRelayStats_StreamsSharingASource(t *testing.T) { + var r relayStats + r.begin() + a, b := counter(100, 1), counter(200, 2) + r.record(1, "203.0.113.7:1000", a) + r.record(1, "203.0.113.7:1000", b) + assert.Equal(t, uint64(300), r.snapshot(true).BytesDown) + + r.record(-1, "203.0.113.7:1000", a) + b.AddDown(50) + s := r.snapshot(true) + assert.Equal(t, uint64(350), s.BytesDown, "the open stream is still read live") + assert.Equal(t, []PeerBytes{{IP: "203.0.113.7", Down: 350, Up: 3}}, s.PeerBytes) + assert.Equal(t, []string{"203.0.113.7:1000"}, s.Peers) +} + +// An ordinary stream and an api.iantem.io stream share a source, and the +// ordinary one closes first. Matched by counter, its close is not mistaken for +// the verifier's, so it leaves the snapshot and stops being held. +func TestClient_OrdinaryCloseBeforeVerifierOnSharedSource(t *testing.T) { + fwd := &fakeForwarder{externalIP: "203.0.113.42"} + c := newTestClient(t, fwd, &fakeBoxService{}, newStubServer(t)) + ctx := context.Background() + require.NoError(t, c.Start(ctx)) + defer func() { require.NoError(t, c.Stop(ctx)) }() + + const src = "5.119.14.105:10304" + ordinary, verifier := counter(300, 30), counter(9, 9) + peerconn.Notify(peerconn.Event{State: +1, Source: src, Destination: "google.com:443", Bytes: ordinary}) + peerconn.Notify(peerconn.Event{State: +1, Source: src, Destination: "api.iantem.io:443", Bytes: verifier}) + peerconn.Notify(peerconn.Event{State: -1, Source: src, Bytes: ordinary}) + assert.Equal(t, 0, c.ActiveClients()) + assert.Empty(t, c.RelaySnapshot().Peers) + peerconn.Notify(peerconn.Event{State: -1, Source: src, Bytes: verifier}) + assert.Equal(t, 0, c.ActiveClients()) + s := c.RelaySnapshot() + assert.Equal(t, uint64(300), s.BytesDown, "only the ordinary stream counts") + c.relay.mu.Lock() + assert.Empty(t, c.relay.conns) + c.relay.mu.Unlock() +} + +// A departed client stays in the snapshot long enough for the app's poller to +// see its final count, then drops out; session totals keep its bytes. +func TestRelayStats_PrunesDepartedClientsAfterRetention(t *testing.T) { + now := time.Unix(1_000, 0) + r := relayStats{now: func() time.Time { return now }} + r.begin() + c := counter(40, 4) + r.record(1, "203.0.113.7:1000", c) + r.record(-1, "203.0.113.7:1000", c) + + now = now.Add(relayRetention) + assert.Len(t, r.snapshot(true).PeerBytes, 1) + now = now.Add(time.Second) + s := r.snapshot(true) + assert.Empty(t, s.PeerBytes) + assert.Equal(t, uint64(40), s.BytesDown) +} + +// With no one polling, departed clients are still pruned as new ones arrive, +// so memory does not grow with lifetime client churn. +func TestRelayStats_PrunesWithoutSnapshots(t *testing.T) { + now := time.Unix(1_000, 0) + r := relayStats{now: func() time.Time { return now }} + r.begin() + old := counter(10, 1) + r.record(1, "203.0.113.7:1000", old) + r.record(-1, "203.0.113.7:1000", old) + now = now.Add(relayRetention + time.Second) + r.record(1, "198.51.100.9:2000", counter(5, 0)) + r.mu.Lock() + _, kept := r.ips["203.0.113.7"] + r.mu.Unlock() + assert.False(t, kept) + assert.Equal(t, uint64(15), r.snapshot(true).BytesDown) +} + +// Stop drops the close events of connections still open, so a stopped +// session's open connections are retired by time: still read live for the +// retention window, then folded into the totals, which never go backwards. +func TestRelayStats_RetiresOpenConnectionsAfterStop(t *testing.T) { + now := time.Unix(1_000, 0) + r := relayStats{now: func() time.Time { return now }} + r.begin() + c := counter(100, 10) + r.record(1, "203.0.113.7:1000", c) + r.end() + c.AddDown(20) // bytes moved while the box closed + assert.Equal(t, uint64(120), r.snapshot(false).BytesDown) + + now = now.Add(relayRetention + time.Second) + s := r.snapshot(false) + assert.Equal(t, uint64(120), s.BytesDown, "folded, not lost") + assert.Equal(t, uint64(10), s.BytesUp) + assert.Empty(t, s.PeerBytes) + r.mu.Lock() + assert.Empty(t, r.conns, "the counter is no longer held") + r.mu.Unlock() +} + +// A listener callback from the previous session can pass its drain check, +// stall, and resume after Stop→Start. Its generation is stale, so its counter +// is not counted in the new session or held by it. +func TestRelayStats_IgnoresCallbacksFromAnEarlierSession(t *testing.T) { + var r relayStats + old := r.begin() + r.end() + current := r.begin() + require.NotEqual(t, old, current) + + stale := counter(500, 50) + r.recordFor(old, 1, "203.0.113.7:1000", stale) + r.recordFor(current, 1, "198.51.100.9:2000", counter(7, 3)) + s := r.snapshot(true) + assert.Equal(t, uint64(7), s.BytesDown) + assert.Equal(t, []string{"198.51.100.9:2000"}, s.Peers) + r.mu.Lock() + _, held := r.conns[stale] + r.mu.Unlock() + assert.False(t, held) +} + +// Like Unbounded's, the epoch is fixed for the process and totals carry +// across sessions, so a Stop and Start between two of the app's polls loses +// none of the stopped session's bytes; its open connections fold in at begin. +func TestRelayStats_TotalsCarryAcrossSessions(t *testing.T) { + now := time.Unix(1_000, 0) + r := relayStats{now: func() time.Time { return now }} + gen := r.begin() + first := r.snapshot(true).Epoch + c := counter(10, 1) + r.recordFor(gen, 1, "203.0.113.7:1000", c) + r.end() + assert.Empty(t, r.snapshot(false).Peers, "a stopped session lists no peers") + + c.AddDown(5) // relayed after the last poll, while the box closed + now = now.Add(time.Second) + gen = r.begin() + s := r.snapshot(true) + assert.Equal(t, first, s.Epoch) + assert.Equal(t, uint64(15), s.BytesDown, "nothing lost across the restart") + assert.Equal(t, []PeerBytes{{IP: "203.0.113.7", Down: 15, Up: 1}}, s.PeerBytes) + + r.recordFor(gen, 1, "198.51.100.9:2000", counter(7, 0)) + assert.Equal(t, uint64(22), r.snapshot(true).BytesDown) +} + +// closeBox can return before the old box finishes closing, so a stopped +// session's connections can keep moving bytes after the next begin. Their +// counters stay readable, though no longer listed as peers, until retention +// ends, and are then folded into the totals with nothing lost. +func TestRelayStats_DrainingCountersSurviveARestart(t *testing.T) { + now := time.Unix(1_000, 0) + r := relayStats{now: func() time.Time { return now }} + gen := r.begin() + old := counter(10, 1) + r.recordFor(gen, 1, "203.0.113.7:1000", old) + r.end() + r.begin() + + old.AddDown(5) // still draining in the old box + s := r.snapshot(true) + assert.Equal(t, uint64(15), s.BytesDown) + assert.Empty(t, s.Peers, "a retired connection is not a current peer") + assert.Equal(t, []PeerBytes{{IP: "203.0.113.7", Down: 15, Up: 1}}, s.PeerBytes) + + now = now.Add(relayRetention + time.Second) + s = r.snapshot(true) + assert.Equal(t, uint64(15), s.BytesDown, "folded at retention, never backwards") + r.mu.Lock() + assert.Empty(t, r.conns, "the counter is released") + r.mu.Unlock() + old.AddDown(100) // after the fold the counter is no longer read + assert.Equal(t, uint64(15), r.snapshot(true).BytesDown) +} + +// A verifier accept can pass the drain check, stall, and resume after Stop, +// where isVerifierConn no longer recognizes it. end advances the generation, +// so the late event is not counted in the stopped session. +func TestRelayStats_IgnoresEventsAfterEnd(t *testing.T) { + var r relayStats + gen := r.begin() + r.end() + r.recordFor(gen, 1, "35.233.177.182:33792", counter(500, 50)) + s := r.snapshot(false) + assert.Zero(t, s.BytesDown) + assert.Empty(t, s.PeerBytes) +} + +// Connections without a counter, and closes with no recorded accept, are +// ignored rather than miscounted. +func TestRelayStats_IgnoresUncountedAndUnmatchedEvents(t *testing.T) { + var r relayStats + r.begin() + r.record(1, "203.0.113.7:1000", nil) + r.record(-1, "203.0.113.7:1001", counter(5, 5)) + s := r.snapshot(true) + assert.Zero(t, s.BytesDown) + assert.Empty(t, s.PeerBytes) +} + +// End to end through a started client: counted connections show up in +// RelaySnapshot, and lantern-cloud's verifier does not. +func TestClient_RelaySnapshot(t *testing.T) { + fwd := &fakeForwarder{externalIP: "203.0.113.42"} + c := newTestClient(t, fwd, &fakeBoxService{}, newStubServer(t)) + ctx := context.Background() + require.NoError(t, c.Start(ctx)) + + client := counter(0, 0) + peerconn.Notify(peerconn.Event{State: +1, Source: "5.119.14.105:10304", Destination: "google.com:443", Bytes: client}) + peerconn.Notify(peerconn.Event{State: +1, Source: "35.233.177.182:33792", Destination: "api.iantem.io:443", Bytes: counter(500, 50)}) + client.AddDown(1200) + client.AddUp(300) + + s := c.RelaySnapshot() + assert.True(t, s.Running) + assert.NotEmpty(t, s.Epoch) + assert.Equal(t, uint64(1200), s.BytesDown) + assert.Equal(t, uint64(300), s.BytesUp) + assert.Equal(t, []string{"5.119.14.105:10304"}, s.Peers) + assert.Equal(t, []PeerBytes{{IP: "5.119.14.105", Down: 1200, Up: 300}}, s.PeerBytes) + + require.NoError(t, c.Stop(ctx)) + s = c.RelaySnapshot() + assert.False(t, s.Running) + assert.Equal(t, uint64(1200), s.BytesDown, "a poll after Stop still sees the final bytes") +}