Merge branch 'v3' into v3-viam

.github/workflows/browser-e2e.yaml

@@ -5,9 +5,11 @@ on:
   pull_request:
     branches:
       - master
+      - v3
   push:
     branches:
       - master
+      - v3
 
 jobs:
   e2e-test:

.github/workflows/codeql-analysis.yml

@@ -20,6 +20,7 @@ on:
   pull_request:
     branches:
       - master
+      - v3
     paths:
       - '**.go'
 

.github/workflows/examples-tests.yaml

@@ -5,9 +5,11 @@ on:
   pull_request:
     branches:
       - master
+      - v3
   push:
     branches:
       - master
+      - v3
 
 jobs:
   pion-to-pion-test:

.github/workflows/test.yaml

@@ -16,28 +16,25 @@ on:
   push:
     branches:
       - master
+      - v3
   pull_request:
 
 jobs:
   test:
     uses: pion/.goassets/.github/workflows/test.reusable.yml@master
     strategy:
       matrix:
-        go: ['1.20', '1.19'] # auto-update/supported-go-version-list
+        go: ["1.20", "1.19"] # auto-update/supported-go-version-list
       fail-fast: false
     with:
       go-version: ${{ matrix.go }}
+    secrets: inherit
 
   test-i386:
     uses: pion/.goassets/.github/workflows/test-i386.reusable.yml@master
     strategy:
       matrix:
-        go: ['1.20', '1.19'] # auto-update/supported-go-version-list
+        go: ["1.20", "1.19"] # auto-update/supported-go-version-list
       fail-fast: false
     with:
       go-version: ${{ matrix.go }}
-
-  test-wasm:
-    uses: pion/.goassets/.github/workflows/test-wasm.reusable.yml@master
-    with:
-      go-version: '1.20' # auto-update/latest-go-version

.github/workflows/tidy-check.yaml

@@ -17,6 +17,7 @@ on:
   push:
     branches:
       - master
+      - v3
 
 jobs:
   tidy:

datachannel.go

