feat: Reset to a new track after a long gap. (#17)

* feat: Reset to a new track after a long gap.

pom.xml

@@ -59,6 +59,12 @@
       <version>${kotest.version}</version>
       <scope>test</scope>
     </dependency>
+      <dependency>
+          <groupId>org.jitsi</groupId>
+          <artifactId>jicoco-test-kotlin</artifactId>
+          <version>1.1-144-ga2c5ec1</version>
+          <scope>test</scope>
+      </dependency>
   </dependencies>
   <build>
     <testSourceDirectory>src/test/kotlin</testSourceDirectory>

src/main/kotlin/org/jitsi/recorder/Config.kt

@@ -19,6 +19,8 @@ package org.jitsi.recorder
 
 import org.jitsi.config.JitsiConfig
 import org.jitsi.metaconfig.config
+import kotlin.time.Duration
+import kotlin.time.toKotlinDuration
 
 class Config {
     companion object {
@@ -36,5 +38,18 @@ class Config {
                 RecordingFormat.valueOf(it.uppercase())
             }
         }
+
+        val maxGapDuration: Duration by config {
+            "$BASE.recording.max-gap-duration".from(JitsiConfig.newConfig).convertFrom<java.time.Duration> {
+                it.toKotlinDuration()
+            }
+        }
+
+        override fun toString(): String = """
+            port: $port
+            recordingDirectory: $recordingDirectory
+            recordingFormat: $recordingFormat
+            maxGapDuration: $maxGapDuration
+        """.trimIndent()
     }
 }

src/main/kotlin/org/jitsi/recorder/Main.kt

@@ -80,8 +80,7 @@ fun Application.module() {
 }
 
 fun main(args: Array<String>) {
-    logger.info("Running main, port=${Config.port}, recordingDirectory=${Config.recordingDirectory}")
-    logger.info("Recording format: ${Config.recordingFormat}")
+    logger.info("Starting jitsi-multitrack-recorder with config:\n $Config")
     metrics.sessionsStarted.inc()
     embeddedServer(Netty, port = Config.port, host = "0.0.0.0", module = Application::module)
         .start(wait = true)

src/main/kotlin/org/jitsi/recorder/MediaJsonMkaRecorder.kt

@@ -20,6 +20,7 @@ package org.jitsi.recorder
 import org.jitsi.mediajson.Event
 import org.jitsi.mediajson.MediaEvent
 import org.jitsi.mediajson.StartEvent
+import org.jitsi.recorder.opus.GapTooLargeException
 import org.jitsi.recorder.opus.OpusPacket
 import org.jitsi.recorder.opus.PacketLossConcealmentInserter
 import org.jitsi.utils.logging2.Logger
@@ -75,7 +76,25 @@ class MediaJsonMkaRecorder(directory: File, parentLogger: Logger) : MediaJsonRec
             }
 
             is MediaEvent -> {
-                trackRecorders[event.media.tag]?.addPacket(event) ?: logger.warn("No track for ${event.media.tag}")
+                val trackRecorder = trackRecorders[event.media.tag] ?: run {
+                    logger.warn("No track for ${event.media.tag}")
+                    return
+                }
+
+                try {
+                    trackRecorder.addPacket(event)
+                } catch (e: GapTooLargeException) {
+                    logger.info("Large gap encountered (${e.gapDuration}), resetting track.")
+                    TrackRecorder(
+                        mkaRecorder,
+                        event.media.tag,
+                        trackRecorder.endpointId,
+                        logger
+                    ).let {
+                        trackRecorders[event.media.tag] = it
+                        it.addPacket(event)
+                    }
+                }
             }
         }
     }
