feat: Enhance SCTP data channel support with improved state management and testing; add new tests for out-of-order data handling and SACK chunk processing

Author: MarlonJD

README.md

@@ -257,8 +257,10 @@ INIT_ACK, COOKIE_ECHO, COOKIE_ACK, DATA, SACK, parameter padding, and CRC32C
 checksum validation, with a unit-tested OpenSSL DTLS application-data
 association bootstrap that can exchange SCTP DATA chunks, SACK responses, and
 reassemble fragmented SCTP DATA messages, with duplicate DATA TSN suppression
-and contiguous cumulative SACK tracking. The shared media/data binder and Room
-live-media startup helper can select that association transport in
+and contiguous cumulative SACK tracking with gap/duplicate reporting, while
+holding out-of-order DATA until contiguous TSNs can be delivered and tracking
+outbound DATA TSNs until peer SACK acknowledgement. The shared media/data binder
+and Room live-media startup helper can select that association transport in
 package-internal opt-in tests, and the LiveKit integration harness now validates
 a default-gated two-client reliable `DataPacket` publish/receive path over that
 standards-shaped SCTP association transport. The public default Room path

Sources/LiveKitNative/Core/ProductionReadiness.swift

@@ -32,7 +32,7 @@ public extension LiveKitNative {
             "Fresh join, resume reconnect, disconnect, and failed signal-loop boundaries regenerate local ICE credentials, clear stale peer negotiation state, parse TURN UDP/TCP/TLS endpoint configuration, order TURN relay fallback candidates as UDP, TCP, then TLS while identifying the currently executable UDP datagram path, select supported TURN UDP relay session configurations from parsed ICE server endpoints with credentials, exercise TURN Allocate, Refresh, CreatePermission, and ChannelBind request/authentication/response-validation primitives with one-shot stale nonce retry, cover TURN ChannelData framing, provide deterministic allocation/permission maintenance scheduling and execution, plan relayed ICE candidates from TURN bindings, compose and execute a bounded TURN relay session setup plan over abstract transports, expose a ChannelData relay transport over an abstract media datagram transport, allocate default UDP TURN relay candidates through the bound Room ICE socket, use ChannelData relay bindings for relayed ICE checks and media datagrams, provide deterministic ICE consent freshness planning plus an injected Room startup loop, run ICEAgent connectivity checks through paced scheduling with queued triggered-check priority plus role-conflict role switching and pair-priority recompute, and rebuild reconnect SyncState SDP/trickle state with fresh local ICE credentials, but TCP/TLS TURN transport execution, LiveKit TURN-only validation, and live media recovery are not complete.",
             "Publisher camera capture can encode H.264 through VideoToolbox with bounded frame backpressure/drop control and publisher microphone capture can encode Opus through AudioToolbox before RTP/SRTP send, while subscriber RTP can pass through jitter buffering, H.264/Opus packet assembly, NACK/PLI feedback, opt-in Opus decode-to-audio-playout scheduling, opt-in H.264 VideoToolbox decode-to-pixel-buffer scheduling, application-provided subscriber video renderer handoff, UIKit/AppKit VideoView sample-buffer rendering, and opt-in iOS voice-chat audio-session lifecycle configuration; real-device video display validation, LiveKit E2E media validation, route/interruption audio recovery, and production runtime pacing remain incomplete.",
             "Production H.264 now uses real VideoToolbox encode output for publish smoke coverage, opt-in VideoToolbox decode output for subscribe smoke coverage, and UIKit/AppKit VideoView rendering for decoded frames, but production readiness still requires real-device display validation, hardware acceleration verification where the OS exposes that signal, and explicit fallback behavior instead of relying on a pure Swift codec implementation.",
-            "SCTP data channel support includes packet-level DCEP/DataPacket planning, manager-assigned stream routing, queued local publish flushing, inbound DataPacket-to-RoomEvent plumbing, publisher SDP data-channel negotiation, subscriber-side data-channel receive-loop plumbing, OpenSSL DTLS application-data transport coverage, default Room shared-media DTLS/SRTP demux binding, deterministic packet fragmentation/reassembly envelopes, fragmented-packet retransmission scheduling on the DTLS-backed packet transport, opt-in standards-shaped SCTP association bootstrap with DATA/SACK exchange, fragmented DATA reassembly, duplicate DATA TSN suppression, contiguous cumulative SACK tracking, recovery reset that reopens LiveKit data channels after association restart, and default-gated two-client LiveKit DataPacket publish/receive smoke over that standards-shaped association transport, but full standards-compliant SCTP association state, congestion control, LiveKit-validated data-channel recovery, and E2E hardening are incomplete.",
+            "SCTP data channel support includes packet-level DCEP/DataPacket planning, manager-assigned stream routing, queued local publish flushing, inbound DataPacket-to-RoomEvent plumbing, publisher SDP data-channel negotiation, subscriber-side data-channel receive-loop plumbing, OpenSSL DTLS application-data transport coverage, default Room shared-media DTLS/SRTP demux binding, deterministic packet fragmentation/reassembly envelopes, fragmented-packet retransmission scheduling on the DTLS-backed packet transport, opt-in standards-shaped SCTP association bootstrap with DATA/SACK exchange, fragmented DATA reassembly, duplicate DATA TSN suppression, contiguous cumulative SACK tracking with gap/duplicate reporting, ordered contiguous TSN delivery, outbound DATA TSN acknowledgement tracking, recovery reset that reopens LiveKit data channels after association restart, and default-gated two-client LiveKit DataPacket publish/receive smoke over that standards-shaped association transport, but full standards-compliant SCTP association state, congestion control, LiveKit-validated data-channel recovery, and E2E hardening are incomplete.",
             "Signal reconnect resets stale peer negotiation state, restarts local ICE credentials, rebuilds retained subscriber answer and publisher offer SDP with fresh ICE credentials, sends SyncState for retained subscription/local media/data publication state, re-sends local ICE trickle/final-trickle when media startup is configured, preserves publisher offer track state for later publish/unpublish re-offers, clears stale local media/data publication state for server/SFU unpublish responses, and resets injected publisher data channels for post-reconnect DCEP reopen in unit tests, but live media recovery and LiveKit integration hardening are incomplete.",
             "General production video meetings require the new audio-session primitives to be validated with meeting-grade capture/playout, echo cancellation, route changes, Bluetooth behavior, interruptions, background/foreground handling, and timing on real iOS devices.",
             "A default subscriber RTP receive pipeline now integrates jitter buffering with bounded NACK/PLI feedback, scheduled RTCP Receiver Reports with DLSR timing from observed RTP/Sender Report state, REMB bitrate-feedback packet planning/sending, deterministic RTCP receiver-report bandwidth estimation, adaptive video quality recommendations, publisher RTCP receiver-report ingestion, H.264 encoder bitrate/FPS recommendation application, manual subscriber quality preset signaling, opt-in automatic subscriber adaptive track-settings dispatch, publisher active video-layer UpdateVideoLayers signaling, single-layer H.264 AddTrack high-quality metadata, and camera publish frame backpressure primitives, but general production video meetings still require full packet-loss recovery, TWCC/full REMB interop or equivalent congestion control, default-on LiveKit-validated subscriber adaptation policy, complete encoder control policy, actual multi-encoder simulcast/SVC media production, and weak-network E2E validation.",

Sources/LiveKitNative/LiveKitNative.docc/LiveKitNative.md

@@ -37,11 +37,13 @@ adds queued local data publish flushing through an injected SCTP packet
 transport, inbound data-channel `DataPacket` event plumbing, and OpenSSL DTLS
 application-data packet transport coverage with deterministic packet
 fragmentation/reassembly and fragmented-packet retransmission scheduling.
-Standards-shaped SCTP association coverage now suppresses duplicate DATA TSNs
-and tracks contiguous cumulative SACK state. Data channel recovery can reset
+Standards-shaped SCTP association coverage now suppresses duplicate DATA TSNs,
+tracks contiguous cumulative SACK state with gap/duplicate reporting, and holds
+out-of-order DATA until contiguous TSNs can be delivered while tracking outbound
+DATA TSNs until peer SACK acknowledgement. Data channel recovery can reset
 LiveKit channels after association restart and Room reconnect responses reset
-injected publisher data channels before post-reconnect publish. Separately
-gated LiveKit integration now validates a two-client reliable `DataPacket`
+injected publisher data channels before post-reconnect publish. Default-gated
+LiveKit integration now validates a two-client reliable `DataPacket`
 publish/receive smoke over the standards-shaped SCTP association transport. A
 shared WebRTC DTLS/SRTP datagram demux and media/data session binder can keep
 persistent OpenSSL DTLS application data and SRTP media on the same selected

Sources/LiveKitNativeWebRTC/SCTPDataChannel.swift

@@ -33,6 +33,7 @@ package enum SCTPDataChannelError: Error, Equatable, Sendable {
     case mismatchedFragmentMetadata(messageID: UInt32)
     case missingFragments(messageID: UInt32)
     case retransmissionAttemptsExhausted(messageID: UInt32, fragmentIndex: UInt16)
+    case invalidSCTPAssociationState(expected: String, actual: String)
     case invalidUTF8
     case duplicateStreamID(UInt16)
     case duplicateLabel(String)
@@ -1436,7 +1437,12 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
     private var peerVerificationTag: UInt32?
     private var nextTSN: UInt32
     private var expectedPeerTSN: UInt32?
+    private var nextPeerDeliveryTSN: UInt32?
     private var receivedPeerTSNs: Set<UInt32> = []
+    private var receivedPeerDataChunks: [UInt32: SCTPDataChunk] = [:]
+    private var pendingOutboundDataChunks: [UInt32: SCTPDataChunk] = [:]
+    private var mutableLastPeerCumulativeTSNAck: UInt32?
+    private var mutableLastPeerSACK: SCTPSACKChunk?
     private var nextStreamSequenceNumbers: [UInt16: UInt16] = [:]
     private var pendingReceivedPackets: [SCTPDataChannelPacket] = []
     private var receivePumpTask: Task<Void, Never>?
@@ -1458,6 +1464,22 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
         state == .established
     }
 
+    package var pendingReceivedPacketCount: Int {
+        pendingReceivedPackets.count
+    }
+
+    package var pendingOutboundTSNCount: Int {
+        pendingOutboundDataChunks.count
+    }
+
+    package var lastPeerCumulativeTSNAck: UInt32? {
+        mutableLastPeerCumulativeTSNAck
+    }
+
+    package var lastPeerSACK: SCTPSACKChunk? {
+        mutableLastPeerSACK
+    }
+
     package func startAssociation() async throws {
         try await ensureAssociation()
     }
@@ -1471,10 +1493,16 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
             streamSequenceNumber: streamSequenceNumber
         )
         nextTSN &+= UInt32(dataChunks.count)
