From 5e3e3cbb7a5367fb24975a3e713039a0f2174bfa Mon Sep 17 00:00:00 2001
From: geenkel <geenkel.cosslara@audinate.com>
Date: Mon, 22 Dec 2025 11:27:58 -0600
Subject: [PATCH] Add Stream buffer limit to WriteSCTP configurable via
 environment variable

---
 association.go | 10 ++++++++++
 stream.go      | 11 +++++++++++
 2 files changed, 21 insertions(+)

diff --git a/association.go b/association.go
index e47b3fee..8d21a746 100644
--- a/association.go
+++ b/association.go
@@ -11,6 +11,7 @@ import (
 	"io"
 	"math"
 	"net"
+	"os"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -1620,6 +1621,14 @@ func (a *Association) AcceptStream() (*Stream, error) {
 
 // createStream creates a stream. The caller should hold the lock and check no stream exists for this id.
 func (a *Association) createStream(streamIdentifier uint16, accept bool) *Stream {
+	var bufferLimit uint64
+	if limitStr := os.Getenv("SCTP_STREAM_BUFFER_LIMIT"); limitStr != "" {
+		_, err := fmt.Sscanf(limitStr, "%d", &bufferLimit)
+		if err != nil || bufferLimit == 0 {
+			bufferLimit = 0
+		}
+	}
+
 	stream := &Stream{
 		association:      a,
 		streamIdentifier: streamIdentifier,
@@ -1627,6 +1636,7 @@ func (a *Association) createStream(streamIdentifier uint16, accept bool) *Stream
 		log:              a.log,
 		name:             fmt.Sprintf("%d:%s", streamIdentifier, a.name),
 		writeDeadline:    deadline.New(),
+		bufferLimit:      bufferLimit,
 	}
 
 	stream.readNotifier = sync.NewCond(&stream.lock)
diff --git a/stream.go b/stream.go
index 3a05f637..3b8cb0ae 100644
--- a/stream.go
+++ b/stream.go
@@ -78,6 +78,7 @@ type Stream struct {
 	state               StreamState
 	log                 logging.LeveledLogger
 	name                string
+	bufferLimit         uint64
 }
 
 // StreamIdentifier returns the Stream identifier associated to the stream.
@@ -269,6 +270,16 @@ func (s *Stream) Write(payload []byte) (n int, err error) {
 
 // WriteSCTP writes len(payload) bytes from payload to the DTLS connection.
 func (s *Stream) WriteSCTP(payload []byte, ppi PayloadProtocolIdentifier) (int, error) {
+	// Check buffer limit.
+	if s.bufferLimit > 0 {
+		s.lock.RLock()
+		currentBuffered := s.bufferedAmount
+		s.lock.RUnlock()
+		if currentBuffered+uint64(len(payload)) > s.bufferLimit {
+			return 0, fmt.Errorf("stream buffer limit exceeded: %d bytes", s.bufferLimit)
+		}
+	}
+
 	maxMessageSize := s.association.MaxMessageSize()
 	if len(payload) > int(maxMessageSize) {
 		return 0, fmt.Errorf("%w: %v", ErrOutboundPacketTooLarge, maxMessageSize)