Support DataChannel messages larger then MaxUint16

SCTP now internally can handle larger messages

Resolves #2712

Author: Sean-Der

constants.go

@@ -3,7 +3,11 @@
 
 package webrtc
 
-import "github.com/pion/dtls/v3"
+import (
+	"math"
+
+	"github.com/pion/dtls/v3"
+)
 
 const (
 	// default as the standard ethernet MTU
@@ -19,6 +23,15 @@ const (
 	// If the total amount of incoming SSRCes exceeds this new requests will be ignored.
 	simulcastMaxProbeRoutines = 25
 
+	// Default Max SCTP Message Size is the largest single DataChannel
+	// message we can send or accept. This default was chosen to match FireFox.
+	defaultMaxSCTPMessageSize = 1073741823
+
+	// If a DataChannel Max Message Size isn't declared by the Remote(max-message-size)
+	// this is the value we default to. This value was chosen because it was the behavior
+	// of Pion before max-message-size was implemented.
+	sctpMaxMessageSizeUnsetValue = math.MaxUint16
+
 	mediaSectionApplication = "application"
 
 	sdpAttributeRid = "rid"

datachannel.go

@@ -10,7 +10,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"math"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -20,7 +19,6 @@ import (
 	"github.com/pion/webrtc/v4/pkg/rtcerr"
 )
 
-const dataChannelBufferSize = math.MaxUint16 // message size limit for Chromium
 var errSCTPNotEstablished = errors.New("SCTP not established")
 
 // DataChannel represents a WebRTC DataChannel
@@ -404,10 +402,24 @@ func (d *DataChannel) readLoop() {
 		d.mu.Unlock()
 		defer close(readLoopActive)
 	}()
-	buffer := make([]byte, dataChannelBufferSize)
+
+	buffer := make([]byte, sctpMaxMessageSizeUnsetValue)
 	for {
 		n, isString, err := d.dataChannel.ReadDataChannel(buffer)
 		if err != nil {
+			if errors.Is(err, io.ErrShortBuffer) {
+				if int64(n) < int64(d.api.settingEngine.getSCTPMaxMessageSize()) {
+					buffer = append(buffer, make([]byte, len(buffer))...) // nolint
+
+					continue
+				}
+
+				d.log.Errorf(
+					"Incoming DataChannel message larger then Max Message size %v",
+					d.api.settingEngine.getSCTPMaxMessageSize(),
+				)
+			}
+
 			d.setReadyState(DataChannelStateClosed)
 			if !errors.Is(err, io.EOF) {
 				d.onError(err)
@@ -417,11 +429,10 @@ func (d *DataChannel) readLoop() {
 			return
 		}
 
-		msg := DataChannelMessage{Data: make([]byte, n), IsString: isString}
-		copy(msg.Data, buffer[:n])
-
-		// NB: Why was DataChannelMessage not passed as a pointer value?
-		d.onMessage(msg) // nolint:staticcheck
+		d.onMessage(DataChannelMessage{
+			Data:     append([]byte{}, buffer[:n]...),
+			IsString: isString,
+		})
 	}
 }
 

datachannel_go_test.go

@@ -8,6 +8,7 @@ package webrtc
 
 import (
 	"bytes"
+	"context"
 	"crypto/rand"
 	"encoding/binary"
 	"io"
@@ -851,3 +852,38 @@ func TestDataChannel_DetachErrors(t *testing.T) {
 		assert.NoError(t, answer.Close())
 	})
 }
+
+func TestDataChannelMessageSize(t *testing.T) {
+	offerPC, answerPC, err := newPair()
+	assert.NoError(t, err)
+
+	dc, err := offerPC.CreateDataChannel("", nil)
+	assert.NoError(t, err)
+
+	answerDataChannelMessages := make(chan []byte)
+	answerPC.OnDataChannel(func(d *DataChannel) {
+		d.OnMessage(func(m DataChannelMessage) {
+			answerDataChannelMessages <- m.Data
+		})
+	})
+
+	assert.NoError(t, signalPair(offerPC, answerPC))
+
+	messagesSent, messagesSentCancel := context.WithCancel(context.Background())
+	dc.OnOpen(func() {
+		for i := 0; i <= 10; i++ {
+			outboundMessage := make([]byte, sctpMaxMessageSizeUnsetValue*i)
+			_, err := rand.Read(outboundMessage)
+			assert.NoError(t, err)
+
+			assert.NoError(t, dc.Send(outboundMessage))
+			inboundMessage := <-answerDataChannelMessages
+
+			assert.Equal(t, outboundMessage, inboundMessage)
+		}
+		messagesSentCancel()
+	})
+
+	<-messagesSent.Done()
+	closePairNow(t, offerPC, answerPC)
+}

datachannel_test.go

@@ -510,11 +510,7 @@ func TestDataChannelParameters(t *testing.T) { //nolint:cyclop
 		})
 
 		go func() {
-			for {
-				if seenAnswerMessage.get() && seenOfferMessage.get() {
-					break
-				}
-
+			for seenAnswerMessage.get() && seenOfferMessage.get() {
 				if offerDatachannel.ReadyState() == DataChannelStateOpen {
 					assert.NoError(t, offerDatachannel.SendText(expectedMessage))
 				}

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/pion/randutil v0.1.0
 	github.com/pion/rtcp v1.2.15
 	github.com/pion/rtp v1.8.12
-	github.com/pion/sctp v1.8.36
+	github.com/pion/sctp v1.8.37
 	github.com/pion/sdp/v3 v3.0.10
 	github.com/pion/srtp/v3 v3.0.4
 	github.com/pion/stun/v3 v3.0.0

go.sum

@@ -53,8 +53,8 @@ github.com/pion/rtcp v1.2.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo=
 github.com/pion/rtcp v1.2.15/go.mod h1:jlGuAjHMEXwMUHK78RgX0UmEJFV4zUKOFHR7OP+D3D0=
 github.com/pion/rtp v1.8.12 h1:nsKs8Wi0jQyBFHU3qmn/OvtZrhktVfJY0vRxwACsL5U=
 github.com/pion/rtp v1.8.12/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4=
-github.com/pion/sctp v1.8.36 h1:owNudmnz1xmhfYje5L/FCav3V9wpPRePHle3Zi+P+M0=
-github.com/pion/sctp v1.8.36/go.mod h1:cNiLdchXra8fHQwmIoqw0MbLLMs+f7uQ+dGMG2gWebE=
+github.com/pion/sctp v1.8.37 h1:ZDmGPtRPX9mKCiVXtMbTWybFw3z/hVKAZgU81wcOrqs=
+github.com/pion/sctp v1.8.37/go.mod h1:cNiLdchXra8fHQwmIoqw0MbLLMs+f7uQ+dGMG2gWebE=
 github.com/pion/sdp/v3 v3.0.10 h1:6MChLE/1xYB+CjumMw+gZ9ufp2DPApuVSnDT8t5MIgA=
 github.com/pion/sdp/v3 v3.0.10/go.mod h1:88GMahN5xnScv1hIMTqLdu/cOcUkj6a9ytbncwMCq2E=
 github.com/pion/srtp/v3 v3.0.4 h1:2Z6vDVxzrX3UHEgrUyIGM4rRouoC7v+NiF1IHtp9B5M=

peerconnection.go

@@ -1563,10 +1563,10 @@ func (pc *PeerConnection) startRTPSenders(currentTransceivers []*RTPTransceiver)
 }
 
 // Start SCTP subsystem.
-func (pc *PeerConnection) startSCTP() {
+func (pc *PeerConnection) startSCTP(maxMessageSize uint32) {
 	// Start sctp
 	if err := pc.sctpTransport.Start(SCTPCapabilities{
-		MaxMessageSize: 0,
+		MaxMessageSize: maxMessageSize,
 	}); err != nil {
 		pc.log.Warnf("Failed to start SCTP: %s", err)
 		if err = pc.sctpTransport.Stop(); err != nil {
@@ -2625,8 +2625,8 @@ func (pc *PeerConnection) startRTP(
 	}
 
 	pc.startRTPReceivers(remoteDesc, currentTransceivers)
-	if haveApplicationMediaSection(remoteDesc.parsed) {
-		pc.startSCTP()
+	if d := haveDataChannel(remoteDesc); d != nil {
+		pc.startSCTP(getMaxMessageSize(d))
 	}
 }
 
@@ -2718,6 +2718,7 @@ func (pc *PeerConnection) generateUnmatchedSDP(
 		mediaSections,
 		pc.ICEGatheringState(),
 		nil,
+		pc.api.settingEngine.getSCTPMaxMessageSize(),
 	)
 }
 
@@ -2884,6 +2885,7 @@ func (pc *PeerConnection) generateMatchedSDP(
 		mediaSections,
 		pc.ICEGatheringState(),
 		bundleGroup,
+		pc.api.settingEngine.getSCTPMaxMessageSize(),
 	)
 }
 

sctptransport.go

@@ -9,7 +9,6 @@ package webrtc
 import (
 	"errors"
 	"io"
-	"math"
 	"sync"
 	"time"
 
@@ -34,10 +33,6 @@ type SCTPTransport struct {
 	// so we need a dedicated field
 	isStarted bool
 
-	// MaxMessageSize represents the maximum size of data that can be passed to
-	// DataChannel's send() method.
-	maxMessageSize float64
-
 	// MaxChannels represents the maximum amount of DataChannel's that can
 	// be used simultaneously.
 	maxChannels *uint16
@@ -74,7 +69,6 @@ func (api *API) NewSCTPTransport(dtls *DTLSTransport) *SCTPTransport {
 		dataChannelIDsUsed: make(map[uint16]struct{}),
 	}
 
-	res.updateMessageSize()
 	res.updateMaxChannels()
 
 	return res
@@ -90,20 +84,30 @@ func (r *SCTPTransport) Transport() *DTLSTransport {
 
 // GetCapabilities returns the SCTPCapabilities of the SCTPTransport.
 func (r *SCTPTransport) GetCapabilities() SCTPCapabilities {
+	var maxMessageSize uint32
+	if a := r.association(); a != nil {
+		maxMessageSize = a.MaxMessageSize()
+	}
+
 	return SCTPCapabilities{
-		MaxMessageSize: 0,
+		MaxMessageSize: maxMessageSize,
 	}
 }
 
 // Start the SCTPTransport. Since both local and remote parties must mutually
 // create an SCTPTransport, SCTP SO (Simultaneous Open) is used to establish
 // a connection over SCTP.
-func (r *SCTPTransport) Start(_ SCTPCapabilities) error {
+func (r *SCTPTransport) Start(capabilities SCTPCapabilities) error {
 	if r.isStarted {
 		return nil
 	}
 	r.isStarted = true
 
+	maxMessageSize := capabilities.MaxMessageSize
+	if maxMessageSize == 0 {
+		maxMessageSize = sctpMaxMessageSizeUnsetValue
+	}
+
 	dtlsTransport := r.Transport()
 	if dtlsTransport == nil || dtlsTransport.conn == nil {
 		return errSCTPTransportDTLS
@@ -115,6 +119,7 @@ func (r *SCTPTransport) Start(_ SCTPCapabilities) error {
 		LoggerFactory:        r.api.settingEngine.LoggerFactory,
 		RTOMax:               float64(r.api.settingEngine.sctp.rtoMax) / float64(time.Millisecond),
 		BlockWrite:           r.api.settingEngine.detach.DataChannels && r.api.settingEngine.dataChannelBlockWrite,
+		MaxMessageSize:       maxMessageSize,
 	})
 	if err != nil {
 		return err
@@ -344,36 +349,6 @@ func (r *SCTPTransport) onDataChannel(dc *DataChannel) (done chan struct{}) {
 	return
 }
 
-func (r *SCTPTransport) updateMessageSize() {
-	r.lock.Lock()
-	defer r.lock.Unlock()
-
-	var remoteMaxMessageSize float64 = 65536 // pion/webrtc#758
-	var canSendSize float64 = 65536          // pion/webrtc#758
-
-	r.maxMessageSize = r.calcMessageSize(remoteMaxMessageSize, canSendSize)
-}
-
-func (r *SCTPTransport) calcMessageSize(remoteMaxMessageSize, canSendSize float64) float64 {
-	switch {
-	case remoteMaxMessageSize == 0 &&
-		canSendSize == 0:
-		return math.Inf(1)
-
-	case remoteMaxMessageSize == 0:
-		return canSendSize
-
-	case canSendSize == 0:
-		return remoteMaxMessageSize
-
-	case canSendSize > remoteMaxMessageSize:
-		return remoteMaxMessageSize
-
-	default:
-		return canSendSize
-	}
-}
-
 func (r *SCTPTransport) updateMaxChannels() {
 	val := sctpMaxChannels
 	r.maxChannels = &val