Refactor PCM to AAC transcoder for improved concurrency and error handling

2026-08-23 15:08:32 +00:00 · 2026-03-19 15:10:46 +00:00
parent 945ceea61c
commit 204ae3f05c
2 changed files with 86 additions and 53 deletions
--- a/machinery/src/capture/main.go
+++ b/machinery/src/capture/main.go
@@ -155,9 +155,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat

 				nextPkt, cursorError = recordingCursor.ReadPacket()

-				now := time.Now().UnixMilli()
-				hardMaxReached := now-startRecording > maxRecordingPeriod-500
-				postRecordingElapsed := startRecording+postRecording-now <= 0
+				packetTime := pkt.CurrentTime
+				hardMaxReached := packetTime-startRecording > maxRecordingPeriod-500
+				postRecordingElapsed := startRecording+postRecording-packetTime <= 0
 				if start && (postRecordingElapsed || hardMaxReached) {
 					rolloverRequested = true
 					if hardMaxReached && !rolloverMaxLogged {
@@ -166,18 +166,22 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					}
 				}

-				if start && // If already recording and current frame is a keyframe and we should stop recording
-					rolloverRequested && nextPkt.IsKeyFrame {
+				closeOnCurrentKeyframe := rolloverRequested && pkt.IsKeyFrame && pkt.CurrentTime > startRecording
+				closeOnNextKeyframe := rolloverRequested && nextPkt.IsKeyFrame

-					pts := convertPTS(pkt.TimeLegacy)
-					if pkt.IsVideo {
-						// Write the last packet
-						if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-							log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
-						}
-					} else if pkt.IsAudio {
-						if err := audioWriter.WritePacket(pkt); err != nil {
-							log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
+				if start && (closeOnCurrentKeyframe || closeOnNextKeyframe) {
+
+					if !closeOnCurrentKeyframe {
+						pts := convertPTS(pkt.TimeLegacy)
+						if pkt.IsVideo {
+							// Write the last packet before the rollover keyframe.
+							if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
+								log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
+							}
+						} else if pkt.IsAudio {
+							if err := audioWriter.WritePacket(pkt); err != nil {
+								log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
+							}
 						}
 					}

@@ -549,18 +553,18 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + cursorError.Error())
 					}

-					now = time.Now().UnixMilli()
 					select {
 					case motion := <-communication.HandleMotion:
-						motionTimestamp = now
+						motionTimestamp = pkt.CurrentTime
 						log.Log.Info("capture.main.HandleRecordStream(motiondetection): motion detected while recording. Expanding recording.")
 						numberOfChanges := motion.NumberOfChanges
 						log.Log.Info("capture.main.HandleRecordStream(motiondetection): Received message with recording data, detected changes to save: " + strconv.Itoa(numberOfChanges))
 					default:
 					}

