Skip to content

Always announce video layers on SetPublisher #1765

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
PratimMallick wants to merge 2 commits into
base: develop
Choose a base branch
from
+152 −22
Open

Always announce video layers on SetPublisher #1765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ae06ad8
fix(core): always announce video layers on SetPublisher
PratimMallick Aug 11, 2026
016d192
fix(core): clear isIceRestarting on SetPublisher error paths
PratimMallick Aug 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 34 additions & 21 deletions ...android-core/src/main/kotlin/io/getstream/video/android/core/call/connection/Publisher.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,8 +195,8 @@ internal class Publisher(
logger.i { "Negotiating with tracks: $trackInfos" }
logger.i { "Offer: ${offer.description}" }

safeCall {
isIceRestarting = iceRestart
isIceRestarting = iceRestart
try {
setLocalDescription(offer).onErrorSuspend {
tracer.trace("negotiate-error-setlocaldescription", it.message ?: "unknown")
}
Expand All @@ -206,25 +206,36 @@ internal class Publisher(
session_id = sessionId,
)
val response = sfuClient.setPublisher(request)
logger.i { "Received answer: ${response.sdp}" }
if (response.error != null) {
logger.e {
"SetPublisherRequest Received error: ${response.error}, SetPublisherRequest: $request"
}
tracer.trace("negotiate-error-setpublisher", response.error.message ?: "unknown")
logger.e { "rejoin cause error in sfuClient.setPublisher, message:${response.error.message}" }

when (response.error.code) {
/**
* We are getting this error right away after joining the call first time
* Full error: 16:04:05.032 Call:PeerC...:publisher E (DefaultDispatcher-worker-17:526) SetPublisherRequest Received error: Error{code=ERROR_CODE_REQUEST_VALIDATION_FAILED, message=Invalid SetPublisher request, should_retry=false}
* This will cause emission of ParticipantLeftEvent to other person
* Historically common right after first join when video layers were omitted
* for non-LIVE tracks. Layers are now always computed from publish options, so
* remaining validation failures are treated as a real publisher/SFU mismatch.
*/
ErrorCode.ERROR_CODE_REQUEST_VALIDATION_FAILED -> rejoin()
ErrorCode.ERROR_CODE_REQUEST_VALIDATION_FAILED -> {
logger.e {
"rejoin cause error in sfuClient.setPublisher, " +
"message:${response.error.message}"
}
rejoin()
}

else -> {}
else -> {
logger.e {
"Unhandled SetPublisher error code=${response.error.code}, " +
"message=${response.error.message}"
}
}
}
return@submit
Comment on lines 209 to +235

@coderabbitai coderabbitai Bot Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Reset isIceRestarting before returning on an SFU error.

When iceRestart is true, Line 235 exits the submit lambda before Line 249 resets isIceRestarting. Every later negotiation then exits at Lines 174-177 as if an ICE restart is still active.

Use try/finally after setting the flag so every error path clears it.

Proposed fix 
             isIceRestarting = iceRestart
-            setLocalDescription(offer).onErrorSuspend {
-                tracer.trace("negotiate-error-setlocaldescription", it.message ?: "unknown")
-            }
-            val request = SetPublisherRequest(
-                sdp = offer.description,
-                tracks = trackInfos,
-                session_id = sessionId,
-            )
-            val response = sfuClient.setPublisher(request)
-            if (response.error != null) {
-                // existing error handling
-                return@submit
-            }
-
-            logger.i { "Received answer: ${response.sdp}" }
-            setRemoteDescription(SessionDescription(SessionDescription.Type.ANSWER, response.sdp))
+            try {
+                setLocalDescription(offer).onErrorSuspend {
+                    tracer.trace("negotiate-error-setlocaldescription", it.message ?: "unknown")
+                }
+                val request = SetPublisherRequest(
+                    sdp = offer.description,
+                    tracks = trackInfos,
+                    session_id = sessionId,
+                )
+                val response = sfuClient.setPublisher(request)
+                if (response.error != null) {
+                    // existing error handling
+                    return@submit
+                }
+
+                logger.i { "Received answer: ${response.sdp}" }
+                setRemoteDescription(SessionDescription(SessionDescription.Type.ANSWER, response.sdp))
+            } finally {
+                isIceRestarting = false
+            }
-        isIceRestarting = false
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/connection/Publisher.kt`
around lines 209 - 235, Wrap the negotiation flow in the relevant Publisher
method with try/finally immediately after setting isIceRestarting, and reset
isIceRestarting in the finally block. Ensure the SFU error return inside the
submit lambda, including the response.error handling around SetPublisherRequest,
still clears the flag before subsequent negotiations.

Comment thread
PratimMallick marked this conversation as resolved.
}

logger.i { "Received answer: ${response.sdp}" }
setRemoteDescription(SessionDescription(SessionDescription.Type.ANSWER, response.sdp))
.onErrorSuspend {
tracer.trace(
Expand All @@ -234,9 +245,12 @@ internal class Publisher(
}.onSuccess {
logger.d { "Publisher negotiation successfully done ✅" }
}
// Set ice trickle
} catch (e: Exception) {
logger.e(e) { "[negotiate] Exception occurred: ${e.message}" }
} finally {
// Must clear even when returning early from @submit (inline safeCall used to skip this).
isIceRestarting = false
}
isIceRestarting = false
}

override suspend fun stats(): ComputedStats? = safeCallWithDefault(null) {
Expand Down Expand Up @@ -649,16 +663,15 @@ internal class Publisher(
}
val isTrackLive = track.state() == MediaStreamTrack.State.LIVE
val isAudio = isAudioTrackType(publishOption.track_type)
// Layer math only needs dimension + PublishOption — not LIVE/frames. Previously we skipped
// computeLayers when !LIVE, announced empty layers, and SFU rejected SetPublisher on first
// join (then we rejoined). Always compute so muted/non-LIVE video still has layers.
val layers = if (!isAudio) {
if (isTrackLive) {
computeLayers(
captureFormat,
track,
publishOption,
)
} else {
transceiverCache.getLayers(publishOption)
}
computeLayers(
captureFormat,
track,
publishOption,
) ?: transceiverCache.getLayers(publishOption)
} else {
null
}
Expand Down
119 changes: 118 additions & 1 deletion ...oid-core/src/test/kotlin/io/getstream/video/android/core/call/connection/PublisherTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,10 @@ import stream.video.sfu.event.VideoSender
import stream.video.sfu.models.AudioBitrateProfile
import stream.video.sfu.models.Codec
import stream.video.sfu.models.DegradationPreference
import stream.video.sfu.models.Error
import stream.video.sfu.models.ErrorCode
import stream.video.sfu.models.PublishOption
import stream.video.sfu.models.TrackInfo
import stream.video.sfu.models.TrackType
import stream.video.sfu.models.VideoDimension
import stream.video.sfu.signal.SetPublisherResponse
Expand Down Expand Up @@ -92,6 +95,7 @@ class PublisherTest {
private lateinit var publisher: Publisher
private val coroutineContext = UnconfinedTestDispatcher()
private val testScope = TestScope(coroutineContext)
private var rejoinInvocations = 0

//region Example PublishOptions
private val videoPublishOption = PublishOption(
Expand Down Expand Up @@ -129,6 +133,7 @@ class PublisherTest {
@Before
fun setUp() {
MockKAnnotations.init(this, relaxUnitFun = true)
rejoinInvocations = 0

// Mock the mediaManager and peerConnectionFactory so they return mock Audio/Video tracks.
every { mockPeerConnectionFactory.makeAudioTrack(any(), any()) } answers {
Expand Down Expand Up @@ -171,7 +176,7 @@ class PublisherTest {
maxBitRate = 1_500_000,
sfuClient = mockSignalServerService,
sessionId = "session-id",
rejoin = { },
rejoin = { rejoinInvocations++ },
tracer = mockk(relaxed = true),
fastReconnect = {},
transceiverCache = mockTransceiverCache,
Expand Down Expand Up @@ -285,6 +290,118 @@ class PublisherTest {
coVerify(exactly = 0) { mockSignalServerService.setPublisher(any()) }
}

@Test
fun `getAnnouncedTracks includes video layers even when track is not LIVE`() {
val mockVideoTrack = mockk<VideoTrack>(relaxed = true) {
every { id() } returns "video-1"
every { kind() } returns "video"
every { isDisposed } returns false
every { state() } returns MediaStreamTrack.State.ENDED
}
val mockSender = mockk<RtpSender>(relaxed = true) {
every { track() } returns mockVideoTrack
}
val mockTransceiver = mockk<RtpTransceiver>(relaxed = true) {
every { sender } returns mockSender
every { mid } returns "0"
}
every { mockTransceiverCache.items() } returns listOf(
TransceiverId(videoPublishOption, mockTransceiver),
)
every { mockTransceiverCache.indexOf(videoPublishOption) } returns 0
every { mockTransceiverCache.getLayers(videoPublishOption) } returns null

val announced = publisher.getAnnouncedTracks(null, null)

assertEquals(1, announced.size)
assertTrue(announced[0].muted)
assertTrue(announced[0].layers.isNotEmpty())
}

@Test
fun `SetPublisher validation failure rejoins`() = runTest(coroutineContext) {
every { publisher.getAnnouncedTracks(any(), any()) } returns listOf(
TrackInfo(
track_id = "video-1",
track_type = TrackType.TRACK_TYPE_VIDEO,
mid = "0",
muted = false,
layers = listOf(
stream.video.sfu.models.VideoLayer(
rid = "f",
video_dimension = VideoDimension(1280, 720),
bitrate = 1_000_000,
fps = 30,
),
),
publish_option_id = videoPublishOption.id,
),
)
coEvery { publisher.setLocalDescription(any()) } returns Result.Success(Unit)
coEvery { publisher.setRemoteDescription(any()) } returns Result.Success(Unit)
coEvery { mockSignalServerService.setPublisher(any()) } returns SetPublisherResponse(
sdp = "",
error = Error(
code = ErrorCode.ERROR_CODE_REQUEST_VALIDATION_FAILED,
message = "Invalid SetPublisher request",
should_retry = false,
),
)

publisher.negotiate(source = "test")

coVerify(exactly = 1) { mockSignalServerService.setPublisher(any()) }
coVerify(exactly = 0) { publisher.setRemoteDescription(any()) }
assertEquals(1, rejoinInvocations)
}
Comment thread
PratimMallick marked this conversation as resolved.

@Test
fun `iceRestart SetPublisher error clears isIceRestarting so later negotiate runs`() = runTest(
coroutineContext,
) {
every { publisher.getAnnouncedTracks(any(), any()) } returns listOf(
TrackInfo(
track_id = "video-1",
track_type = TrackType.TRACK_TYPE_VIDEO,
mid = "0",
muted = false,
layers = listOf(
stream.video.sfu.models.VideoLayer(
rid = "f",
video_dimension = VideoDimension(1280, 720),
bitrate = 1_000_000,
fps = 30,
),
),
publish_option_id = videoPublishOption.id,
),
)
coEvery { publisher.setLocalDescription(any()) } returns Result.Success(Unit)
coEvery { publisher.setRemoteDescription(any()) } returns Result.Success(Unit)
coEvery { mockSignalServerService.setPublisher(any()) } returnsMany listOf(
SetPublisherResponse(
sdp = "",
error = Error(
code = ErrorCode.ERROR_CODE_PARTICIPANT_NOT_FOUND,
message = "participant not found",
should_retry = false,
),
),
SetPublisherResponse(
sdp = fakeSdpAnswer.description,
error = null,
),
)

publisher.negotiate(source = "ice-restart", iceRestart = true)
assertEquals(0, rejoinInvocations)

// Without finally clearing isIceRestarting, this second call would no-op.
publisher.negotiate(source = "after-ice-restart-error")

coVerify(exactly = 2) { mockSignalServerService.setPublisher(any()) }
}

@Test
fun `close with stopTracks = true stops publishing and closes connection`() = runTest {
val mockVideoTrack = mockk<VideoTrack>(relaxed = true) {
Expand Down
Loading