-        try await sendPacket(
-            verificationTag: try requirePeerVerificationTag(),
-            chunks: dataChunks.map { .data($0) }
-        )
+        trackOutboundDataChunks(dataChunks)
+        do {
+            try await sendPacket(
+                verificationTag: try requirePeerVerificationTag(),
+                chunks: dataChunks.map { .data($0) }
+            )
+        } catch {
+            untrackOutboundDataChunks(dataChunks)
+            throw error
+        }
     }
 
     package func receive() async throws -> SCTPDataChannelPacket {
@@ -1541,7 +1569,7 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
             case .data:
                 try await acceptData(chunk, packet: packet)
             case .sack:
-                _ = try SCTPSACKChunk(chunk: chunk)
+                try acceptSACK(chunk)
             default:
                 continue
             }
@@ -1633,8 +1661,7 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
         try validateVerificationTag(packet.verificationTag, expected: 0)
         let initChunk = try SCTPInitChunk(chunk: chunk)
         peerVerificationTag = initChunk.initiateTag
-        expectedPeerTSN = initChunk.initialTSN
-        receivedPeerTSNs.removeAll()
+        resetPeerReceiveState(initialTSN: initChunk.initialTSN)
 
         let initAck = SCTPInitChunk(
             type: .initAck,
@@ -1657,11 +1684,14 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
     }
 
     private func acceptInitAck(_ chunk: SCTPChunk, packet: SCTPPacket) async throws {
+        try requireAssociationState(
+            allowed: [.initSent, .initAckSent],
+            expected: "INIT_ACK after local INIT"
+        )
         try validateVerificationTag(packet.verificationTag, expected: configuration.localInitiateTag)
         let initAck = try SCTPInitChunk(chunk: chunk)
         peerVerificationTag = initAck.initiateTag
-        expectedPeerTSN = initAck.initialTSN
-        receivedPeerTSNs.removeAll()
+        resetPeerReceiveState(initialTSN: initAck.initialTSN)
         guard let stateCookie = initAck.parameters.first(where: { $0.type == SCTPParameterType.stateCookie })?.value else {
             throw SCTPDataChannelError.missingSCTPStateCookie
         }
@@ -1674,6 +1704,10 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
     }
 
     private func acceptCookieEcho(packet: SCTPPacket) async throws {
+        try requireAssociationState(
+            allowed: [.initAckSent, .cookieEchoSent, .established],
+            expected: "COOKIE_ECHO after local INIT_ACK"
+        )
         try validateVerificationTag(packet.verificationTag, expected: configuration.localInitiateTag)
         try await sendPacket(
             verificationTag: try requirePeerVerificationTag(),
@@ -1683,31 +1717,109 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
     }
 
     private func acceptCookieAck(packet: SCTPPacket) throws {
+        try requireAssociationState(
+            allowed: [.cookieEchoSent, .established],
+            expected: "COOKIE_ACK after local COOKIE_ECHO"
+        )
         try validateVerificationTag(packet.verificationTag, expected: configuration.localInitiateTag)
         state = .established
     }
 
+    private func acceptSACK(_ chunk: SCTPChunk) throws {
+        try requireAssociationState(
+            allowed: [.established],
+            expected: "SACK after association establishment"
+        )
+        let sack = try SCTPSACKChunk(chunk: chunk)
+        mutableLastPeerSACK = sack
+        mutableLastPeerCumulativeTSNAck = sack.cumulativeTSNAck
+        acknowledgeOutboundChunks(with: sack)
+    }
+
     private func acceptData(_ chunk: SCTPChunk, packet: SCTPPacket) async throws {
+        try requireAssociationState(
+            allowed: [.established],
+            expected: "DATA after association establishment"
+        )
         try validateVerificationTag(packet.verificationTag, expected: configuration.localInitiateTag)
         let dataChunk = try SCTPDataChunk(chunk: chunk)
         let isNewTSN = markPeerTSNReceived(dataChunk.tsn)
         let cumulativeTSNAck = expectedPeerTSN.map { $0 &- 1 } ?? dataChunk.tsn
+        if isNewTSN {
+            receivedPeerDataChunks[dataChunk.tsn] = dataChunk
+        }
 
         try await sendPacket(
             verificationTag: try requirePeerVerificationTag(),
             chunks: [
                 .sack(SCTPSACKChunk(
                     cumulativeTSNAck: cumulativeTSNAck,
-                    advertisedReceiverWindowCredit: configuration.advertisedReceiverWindowCredit
+                    advertisedReceiverWindowCredit: configuration.advertisedReceiverWindowCredit,
+                    gapAckBlocks: sackGapAckBlocks(cumulativeTSNAck: cumulativeTSNAck),
+                    duplicateTSNs: isNewTSN ? [] : [dataChunk.tsn]
                 )),
             ]
         )
         guard isNewTSN else {
             return
         }
 
-        if let packet = try fragmentReassembler.append(dataChunk) {
-            pendingReceivedPackets.append(packet)
+        try drainContiguousPeerDataChunks()
+    }
+
+    private func resetPeerReceiveState(initialTSN: UInt32) {
+        expectedPeerTSN = initialTSN
+        nextPeerDeliveryTSN = initialTSN
+        receivedPeerTSNs.removeAll()
+        receivedPeerDataChunks.removeAll()
+        fragmentReassembler = SCTPDataChunkFragmentReassembler()
+    }
+
+    private func trackOutboundDataChunks(_ chunks: [SCTPDataChunk]) {
+        for chunk in chunks {
+            pendingOutboundDataChunks[chunk.tsn] = chunk
+        }
+    }
+
+    private func untrackOutboundDataChunks(_ chunks: [SCTPDataChunk]) {
+        for chunk in chunks {
+            pendingOutboundDataChunks.removeValue(forKey: chunk.tsn)
+        }
+    }
+
+    private func acknowledgeOutboundChunks(with sack: SCTPSACKChunk) {
+        let cumulativelyAcknowledgedTSNs = pendingOutboundDataChunks.keys.filter {
+            tsnIsAcknowledged($0, by: sack.cumulativeTSNAck)
+        }
+        for tsn in cumulativelyAcknowledgedTSNs {
+            pendingOutboundDataChunks.removeValue(forKey: tsn)
+        }
+
+        for block in sack.gapAckBlocks {
+            var offset = block.start
+            while offset <= block.end {
+                pendingOutboundDataChunks.removeValue(forKey: sack.cumulativeTSNAck &+ UInt32(offset))
+                if offset == UInt16.max {
+                    break
+                }
+                offset += 1
+            }
+        }
+    }
+
+    private func tsnIsAcknowledged(_ tsn: UInt32, by cumulativeTSNAck: UInt32) -> Bool {
+        tsn == cumulativeTSNAck || Int32(bitPattern: cumulativeTSNAck &- tsn) > 0
+    }
+
+    private func requireAssociationState(
+        allowed: [AssociationState],
+        expected: String
+    ) throws {
+        guard allowed.contains(state) else {
+            throw SCTPDataChannelError.invalidSCTPAssociationState(
+                expected: expected,
+                actual: String(describing: state)
+            )
         }
     }
 
@@ -1732,6 +1844,49 @@ package actor DTLSSCTPAssociationDataChannelPacketTransport: SCTPDataChannelPack
         expectedPeerTSN = nextTSN
     }
 
+    private func sackGapAckBlocks(cumulativeTSNAck: UInt32) -> [SCTPSACKGapAckBlock] {
+        let offsets = receivedPeerTSNs.compactMap { tsn -> UInt16? in
+            let offset = tsn &- cumulativeTSNAck
+            guard offset > 0, offset <= UInt32(UInt16.max) else {
+                return nil
+            }
+            return UInt16(offset)
+        }.sorted()
+
+        guard !offsets.isEmpty else {
+            return []
+        }
+
+        var blocks: [SCTPSACKGapAckBlock] = []
+        var start = offsets[0]
+        var end = offsets[0]
+        for offset in offsets.dropFirst() {
+            guard offset == end &+ 1 else {
+                blocks.append(SCTPSACKGapAckBlock(start: start, end: end))
+                start = offset
+                end = offset
+                continue
+            }
+            end = offset
+        }
+        blocks.append(SCTPSACKGapAckBlock(start: start, end: end))
+        return blocks
+    }
+
+    private func drainContiguousPeerDataChunks() throws {
+        guard var nextTSN = nextPeerDeliveryTSN else {
+            return
+        }
+
+        while let chunk = receivedPeerDataChunks.removeValue(forKey: nextTSN) {
+            if let packet = try fragmentReassembler.append(chunk) {
+                pendingReceivedPackets.append(packet)
+            }
+            nextTSN &+= 1
+        }
+        nextPeerDeliveryTSN = nextTSN
+    }
+
     private func dataChunks(
         for packet: SCTPDataChannelPacket,
         firstTSN: UInt32,