@@ -40,6 +40,8 @@ type DataChannel struct {
 	readyState                 atomic.Value // DataChannelState
 	bufferedAmountLowThreshold uint64
 	detachCalled               bool
+	readLoopActive             chan struct{}
+	isGracefulClosed           bool
 
 	// The binaryType represents attribute MUST, on getting, return the value to
 	// which it was last set. On setting, if the new value is either the string
@@ -225,6 +227,10 @@ func (d *DataChannel) OnOpen(f func()) {
 func (d *DataChannel) onOpen() {
 	d.mu.RLock()
 	handler := d.onOpenHandler
+	if d.isGracefulClosed {
+		d.mu.RUnlock()
+		return
+	}
 	d.mu.RUnlock()
 
 	if handler != nil {
@@ -252,6 +258,10 @@ func (d *DataChannel) OnDial(f func()) {
 func (d *DataChannel) onDial() {
 	d.mu.RLock()
 	handler := d.onDialHandler
+	if d.isGracefulClosed {
+		d.mu.RUnlock()
+		return
+	}
 	d.mu.RUnlock()
 
 	if handler != nil {
@@ -261,6 +271,10 @@ func (d *DataChannel) onDial() {
 
 // OnClose sets an event handler which is invoked when
 // the underlying data transport has been closed.
+// Note: Due to backwards compatibility, there is a chance that
+// OnClose can be called, even if the GracefulClose is used.
+// If this is the case for you, you can deregister OnClose
+// prior to GracefulClose.
 func (d *DataChannel) OnClose(f func()) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
@@ -292,6 +306,10 @@ func (d *DataChannel) OnMessage(f func(msg DataChannelMessage)) {
 func (d *DataChannel) onMessage(msg DataChannelMessage) {
 	d.mu.RLock()
 	handler := d.onMessageHandler
+	if d.isGracefulClosed {
+		d.mu.RUnlock()
+		return
+	}
 	d.mu.RUnlock()
 
 	if handler == nil {
@@ -302,6 +320,10 @@ func (d *DataChannel) onMessage(msg DataChannelMessage) {
 
 func (d *DataChannel) handleOpen(dc *datachannel.DataChannel, isRemote, isAlreadyNegotiated bool) {
 	d.mu.Lock()
+	if d.isGracefulClosed {
+		d.mu.Unlock()
+		return
+	}
 	d.dataChannel = dc
 	bufferedAmountLowThreshold := d.bufferedAmountLowThreshold
 	onBufferedAmountLow := d.onBufferedAmountLow
@@ -326,7 +348,12 @@ func (d *DataChannel) handleOpen(dc *datachannel.DataChannel, isRemote, isAlread
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
+	if d.isGracefulClosed {
+		return
+	}
+
 	if !d.api.settingEngine.detach.DataChannels {
+		d.readLoopActive = make(chan struct{})
 		go d.readLoop()
 	}
 }
@@ -342,6 +369,10 @@ func (d *DataChannel) OnError(f func(err error)) {
 func (d *DataChannel) onError(err error) {
 	d.mu.RLock()
 	handler := d.onErrorHandler
+	if d.isGracefulClosed {
+		d.mu.RUnlock()
+		return
+	}
 	d.mu.RUnlock()
 
 	if handler != nil {
@@ -356,6 +387,12 @@ var rlBufPool = sync.Pool{New: func() interface{} {
 }}
 
 func (d *DataChannel) readLoop() {
+	defer func() {
+		d.mu.Lock()
+		readLoopActive := d.readLoopActive
+		d.mu.Unlock()
+		defer close(readLoopActive)
+	}()
 	for {
 		buffer := rlBufPool.Get().([]byte) //nolint:forcetypeassert
 		n, isString, err := d.dataChannel.ReadDataChannel(buffer)
@@ -438,7 +475,32 @@ func (d *DataChannel) Detach() (datachannel.ReadWriteCloser, error) {
 // Close Closes the DataChannel. It may be called regardless of whether
 // the DataChannel object was created by this peer or the remote peer.
 func (d *DataChannel) Close() error {
+	return d.close(false)
+}
+
+// GracefulClose Closes the DataChannel. It may be called regardless of whether
+// the DataChannel object was created by this peer or the remote peer. It also waits
+// for any goroutines it started to complete. This is only safe to call outside of
+// DataChannel callbacks or if in a callback, in its own goroutine.
+func (d *DataChannel) GracefulClose() error {
+	return d.close(true)
+}
+
+// Normally, close only stops writes from happening, so graceful=true
+// will wait for reads to be finished based on underlying SCTP association
+// closure or a SCTP reset stream from the other side. This is safe to call
+// with graceful=true after tearing down a PeerConnection but not
+// necessarily before. For example, if you used a vnet and dropped all packets
+// right before closing the DataChannel, you'd need never see a reset stream.
+func (d *DataChannel) close(shouldGracefullyClose bool) error {
 	d.mu.Lock()
+	d.isGracefulClosed = true
+	readLoopActive := d.readLoopActive
+	if shouldGracefullyClose && readLoopActive != nil {
+		defer func() {
+			<-readLoopActive
+		}()
+	}
 	haveSctpTransport := d.dataChannel != nil
 	d.mu.Unlock()
 

go.mod

@@ -5,7 +5,7 @@ go 1.17
 require (
 	github.com/pion/datachannel v1.5.8
 	github.com/pion/dtls/v2 v2.2.12
-	github.com/pion/ice/v2 v2.3.31
+	github.com/pion/ice/v2 v2.3.34
 	github.com/pion/interceptor v0.1.29
 	github.com/pion/logging v0.2.2
 	github.com/pion/randutil v0.1.0
@@ -15,7 +15,7 @@ require (
 	github.com/pion/sdp/v3 v3.0.9
 	github.com/pion/srtp/v2 v2.0.20
 	github.com/pion/stun v0.6.1
-	github.com/pion/transport/v2 v2.2.8
+	github.com/pion/transport/v2 v2.2.10
 	github.com/sclevine/agouti v3.0.0+incompatible
 	github.com/stretchr/testify v1.9.0
 	golang.org/x/net v0.22.0

go.sum

@@ -45,8 +45,8 @@ github.com/pion/datachannel v1.5.8/go.mod h1:PgmdpoaNBLX9HNzNClmdki4DYW5JtI7Yibu
 github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
 github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk=
 github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE=
-github.com/pion/ice/v2 v2.3.31 h1:qag/YqiOn5qPi0kgeVdsytxjx8szuriWSIeXKu8dDQc=
-github.com/pion/ice/v2 v2.3.31/go.mod h1:8fac0+qftclGy1tYd/nfwfHC729BLaxtVqMdMVCAVPU=
+github.com/pion/ice/v2 v2.3.34 h1:Ic1ppYCj4tUOcPAp76U6F3fVrlSw8A9JtRXLqw6BbUM=
+github.com/pion/ice/v2 v2.3.34/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ=
 github.com/pion/interceptor v0.1.29 h1:39fsnlP1U8gw2JzOFWdfCU82vHvhW9o0rZnZF56wF+M=
 github.com/pion/interceptor v0.1.29/go.mod h1:ri+LGNjRUc5xUNtDEPzfdkmSqISixVTBF/z/Zms/6T4=
 github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
@@ -74,8 +74,8 @@ github.com/pion/stun v0.6.1/go.mod h1:/hO7APkX4hZKu/D0f2lHzNyvdkTGtIy3NDmLR7kSz/
 github.com/pion/transport/v2 v2.2.1/go.mod h1:cXXWavvCnFF6McHTft3DWS9iic2Mftcz1Aq29pGcU5g=
 github.com/pion/transport/v2 v2.2.3/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0=
 github.com/pion/transport/v2 v2.2.4/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0=
-github.com/pion/transport/v2 v2.2.8 h1:HzsqGBChgtF4Cj47gu51l5hONuK/NwgbZL17CMSuwS0=
-github.com/pion/transport/v2 v2.2.8/go.mod h1:sq1kSLWs+cHW9E+2fJP95QudkzbK7wscs8yYgQToO5E=
+github.com/pion/transport/v2 v2.2.10 h1:ucLBLE8nuxiHfvkFKnkDQRYWYfp8ejf4YBOPfaQpw6Q=
+github.com/pion/transport/v2 v2.2.10/go.mod h1:sq1kSLWs+cHW9E+2fJP95QudkzbK7wscs8yYgQToO5E=
 github.com/pion/transport/v3 v3.0.1/go.mod h1:UY7kiITrlMv7/IKgd5eTUcaahZx5oUN3l9SzK5f5xE0=
 github.com/pion/transport/v3 v3.0.2 h1:r+40RJR25S9w3jbA6/5uEPTzcdn7ncyU44RWCbHkLg4=
 github.com/pion/transport/v3 v3.0.2/go.mod h1:nIToODoOlb5If2jF9y2Igfx3PFYWfuXi37m0IlWa/D0=

icegatherer.go

@@ -188,13 +188,31 @@ func (g *ICEGatherer) Gather() error {
 
 // Close prunes all local candidates, and closes the ports.
 func (g *ICEGatherer) Close() error {
+	return g.close(false /* shouldGracefullyClose */)
+}
+
+// GracefulClose prunes all local candidates, and closes the ports. It also waits
+// for any goroutines it started to complete. This is only safe to call outside of
+// ICEGatherer callbacks or if in a callback, in its own goroutine.
+func (g *ICEGatherer) GracefulClose() error {
+	return g.close(true /* shouldGracefullyClose */)
+}
+
+func (g *ICEGatherer) close(shouldGracefullyClose bool) error {
 	g.lock.Lock()
 	defer g.lock.Unlock()
 
 	if g.agent == nil {
 		return nil
-	} else if err := g.agent.Close(); err != nil {
-		return err
+	}
+	if shouldGracefullyClose {
+		if err := g.agent.GracefulClose(); err != nil {
+			return err
+		}
+	} else {
+		if err := g.agent.Close(); err != nil {
+			return err
+		}
 	}
 
 	g.agent = nil

icetransport.go

@@ -16,6 +16,7 @@ import (
 	"github.com/pion/ice/v2"
 	"github.com/pion/logging"
 	"github.com/viamrobotics/webrtc/v3/internal/mux"
+	"github.com/viamrobotics/webrtc/v3/internal/util"
 )
 
 // ICETransport allows an application access to information about the ICE
@@ -187,6 +188,17 @@ func (t *ICETransport) restart() error {
 
 // Stop irreversibly stops the ICETransport.
 func (t *ICETransport) Stop() error {
+	return t.stop(false /* shouldGracefullyClose */)
+}
+
+// GracefulStop irreversibly stops the ICETransport. It also waits
+// for any goroutines it started to complete. This is only safe to call outside of
+// ICETransport callbacks or if in a callback, in its own goroutine.
+func (t *ICETransport) GracefulStop() error {
+	return t.stop(true /* shouldGracefullyClose */)
+}
+
+func (t *ICETransport) stop(shouldGracefullyClose bool) error {
 	t.lock.Lock()
 	defer t.lock.Unlock()
 
@@ -197,8 +209,18 @@ func (t *ICETransport) Stop() error {
 	}
 
 	if t.mux != nil {
-		return t.mux.Close()
+		var closeErrs []error
+		if shouldGracefullyClose && t.gatherer != nil {
+			// we can't access icegatherer/icetransport.Close via
+			// mux's net.Conn Close so we call it earlier here.
+			closeErrs = append(closeErrs, t.gatherer.GracefulClose())
+		}
+		closeErrs = append(closeErrs, t.mux.Close())
+		return util.FlattenErrs(closeErrs)
 	} else if t.gatherer != nil {
+		if shouldGracefullyClose {
+			return t.gatherer.GracefulClose()
+		}
 		return t.gatherer.Close()
 	}
 	return nil

internal/mux/mux.go

@@ -120,6 +120,10 @@ func (m *Mux) readLoop() {
 		}
 
 		if err = m.dispatch(buf[:n]); err != nil {
+			if errors.Is(err, io.ErrClosedPipe) {
+				// if the buffer was closed, that's not an error we care to report
+				return
+			}
 			m.log.Errorf("mux: ending readLoop dispatch error %s", err.Error())
 			return
 		}