@@ -90,13 +109,13 @@ class MediaJsonMkaRecorder(directory: File, parentLogger: Logger) : MediaJsonRec
 private class TrackRecorder(
     private val mkaRecorder: MkaRecorder,
     private val trackName: String,
-    endpointId: String?,
+    val endpointId: String?,
     parentLogger: Logger
 ) {
     private val logger: Logger = parentLogger.createChildLogger(this.javaClass.name).apply {
         addContext("track", trackName)
     }
-    private val plcInserter = PacketLossConcealmentInserter(logger)
+    private val plcInserter = PacketLossConcealmentInserter(Config.maxGapDuration, logger)
     private var stereo = false
 
     init {

src/main/kotlin/org/jitsi/recorder/opus/PacketLossConcealmentInserter.kt

@@ -19,11 +19,17 @@ package org.jitsi.recorder.opus
 
 import org.jitsi.utils.logging2.Logger
 import org.jitsi.utils.logging2.LoggerImpl
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
 
 /**
  * Insert packet loss concealment packets into a stream of [OpusPacket].
  */
-class PacketLossConcealmentInserter(parentLogger: Logger = LoggerImpl("PacketLossConcealmentInserter")) {
+class PacketLossConcealmentInserter(
+    private val maxGapDuration: Duration = 1.minutes,
+    parentLogger: Logger = LoggerImpl("PacketLossConcealmentInserter")
+) {
     private val logger: Logger = parentLogger.createChildLogger(this.javaClass.name)
     private var nextSampleTs = -1L
     private var previousToc: OpusToc? = null
@@ -51,6 +57,9 @@ class PacketLossConcealmentInserter(parentLogger: Logger = LoggerImpl("PacketLos
         } else {
             val missing = ts48kHz - nextSampleTs
             val missingMs = missing.toMs()
+            if (missingMs.milliseconds > maxGapDuration) {
+                throw GapTooLargeException(missingMs.milliseconds)
+            }
             logger.warn("Missing $missing ticks = $missingMs ms (and $remainingMs ms remaining)")
 
             val plc = OpusPacket.generatePlc(missingMs + remainingMs, previousToc ?: opusPacket.toc())
@@ -76,3 +85,4 @@ fun Long.toMs(): Double = this.toDouble() / 48.0
 fun Double.to48kHz(): Long = (this * 48).toLong()
 
 data class OpusPacketAndTimestamp(val packet: OpusPacket, val timestampMs: Long)
+class GapTooLargeException(val gapDuration: Duration) : Exception("Gap too large")

src/main/resources/reference.conf

@@ -3,6 +3,8 @@ jitsi-multitrack-recorder {
     directory = "/tmp"
     // Supported formats: "mka", "mka"
     format = "mka"
+    // The maximum duration of a gap to repair using Opus PLC.
+    max-gap-duration = 5 minutes
   }
   port = 8989
 }

src/test/kotlin/org/jitsi/recorder/KotestProjectConfig.kt

@@ -0,0 +1,29 @@
+/*
+ * Jitsi Multi Track Recorder
+ *
+ * Copyright @ 2024-Present 8x8, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jitsi.recorder
+
+import io.kotest.core.config.AbstractProjectConfig
+import org.jitsi.metaconfig.MetaconfigSettings
+
+class KotestProjectConfig : AbstractProjectConfig() {
+    override suspend fun beforeProject() = super.beforeProject().also {
+        // The only purpose of config caching is performance. We always want caching disabled in tests (so we can
+        // freely modify the config without affecting other tests executing afterwards).
+        MetaconfigSettings.cacheEnabled = false
+    }
+}

src/test/kotlin/org/jitsi/recorder/MkaRecorderTest.kt

@@ -21,11 +21,13 @@ import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import io.kotest.assertions.fail
 import io.kotest.core.spec.style.ShouldSpec
 import io.kotest.matchers.ints.shouldBeGreaterThanOrEqual
+import io.kotest.matchers.shouldBe
 import org.ebml.EBMLReader
 import org.ebml.Element
 import org.ebml.MasterElement
 import org.ebml.io.DataSource
 import org.ebml.io.FileDataSource
+import org.jitsi.config.withNewConfig
 import org.jitsi.mediajson.Event
 import org.jitsi.mediajson.MediaEvent
 import org.jitsi.mediajson.StartEvent
@@ -34,19 +36,15 @@ import java.nio.file.Files
 import kotlin.io.encoding.Base64
 import kotlin.io.encoding.ExperimentalEncodingApi
 import kotlin.random.Random
+import kotlin.time.Duration.Companion.seconds
 
 @OptIn(ExperimentalEncodingApi::class)
 class MkaRecorderTest : ShouldSpec() {
     private val logger = createLogger()
 
     val debug = false
-    val sample = "/opus-stereo.json"
     val loss = 0
-    val input = javaClass.getResource(sample)?.readText()?.lines()?.dropLast(1) ?: fail("Can not read $sample")
     val objectMapper = jacksonObjectMapper()
-    val inputJson: List<Event> = input.map { objectMapper.readValue(it, Event::class.java) }.also {
-        logger.info("Parsed ${it.size} events")
-    }
 
     init {
         setupInPlaceIoPool()
@@ -58,6 +56,7 @@ class MkaRecorderTest : ShouldSpec() {
             val mkaFile = "$directory/recording.mka"
             val recorder = MkaRecorder(directory)
             runOnce(
+                "/sample-stereo.json",
                 mkaFile,
                 {
                     when (it) {
@@ -75,14 +74,38 @@ class MkaRecorderTest : ShouldSpec() {
             val mkaFile = "$directory/recording.mka"
             val recorder = MediaJsonMkaRecorder(directory, logger)
             runOnce(
+                "/sample-stereo.json",
                 mkaFile,
                 { recorder.addEvent(it) },
                 { recorder.stop() }
             )
         }
+        context("Test PLC with a big gap") {
+            withNewConfig("jitsi-multitrack-recorder.recording.max-gap-duration = 5 seconds") {
+                Config.maxGapDuration shouldBe 5.seconds
+
+                val directory = Files.createTempDirectory("MediaJsonMkaRecorderTest").toFile()
+                val mkaFile = "$directory/recording.mka"
+                val recorder = MediaJsonMkaRecorder(directory, logger)
+                runOnce(
+                    "/sample-gap.json",
+                    mkaFile,
+                    { recorder.addEvent(it) },
+                    { recorder.stop() }
+                )
+
+                // The sample has 2 endpoints, but one has a gap of 10 seconds, so it should be split into 2 tracks.
+                traverseMka(mkaFile) { it.elementType.name == "TrackEntry" } shouldBe 3
+            }
+        }
     }
 
-    fun runOnce(mkaFile: String, addEvent: (Event) -> Unit, close: () -> Unit) {
+    fun runOnce(sample: String, mkaFile: String, addEvent: (Event) -> Unit, close: () -> Unit) {
+        val input = javaClass.getResource(sample)?.readText()?.lines()?.dropLast(1) ?: fail("Can not read $sample")
+        val inputJson: List<Event> = input.map { objectMapper.readValue(it, Event::class.java) }.also {
+            logger.info("Parsed ${it.size} events")
+        }
+
         var mediaPackets = 0
         var lostPackets = 0
         logger.warn("Using ${loss * 100}% packet loss.")