-					hardMaxReached := now-startRecording > maxRecordingPeriod-500
-					postRecordingElapsed := motionTimestamp+postRecording-now < 0
+					packetTime := pkt.CurrentTime
+					hardMaxReached := packetTime-startRecording > maxRecordingPeriod-500
+					postRecordingElapsed := motionTimestamp+postRecording-packetTime < 0
 					if start && (postRecordingElapsed || hardMaxReached) {
 						rolloverRequested = true
 						if hardMaxReached && !rolloverMaxLogged {
@@ -569,10 +573,13 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}

-					if start && rolloverRequested && nextPkt.IsKeyFrame {
-						log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-now < 0  - " + strconv.FormatInt(motionTimestamp+postRecording-now, 10) + " < 0")
-						log.Log.Info("capture.main.HandleRecordStream(motiondetection): now-startRecording > maxRecordingPeriod-500 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 10))
-						log.Log.Info("capture.main.HandleRecordStream(motiondetection): closing recording (timestamp: " + strconv.FormatInt(motionTimestamp, 10) + ", postRecording: " + strconv.FormatInt(postRecording, 10) + ", now: " + strconv.FormatInt(now, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10))
+					closeOnCurrentKeyframe := rolloverRequested && pkt.IsKeyFrame && pkt.CurrentTime > startRecording
+					closeOnNextKeyframe := rolloverRequested && nextPkt.IsKeyFrame
+
+					if start && (closeOnCurrentKeyframe || closeOnNextKeyframe) {
+						log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-packetTime < 0  - " + strconv.FormatInt(motionTimestamp+postRecording-packetTime, 10) + " < 0")
+						log.Log.Info("capture.main.HandleRecordStream(motiondetection): packetTime-startRecording > maxRecordingPeriod-500 - " + strconv.FormatInt(packetTime-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 10))
+						log.Log.Info("capture.main.HandleRecordStream(motiondetection): closing recording (timestamp: " + strconv.FormatInt(motionTimestamp, 10) + ", postRecording: " + strconv.FormatInt(postRecording, 10) + ", packetTime: " + strconv.FormatInt(packetTime, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10))
 						break
 					}
 					if pkt.IsKeyFrame && !start && pkt.CurrentTime >= startRecording {
--- a/machinery/src/capture/pcmu_to_aac_transcoder.go
+++ b/machinery/src/capture/pcmu_to_aac_transcoder.go
@@ -44,6 +44,9 @@ type ffmpegToAACTranscoder struct {
 	closed      bool
 	stdinClosed bool
 	closeOnce   sync.Once
+	stdoutDone  chan struct{}
+	waitDone    chan struct{}
+	waitErr     error
 }

 func newFFmpegToAACTranscoder(inputFormat string, sampleRate int, channels int) (*ffmpegToAACTranscoder, error) {
@@ -88,13 +91,16 @@ func newFFmpegToAACTranscoder(inputFormat string, sampleRate int, channels int)
 	}

 	t := &ffmpegToAACTranscoder{
-		cmd:    cmd,
-		stdin:  stdin,
-		stdout: stdout,
-		stderr: stderr,
+		cmd:        cmd,
+		stdin:      stdin,
+		stdout:     stdout,
+		stderr:     stderr,
+		stdoutDone: make(chan struct{}),
+		waitDone:   make(chan struct{}),
 	}

 	go func() {
+		defer close(t.stdoutDone)
 		buf := make([]byte, 4096)
 		for {
 			n, readErr := stdout.Read(buf)
@@ -112,6 +118,11 @@ func newFFmpegToAACTranscoder(inputFormat string, sampleRate int, channels int)
 		}
 	}()

+	go func() {
+		t.waitErr = cmd.Wait()
+		close(t.waitDone)
+	}()
+
 	log.Log.Info("capture.audio_to_aac: " + strings.ToUpper(inputFormat) + " -> AAC transcoder initialised (ffmpeg process)")
 	return t, nil
 }
@@ -161,17 +172,10 @@ func (t *ffmpegToAACTranscoder) Transcode(input []byte) ([]byte, error) {
 		return nil, err
 	}

-	deadline := time.Now().Add(75 * time.Millisecond)
-	for {
-		data := t.readAvailable()
-		if len(data) > 0 {
-			return data, nil
-		}
-		if time.Now().After(deadline) {
-			return nil, nil
-		}
-		time.Sleep(5 * time.Millisecond)
-	}
+	// Do not block the recording loop waiting for the encoder to emit output.
+	// FFmpeg can buffer for a while, and polling here per RTP packet causes the
+	// recorder to fall behind real time and drop trailing media before close.
+	return t.readAvailable(), nil
 }

 func (t *ffmpegToAACTranscoder) Flush() ([]byte, error) {
@@ -193,25 +197,39 @@ func (t *ffmpegToAACTranscoder) Flush() ([]byte, error) {
 	}
 	t.mu.Unlock()

-	deadline := time.Now().Add(750 * time.Millisecond)
-	previousLen := -1
-	stableReads := 0
-	for {
-		buffered := t.bufferedLen()
-		if buffered == previousLen {
-			stableReads++
-		} else {
-			stableReads = 0
-			previousLen = buffered
+	processExited := false
+	readerFinished := false
+	deadline := time.Now().Add(3 * time.Second)
+	for !processExited || !readerFinished {
+		if !processExited {
+			select {
+			case <-t.waitDone:
+				processExited = true
+			default:
+			}
+		}
+		if !readerFinished {
+			select {
+			case <-t.stdoutDone:
+				readerFinished = true
+			default:
+			}
 		}

-		if stableReads >= 3 || time.Now().After(deadline) {
+		if processExited && readerFinished {
+			break
+		}
+		if time.Now().After(deadline) {
 			break
 		}

 		time.Sleep(15 * time.Millisecond)
 	}

+	if processExited && t.waitErr != nil {
+		return t.readAvailable(), t.waitErr
+	}
+
 	return t.readAvailable(), nil
 }

@@ -229,13 +247,21 @@ func (t *ffmpegToAACTranscoder) Close() {
 		}
 		t.mu.Unlock()

-		if t.stdout != nil {
-			_ = t.stdout.Close()
+		processExited := false
+		select {
+		case <-t.waitDone:
+			processExited = true
+		case <-time.After(250 * time.Millisecond):
 		}

-		if t.cmd != nil && t.cmd.Process != nil {
-			_ = t.cmd.Process.Kill()
-			_, _ = t.cmd.Process.Wait()
+		if !processExited {
+			if t.stdout != nil {
+				_ = t.stdout.Close()
+			}
+			if t.cmd != nil && t.cmd.Process != nil {
+				_ = t.cmd.Process.Kill()
+				<-t.waitDone
+			}
 		}

 		if stderr := t.stderrString(); stderr != "" {