feat: record the first join as a latency DAG (3RTT T01)

README.md

@@ -104,6 +104,21 @@ if _, err := call.AddTrack(track.TrackInfo(), track); err != nil {
 return writer.Write(audio.FromInt16(samples, 24000, 1))
 ```
 
+## Join latency
+
+A call records how its first join went: every network step, timer and local step, what it
+waited for, and each step in round trips to its peer. `String()` draws it as a DAG with
+the critical path marked; it marshals to JSON.
+
+```go
+call.OnJoinTrace(func(trace jointrace.Trace) {
+	log.Print(trace) // or json.Marshal(trace)
+})
+```
+
+It fires once media flows both ways, or after `rtc.JoinTraceTimeout`. `call.JoinTrace()`
+returns what has been recorded so far.
+
 ## Examples
 
 ```bash

call.go

@@ -28,8 +28,10 @@ import (
 	"github.com/GetStream/getstream-go-webrtc/coordinator/models"
 	"github.com/GetStream/getstream-go-webrtc/event"
 	"github.com/GetStream/getstream-go-webrtc/internal/atomicx"
+	"github.com/GetStream/getstream-go-webrtc/internal/netdelay"
 	"github.com/GetStream/getstream-go-webrtc/internal/ratelimit"
 	"github.com/GetStream/getstream-go-webrtc/internal/xerr"
+	"github.com/GetStream/getstream-go-webrtc/jointrace"
 	"github.com/GetStream/getstream-go-webrtc/logger"
 	"github.com/GetStream/getstream-go-webrtc/pc"
 	"github.com/GetStream/getstream-go-webrtc/rtcstats"
@@ -274,6 +276,8 @@ type Call struct {
 
 	joinOptions      []JoinOption
 	coordinatorState atomic.Pointer[CallState]
+	// trace records the first join's steps; see JoinTrace.
+	trace joinTracer
 
 	GetCred GetCredentialsFunc
 	cred    atomic.Pointer[models.Credentials]
@@ -382,6 +386,7 @@ func newCall(cc *Client, callType, callID string) *Call {
 	}
 	c.runReconnect = c.reconnect
 	c.fastReconnectViable = c.canFastReconnect
+	c.trace.udpRTT = c.selectedPairRTT
 	c.callCtx, c.callCancel = context.WithCancel(context.Background())
 	c.connState.Store(CallConnectionStateConnecting)
 	c.nextReconnectStrategy = sfu_models.WebsocketReconnectStrategy_WEBSOCKET_RECONNECT_STRATEGY_REJOIN
@@ -400,7 +405,7 @@ func (c *Call) joinCoordinator(ctx context.Context, options joinOptions) error {
 		return nil
 	}
 
-	result, getCred, err := c.cc.joinCoordinator(ctx, c.Type, c.Id, options.coordinatorRequest())
+	result, getCred, err := c.cc.joinCoordinator(ctx, c.Type, c.Id, options.coordinatorRequest(), c.trace.recorder())
 	if err != nil {
 		return xerr.Wrap(err)
 	}
@@ -804,6 +809,11 @@ func (c *Call) Join(ctx context.Context, opts ...JoinOption) (*sfu_events.JoinRe
 	for _, o := range opts {
 		o(&options)
 	}
+	reconnecting := options.reconnectDetails.GetStrategy() != sfu_models.WebsocketReconnectStrategy_WEBSOCKET_RECONNECT_STRATEGY_UNSPECIFIED
+	rec := c.trace.begin(time.Now(), reconnecting)
+	if !reconnecting {
+		c.cc.claimConnectTrace(rec)
+	}
 
 	if err := c.joinCoordinator(ctx, options); err != nil {
 		return nil, err
@@ -879,12 +889,18 @@ func (c *Call) Join(ctx context.Context, opts ...JoinOption) (*sfu_events.JoinRe
 		c.Client().Tracing.Store(c.Tracing.Load())
 	}
 
+	pcsStart := time.Now()
 	if err := c.initPubAndSub(options); err != nil {
 		return nil, xerr.Wrap(err)
 	}
+	rec.Add(jointrace.Span{
+		Name: jointrace.PCsCreate, After: afterFirst(rec, jointrace.CoordJoin),
+		Start: pcsStart, End: time.Now(), Kind: jointrace.KindLocal, Peer: jointrace.PeerLocal,
+	})
 
 	var resp *sfu_events.JoinResponse
-	resp, err = c.Client().Connect(ctx, req)
+	dialStart := time.Now()
+	resp, err = c.Client().Connect(jointrace.WithStep(ctx, rec, jointrace.SFUWSDial, jointrace.PeerSFU), req)
 	if err != nil {
 		// If ReconnectStrategy is still not specified here, it means the first join failed and we did not start the retry mechanism.
 		// We need to start it now, in the other case, we just need to return the error
@@ -895,6 +911,19 @@ func (c *Call) Join(ctx context.Context, opts ...JoinOption) (*sfu_events.JoinRe
 			return nil, xerr.Wrap(err)
 		}
 	}
+	joinedAt := time.Now()
+	dialedAt := c.Client().DialedAt()
+	rec.Add(jointrace.Span{
+		Name: jointrace.SFUWSDial, After: afterFirst(rec, jointrace.PCsCreate),
+		Start: dialStart, End: dialedAt, Kind: jointrace.KindNet, Peer: jointrace.PeerSFU,
+	})
+	rec.Add(jointrace.Span{
+		Name: jointrace.SFUJoin, After: []string{jointrace.SFUWSDial},
+		Start: dialedAt, End: joinedAt, Kind: jointrace.KindNet, Peer: jointrace.PeerSFU,
+	})
+	if !reconnecting {
+		c.trace.markJoined()
+	}
 	c.store.Store(NewParticipantStore(c, resp.CallState))
 	c.applyJoinResponse(resp)
 
@@ -989,6 +1018,16 @@ func (c *Call) initPubAndSub(options joinOptions) error {
 	c.sentSubscriptions = nil
 	c.subscribedTracksMu.Unlock()
 
+	if c.cc.networkDelay > 0 {
+		for _, conf := range []*pc.PeerConfig{&options.subscriberPeerConfig, &options.publisherPeerConfig} {
+			delayed, err := netdelay.NewNet(c.cc.networkDelay)
+			if err != nil {
+				return xerr.Wrap(err)
+			}
+			conf.SettingEngine.SetNet(delayed)
+		}
+	}
+
 	sub, err := newSubscriber(c, options.subscriber, options.subscriberPeerConfig, options.beforeSubscriberSendAnswer)
 	if err != nil {
 		return xerr.Wrap(err)
@@ -1045,6 +1084,9 @@ func (c *Call) signalOptions() []signal.Option {
 	if c.statsEnabled() {
 		opts = append(opts, signal.WithTracing())
 	}
+	if c.cc.networkDelay > 0 {
+		opts = append(opts, signal.WithDialContext(netdelay.Dialer(c.cc.networkDelay, nil)))
+	}
 	return opts
 }
 
@@ -1121,7 +1163,7 @@ func (c *Call) RefreshState(ctx context.Context) error {
 		return xerr.Error("call has not been joined")
 	}
 
-	result, _, err := c.cc.joinCoordinator(ctx, c.Type, c.Id, *state.JoinCallRequest)
+	result, _, err := c.cc.joinCoordinator(ctx, c.Type, c.Id, *state.JoinCallRequest, nil)
 	if err != nil {
 		return xerr.Wrap(err)
 	}
@@ -1148,6 +1190,7 @@ func (c *Call) RawHandler(event *sfu_events.SfuEvent) {
 // REJOIN or a Leave performs, and this runs on the signalling read loop, so a
 // panic here would take the process down with it.
 func (c *Call) OnSubscriberOffer(offer *sfu_events.SfuEvent_SubscriberOffer) {
+	c.subscriberOffer(time.Now())
 	sub := c.subscriberPeer()
 	if sub == nil {
 		c.logger.Warn("dropping subscriber offer: no subscriber peer connection")
@@ -1258,13 +1301,20 @@ func (c *Call) SubscribeToTracks(ctx context.Context, trackDetails ...*signal_rp
 // Callers that have to reach the SFU regardless -- a reconnect restoring state
 // onto a session that knows nothing about it -- go through here.
 func (c *Call) sendSubscriptions(ctx context.Context, trackDetails []*signal_rpc.TrackSubscriptionDetails) error {
-	resp, err := c.Client().UpdateSubscriptions(ctx, &signal_rpc.UpdateSubscriptionsRequest{
-		SessionId: c.SessionID.Load(),
-		Tracks:    trackDetails,
-	})
+	rec := c.trace.recorder()
+	start := time.Now()
+	resp, err := c.Client().UpdateSubscriptions(jointrace.WithStep(ctx, rec, jointrace.SubSubscribe, jointrace.PeerSFU),
+		&signal_rpc.UpdateSubscriptionsRequest{
+			SessionId: c.SessionID.Load(),
+			Tracks:    trackDetails,
+		})
 	if err != nil {
 		return xerr.Wrap(err)
 	}
+	rec.Add(jointrace.Span{
+		Name: jointrace.SubSubscribe, After: []string{jointrace.SFUJoin},
+		Start: start, End: time.Now(), Kind: jointrace.KindNet, Peer: jointrace.PeerSFU,
+	})
 
 	if err := resp.GetError(); err != nil {
 		return xerr.Wrap(fmt.Errorf("SubscribeToTracks error: %s", err.String()))

client.go

@@ -2,6 +2,7 @@ package rtc
 
 import (
 	"context"
+	"net/http"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -14,7 +15,9 @@ import (
 	"github.com/GetStream/getstream-go-webrtc/coordinator"
 	"github.com/GetStream/getstream-go-webrtc/coordinator/models"
 	"github.com/GetStream/getstream-go-webrtc/internal/atomicx"
+	"github.com/GetStream/getstream-go-webrtc/internal/netdelay"
 	"github.com/GetStream/getstream-go-webrtc/internal/xerr"
+	"github.com/GetStream/getstream-go-webrtc/jointrace"
 	"github.com/GetStream/getstream-go-webrtc/logger"
 	"github.com/GetStream/getstream-go-webrtc/rtcstats"
 )
@@ -58,6 +61,23 @@ type options struct {
 	// user is who NewRTCClient connects as. NewClient takes its user as an
 	// argument instead.
 	user User
+
+	// networkDelay is the round-trip time WithNetworkDelay adds to every connection.
+	networkDelay time.Duration
+}
+
+// WithNetworkDelay makes every connection the client opens behave as if it crossed a
+// network with round-trip time rtt: the coordinator REST and websocket connections, the
+// SFU websocket and RPCs, the location hint, and the peer connections' UDP sockets. Each
+// packet is held rtt/2 when sent and rtt/2 when received, in order, and a TCP connect
+// takes one rtt. It lets a test or bench against a local stack measure round trips as
+// if the stack were remote.
+//
+// For tests and benches only.
+func WithNetworkDelay(rtt time.Duration) Option {
+	return func(o *options) {
+		o.networkDelay = rtt
+	}
 }
 
 type Option func(*options)
@@ -285,6 +305,25 @@ type Client struct {
 	OwnUser      atomic.Pointer[models.OwnUserResponse]
 	Tracing      atomic.Pointer[rtcstats.TraceBuffer]
 	muStats      sync.RWMutex
+
+	// connectTrace holds the coordinator websocket's spans from NewClient. The first
+	// call to join claims them: later joins find the websocket already open.
+	connectTrace   *jointrace.Recorder
+	connectClaimed atomic.Bool
+}
+
+// claimConnectTrace copies the client's own connection spans into the first join's trace.
+func (c *Client) claimConnectTrace(rec *jointrace.Recorder) {
+	if rec == nil || c.connectTrace == nil || !c.connectClaimed.CompareAndSwap(false, true) {
+		return
+	}
+	t := c.connectTrace.Trace()
+	for _, s := range t.Spans {
+		rec.Add(s)
+	}
+	for peer, rtt := range t.RTT {
+		rec.SetRTT(peer, rtt)
+	}
 }
 
 func defaultClientOptions() options {
@@ -449,6 +488,9 @@ func newClient(apiKey string, user User, token TokenProvider, o options) (*Clien
 	if !c.withCoordinatorWS {
 		coordOptions = append(coordOptions, coordinator.WithoutWebsocket())
 	}
+	if o.networkDelay > 0 {
+		coordOptions = append(coordOptions, coordinator.WithDialContext(netdelay.Dialer(o.networkDelay, nil)))
+	}
 
 	cc, err := coordinator.NewClient(apiKey, userID, coordinator.StaticTokenProvider(tok),
 		coordinator.NoopHandler{}, coordOptions...)
@@ -470,10 +512,27 @@ func newClient(apiKey string, user User, token TokenProvider, o options) (*Clien
 			Token: tok,
 		}
 		c.Tracing.Load().Emit(rtcstats.CoordinatorWSConnectEvent, auth)
-		resp, err := connectWsWithRetries(ctx, cc, &auth)
+		rec := jointrace.NewRecorder(time.Now())
+		dialCtx := jointrace.WithStep(ctx, rec, jointrace.CoordWSDial, jointrace.PeerCoordinator)
+		start := time.Now()
+		resp, err := connectWsWithRetries(dialCtx, cc, &auth)
 		if err != nil {
 			return nil, err
 		}
+		// The upgrade response is the first byte back; the auth exchange follows it.
+		upgraded := jointrace.FirstByte(dialCtx)
+		if upgraded.IsZero() {
+			upgraded = start
+		}
+		rec.Add(jointrace.Span{
+			Name: jointrace.CoordWSDial, Start: start, End: upgraded,
+			Kind: jointrace.KindNet, Peer: jointrace.PeerCoordinator,
+		})
+		rec.Add(jointrace.Span{
+			Name: jointrace.CoordWSAuth, After: []string{jointrace.CoordWSDial}, Start: upgraded, End: time.Now(),
+			Kind: jointrace.KindNet, Peer: jointrace.PeerCoordinator,
+		})
+		c.connectTrace = rec
 		c.Tracing.Load().Emit(rtcstats.CoordinatorWSConnectedEvent, resp)
 		c.ConnectionID.Store(resp.ConnectionID)
 		c.OwnUser.Store(&resp.Me)
@@ -484,7 +543,13 @@ func newClient(apiKey string, user User, token TokenProvider, o options) (*Clien
 	c.CoordinatorClientInterface = cc
 	c.token.Store(tok)
 	if o.detectLocation {
-		c.locationDiscovery = NewCloudFrontDiscovery(o.locationHintURL, 3, locationHTTPClient, c.logger)
+		httpClient := locationHTTPClient
+		if o.networkDelay > 0 {
+			tr := locationHTTPClient.Transport.(*http.Transport).Clone()
+			tr.DialContext = netdelay.Dialer(o.networkDelay, tr.DialContext)
+			httpClient = &http.Client{Transport: tr, Timeout: locationHTTPClient.Timeout + 3*o.networkDelay}
+		}
+		c.locationDiscovery = NewCloudFrontDiscovery(o.locationHintURL, 3, httpClient, c.logger)
 	}
 	return c, nil
 }
@@ -554,10 +619,15 @@ func (c *Client) AddVideoSourceStatsProviders(providers ...VideoSourceStatsProvi
 	c.videoSourceStatsProviders = append(c.videoSourceStatsProviders, providers...)
 }
 
-func (c *Client) detectLocationCached(ctx context.Context) string {
+func (c *Client) detectLocationCached(ctx context.Context, rec *jointrace.Recorder) string {
 	if c.locationCache.Load() == "" {
 		c.logger.Info("Join call request without location, discovering location...")
-		loc := c.locationDiscovery.Discover(ctx)
+		start := time.Now()
+		loc := c.locationDiscovery.Discover(jointrace.WithStep(ctx, rec, jointrace.LocationHint, jointrace.PeerCloudFront))
+		rec.Add(jointrace.Span{
+			Name: jointrace.LocationHint, After: afterFirst(rec, jointrace.CoordWSAuth),
+			Start: start, End: time.Now(), Kind: jointrace.KindNet, Peer: jointrace.PeerCloudFront,
+		})
 		c.locationCache.Store(loc)
 		c.logger.Infof("%q location discovered", loc)
 	}
@@ -653,16 +723,27 @@ func (c *Client) joinCoordinator(
 	ctx context.Context,
 	callType, id string,
 	joinCallRequest models.JoinCallRequest,
+	rec *jointrace.Recorder,
 ) (*models.JoinCallResponse, GetCredentialsFunc, error) {
 	if joinCallRequest.Location == "" && c.detectLocation {
-		joinCallRequest.Location = c.detectLocationCached(ctx)
+		joinCallRequest.Location = c.detectLocationCached(ctx, rec)
 	}
 
 	c.Tracing.Load().Emit(rtcstats.CoordinatorConnectEvent, joinCallRequest)
-	result, err := c.connectWithRetries(ctx, callType, id, joinCallRequest)
+	start := time.Now()
+	joinCtx := jointrace.WithStep(ctx, rec, jointrace.CoordJoin, jointrace.PeerCoordinator)
+	result, err := c.connectWithRetries(joinCtx, callType, id, joinCallRequest)
 	if err != nil {
 		return nil, nil, err
 	}
+	note := ""
+	if jointrace.Reused(joinCtx) {
+		note = "reused connection"
+	}
+	rec.Add(jointrace.Span{
+		Name: jointrace.CoordJoin, After: afterFirst(rec, jointrace.LocationHint, jointrace.CoordWSAuth),
+		Start: start, End: time.Now(), Kind: jointrace.KindNet, Peer: jointrace.PeerCoordinator, Note: note,
+	})
 	c.Tracing.Load().Emit(rtcstats.CoordinatorConnectedEvent, result)
 
 	getCred := func(forceReload bool, excludeSFUID string) (models.Credentials, error) {