ee321a7 fix: address CodeRabbit findings for transcoding pipeline · dumber

M internal/application/port/transcoder.go +2 -0


@@ -30,4 +30,6 @@ type MediaTranscoder interface {
30	30		// Start begins a GPU transcode session, fetching from sourceURL.
31	31		// Headers are forwarded to the source HTTP request (cookies, auth).
32	32		Start(ctx context.Context, sourceURL string, headers map[string]string) (TranscodeSession, error)
	33	+	// Close shuts down all active transcode sessions and releases resources.
	34	+	Close()
33	35		}


@@ -30,4 +30,6 @@ type MediaTranscoder interface {
30	// Start begins a GPU transcode session, fetching from sourceURL.	30		// Start begins a GPU transcode session, fetching from sourceURL.
31	// Headers are forwarded to the source HTTP request (cookies, auth).	31		// Headers are forwarded to the source HTTP request (cookies, auth).
32	Start(ctx context.Context, sourceURL string, headers map[string]string) (TranscodeSession, error)	32		Start(ctx context.Context, sourceURL string, headers map[string]string) (TranscodeSession, error)
		33	+	// Close shuts down all active transcode sessions and releases resources.
		34	+	Close()
33	}	35		}

M internal/infrastructure/cef/transcoding_handler.go +8 -6


@@ -3,6 +3,7 @@ package cef
3	3		import (
4	4		"context"
5	5		"sync"
	6	+	"time"
6	7		"unsafe"
7	8
8	9		purecef "github.com/bnema/purego-cef/cef"

@@ -72,11 +73,9 @@ func (rh *transcodingResourceHandler) Read(
72	73		return 0
73	74		}
74	75
75		-	buf := make([]byte, bytesToRead)
76		-	n, err := rh.session.Read(buf)
	76	+	dst := unsafe.Slice((*byte)(dataOut), int(bytesToRead))
	77	+	n, err := rh.session.Read(dst)
77	78		if n > 0 {
78		-	dst := unsafe.Slice((*byte)(dataOut), n)
79		-	copy(dst, buf[:n])
80	79		(int32)(bytesRead) = int32(n)
81	80		return 1
82	81		}

@@ -177,9 +176,12 @@ func (h *transcodingRequestHandler) GetResourceHandler(_ purecef.Browser, _ pure
177	176		if !ok {
178	177		return nil
179	178		}
180		-	info := val.(requestInfo)
	179	+	info, ok := val.(requestInfo)
	180	+	if !ok {
	181	+	return nil
	182	+	}
181	183
182		-	ctx, cancel := context.WithCancel(context.Background())
	184	+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
183	185		return purecef.NewResourceHandler(&transcodingResourceHandler{
184	186		transcoder: h.transcoder,
185	187		sourceURL: url,

 import (
 	"context"
 	"sync"
+	"time"
 	"unsafe"
 	purecef "github.com/bnema/purego-cef/cef"
 		return 0
 	}
-	buf := make([]byte, bytesToRead)
+	dst := unsafe.Slice((*byte)(dataOut), int(bytesToRead))
-	n, err := rh.session.Read(buf)
+	n, err := rh.session.Read(dst)
 	if n > 0 {
-		dst := unsafe.Slice((*byte)(dataOut), n)
-		copy(dst, buf[:n])
 		*(*int32)(bytesRead) = int32(n)
 		return 1
 	}
 	if !ok {
 		return nil
 	}
-	info := val.(requestInfo)
+	info, ok := val.(requestInfo)
+	if !ok {
+		return nil
+	}
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	return purecef.NewResourceHandler(&transcodingResourceHandler{
 		transcoder: h.transcoder,
 		sourceURL:  url,

M internal/infrastructure/transcoder/audio.go +48 -30


@@ -6,6 +6,7 @@ import (
6	6		"unsafe"
7	7
8	8		"github.com/bnema/purego-ffmpeg/ffmpeg"
	9	+	"github.com/rs/zerolog"
9	10		)
10	11
11	12		// opusSampleRate is the standard output sample rate for Opus encoding.

@@ -26,6 +27,8 @@ type audioTranscoder struct {
26	27		encoder unsafe.Pointer // AVCodecContext* for Opus encoding
27	28		resampler unsafe.Pointer // SwrContext* (nil if no resampling needed)
28	29		frame unsafe.Pointer // reusable AVFrame for resampled output
	30	+	decFrame unsafe.Pointer // reusable AVFrame for decoded output
	31	+	encPkt unsafe.Pointer // reusable AVPacket for encoded output
29	32		inStream int // input stream index in the source container
30	33		outStream int // output stream index in the muxer
31	34		}

@@ -81,13 +84,7 @@ func newAudioTranscoder(inCodecPar, outFmtCtx unsafe.Pointer, inStreamIdx int) (
81	84		// Set time base to 1/sample_rate for audio.
82	85		ffmpeg.CodecCtxSetTimeBase(encCtx, ffmpeg.AVRational{Num: 1, Den: opusSampleRate})
83	86
84		-	// If the output format requires global headers, set the flag on the encoder.
85		-	outFlags := ffmpeg.FmtCtxFlags(outFmtCtx)
86		-	oformat := ffmpeg.FmtCtxOformat(outFmtCtx)
87		-	if oformat != nil {
88		-	// Check output format flags via the format context flags.
89		-	_ = outFlags // Format flags are on FmtCtx, not oformat.
90		-	}
	87	+	// Note: WebM container does not require AV_CODEC_FLAG_GLOBAL_HEADER for Opus.
91	88
92	89		if ret := ffmpeg.CodecOpen2(encCtx, enc, nil); ret < 0 {
93	90		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))

@@ -161,11 +158,40 @@ func newAudioTranscoder(inCodecPar, outFmtCtx unsafe.Pointer, inStreamIdx int) (
161	158		codecCtxCopyChLayout(resampledFrame, decCtx)
162	159		}
163	160
	161	+	decFrame := ffmpeg.FrameAlloc()
	162	+	if decFrame == nil {
	163	+	if resampledFrame != nil {
	164	+	ffmpeg.FrameFree(unsafe.Pointer(&resampledFrame))
	165	+	}
	166	+	if swr != nil {
	167	+	ffmpeg.SwrFree(unsafe.Pointer(&swr))
	168	+	}
	169	+	ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
	170	+	ffmpeg.CodecFreeContext(unsafe.Pointer(&decCtx))
	171	+	return nil, errors.New("failed to allocate audio decode frame")
	172	+	}
	173	+
	174	+	encPkt := ffmpeg.PacketAlloc()
	175	+	if encPkt == nil {
	176	+	ffmpeg.FrameFree(unsafe.Pointer(&decFrame))
	177	+	if resampledFrame != nil {
	178	+	ffmpeg.FrameFree(unsafe.Pointer(&resampledFrame))
	179	+	}
	180	+	if swr != nil {
	181	+	ffmpeg.SwrFree(unsafe.Pointer(&swr))
	182	+	}
	183	+	ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
	184	+	ffmpeg.CodecFreeContext(unsafe.Pointer(&decCtx))
	185	+	return nil, errors.New("failed to allocate audio encode packet")
	186	+	}
	187	+
164	188		return &audioTranscoder{
165	189		decoder: decCtx,
166	190		encoder: encCtx,
167	191		resampler: swr,
168	192		frame: resampledFrame,
	193	+	decFrame: decFrame,
	194	+	encPkt: encPkt,
169	195		inStream: inStreamIdx,
170	196		outStream: outStreamIdx,
171	197		}, nil

@@ -180,17 +206,8 @@ func (a *audioTranscoder) processPacket(pkt, outFmtCtx unsafe.Pointer) error {
180	206		return fmt.Errorf("audio decode send packet: %d", ret)
181	207		}
182	208
183		-	frame := ffmpeg.FrameAlloc()
184		-	if frame == nil {
185		-	return errors.New("failed to allocate audio decode frame")
186		-	}
187		-	defer ffmpeg.FrameFree(unsafe.Pointer(&frame))
188		-
189		-	encPkt := ffmpeg.PacketAlloc()
190		-	if encPkt == nil {
191		-	return errors.New("failed to allocate audio encode packet")
192		-	}
193		-	defer ffmpeg.PacketFree(unsafe.Pointer(&encPkt))
	209	+	frame := a.decFrame
	210	+	encPkt := a.encPkt
194	211
195	212		for {
196	213		ret = ffmpeg.CodecReceiveFrame(a.decoder, frame)

@@ -272,21 +289,14 @@ func (a *audioTranscoder) processPacket(pkt, outFmtCtx unsafe.Pointer) error {
272	289		}
273	290
274	291		// flush drains remaining frames from the decoder and encoder.
275		-	func (a *audioTranscoder) flush(outFmtCtx unsafe.Pointer) error {
	292	+	// The logger parameter is used to report non-fatal resampling errors
	293	+	// that occur during flush (we continue flushing rather than aborting).
	294	+	func (a *audioTranscoder) flush(outFmtCtx unsafe.Pointer, logger zerolog.Logger) error {
276	295		// Flush decoder by sending NULL packet.
277	296		ffmpeg.CodecSendPacket(a.decoder, nil)
278	297
279		-	frame := ffmpeg.FrameAlloc()
280		-	if frame == nil {
281		-	return errors.New("failed to allocate frame for audio flush")
282		-	}
283		-	defer ffmpeg.FrameFree(unsafe.Pointer(&frame))
284		-
285		-	encPkt := ffmpeg.PacketAlloc()
286		-	if encPkt == nil {
287		-	return errors.New("failed to allocate packet for audio flush")
288		-	}
289		-	defer ffmpeg.PacketFree(unsafe.Pointer(&encPkt))
	298	+	frame := a.decFrame
	299	+	encPkt := a.encPkt
290	300
291	301		// Drain decoded frames.
292	302		for {

@@ -304,6 +314,7 @@ func (a *audioTranscoder) flush(outFmtCtx unsafe.Pointer) error {
304	314
305	315		frameSetNbSamples(a.frame, opusFrameSize)
306	316		if ret := ffmpeg.FrameGetBuffer(a.frame, 0); ret < 0 {
	317	+	logger.Warn().Int32("ret", ret).Msg("failed to allocate resampled frame buffer during flush")
307	318		ffmpeg.FrameUnref(frame)
308	319		continue
309	320		}

@@ -314,6 +325,7 @@ func (a *audioTranscoder) flush(outFmtCtx unsafe.Pointer) error {
314	325		unsafe.Pointer(inData), inSamples,
315	326		)
316	327		if converted < 0 {
	328	+	logger.Warn().Int32("ret", converted).Msg("audio resample failed during flush")
317	329		ffmpeg.FrameUnref(frame)
318	330		continue
319	331		}

@@ -358,6 +370,12 @@ func (a *audioTranscoder) flush(outFmtCtx unsafe.Pointer) error {
358	370
359	371		// close frees all FFmpeg resources owned by the audio transcoder.
360	372		func (a *audioTranscoder) close() {
	373	+	if a.encPkt != nil {
	374	+	ffmpeg.PacketFree(unsafe.Pointer(&a.encPkt))
	375	+	}
	376	+	if a.decFrame != nil {
	377	+	ffmpeg.FrameFree(unsafe.Pointer(&a.decFrame))
	378	+	}
361	379		if a.frame != nil {
362	380		ffmpeg.FrameFree(unsafe.Pointer(&a.frame))
363	381		}

 	"unsafe"
 	"github.com/bnema/purego-ffmpeg/ffmpeg"
+	"github.com/rs/zerolog"
 )
 // opusSampleRate is the standard output sample rate for Opus encoding.
 	encoder   unsafe.Pointer // AVCodecContext* for Opus encoding
 	resampler unsafe.Pointer // SwrContext* (nil if no resampling needed)
 	frame     unsafe.Pointer // reusable AVFrame for resampled output
+	decFrame  unsafe.Pointer // reusable AVFrame for decoded output
+	encPkt    unsafe.Pointer // reusable AVPacket for encoded output
 	inStream  int            // input stream index in the source container
 	outStream int            // output stream index in the muxer
 }
 	// Set time base to 1/sample_rate for audio.
 	ffmpeg.CodecCtxSetTimeBase(encCtx, ffmpeg.AVRational{Num: 1, Den: opusSampleRate})
-	// If the output format requires global headers, set the flag on the encoder.
+	// Note: WebM container does not require AV_CODEC_FLAG_GLOBAL_HEADER for Opus.
-	outFlags := ffmpeg.FmtCtxFlags(outFmtCtx)
-	oformat := ffmpeg.FmtCtxOformat(outFmtCtx)
-	if oformat != nil {
-		// Check output format flags via the format context flags.
-		_ = outFlags // Format flags are on FmtCtx, not oformat.
-	}
 	if ret := ffmpeg.CodecOpen2(encCtx, enc, nil); ret < 0 {
 		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
 		codecCtxCopyChLayout(resampledFrame, decCtx)
 	}
+	decFrame := ffmpeg.FrameAlloc()
+	if decFrame == nil {
+		if resampledFrame != nil {
+			ffmpeg.FrameFree(unsafe.Pointer(&resampledFrame))
+		}
+		if swr != nil {
+			ffmpeg.SwrFree(unsafe.Pointer(&swr))
+		}
+		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
+		ffmpeg.CodecFreeContext(unsafe.Pointer(&decCtx))
+		return nil, errors.New("failed to allocate audio decode frame")
+	}
+	encPkt := ffmpeg.PacketAlloc()
+	if encPkt == nil {
+		ffmpeg.FrameFree(unsafe.Pointer(&decFrame))
+		if resampledFrame != nil {
+			ffmpeg.FrameFree(unsafe.Pointer(&resampledFrame))
+		}
+		if swr != nil {
+			ffmpeg.SwrFree(unsafe.Pointer(&swr))
+		}
+		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
+		ffmpeg.CodecFreeContext(unsafe.Pointer(&decCtx))
+		return nil, errors.New("failed to allocate audio encode packet")
+	}
 	return &audioTranscoder{
 		decoder:   decCtx,
 		encoder:   encCtx,
 		resampler: swr,
 		frame:     resampledFrame,
+		decFrame:  decFrame,
+		encPkt:    encPkt,
 		inStream:  inStreamIdx,
 		outStream: outStreamIdx,
 	}, nil
 		return fmt.Errorf("audio decode send packet: %d", ret)
 	}
-	frame := ffmpeg.FrameAlloc()
+	frame := a.decFrame
-	if frame == nil {
+	encPkt := a.encPkt
-		return errors.New("failed to allocate audio decode frame")
-	}
-	defer ffmpeg.FrameFree(unsafe.Pointer(&frame))
-	encPkt := ffmpeg.PacketAlloc()
-	if encPkt == nil {
-		return errors.New("failed to allocate audio encode packet")
-	}
-	defer ffmpeg.PacketFree(unsafe.Pointer(&encPkt))
 	for {
 		ret = ffmpeg.CodecReceiveFrame(a.decoder, frame)
 }
 // flush drains remaining frames from the decoder and encoder.
-func (a *audioTranscoder) flush(outFmtCtx unsafe.Pointer) error {
+// The logger parameter is used to report non-fatal resampling errors
+// that occur during flush (we continue flushing rather than aborting).
+func (a *audioTranscoder) flush(outFmtCtx unsafe.Pointer, logger zerolog.Logger) error {
 	// Flush decoder by sending NULL packet.
 	ffmpeg.CodecSendPacket(a.decoder, nil)
-	frame := ffmpeg.FrameAlloc()
+	frame := a.decFrame
-	if frame == nil {
+	encPkt := a.encPkt
-		return errors.New("failed to allocate frame for audio flush")
-	}
-	defer ffmpeg.FrameFree(unsafe.Pointer(&frame))
-	encPkt := ffmpeg.PacketAlloc()
-	if encPkt == nil {
-		return errors.New("failed to allocate packet for audio flush")
-	}
-	defer ffmpeg.PacketFree(unsafe.Pointer(&encPkt))
 	// Drain decoded frames.
 	for {
 			frameSetNbSamples(a.frame, opusFrameSize)
 			if ret := ffmpeg.FrameGetBuffer(a.frame, 0); ret < 0 {
+				logger.Warn().Int32("ret", ret).Msg("failed to allocate resampled frame buffer during flush")
 				ffmpeg.FrameUnref(frame)
 				continue
 			}
 				unsafe.Pointer(inData), inSamples,
 			)
 			if converted < 0 {
+				logger.Warn().Int32("ret", converted).Msg("audio resample failed during flush")
 				ffmpeg.FrameUnref(frame)
 				continue
 			}
 // close frees all FFmpeg resources owned by the audio transcoder.
 func (a *audioTranscoder) close() {
+	if a.encPkt != nil {
+		ffmpeg.PacketFree(unsafe.Pointer(&a.encPkt))
+	}
+	if a.decFrame != nil {
+		ffmpeg.FrameFree(unsafe.Pointer(&a.decFrame))
+	}
 	if a.frame != nil {
 		ffmpeg.FrameFree(unsafe.Pointer(&a.frame))
 	}

M internal/infrastructure/transcoder/hwprobe.go +1 -3

@@ -1,8 +1,6 @@
1	1		package transcoder
2	2
3	3		import (
4		-	"unsafe"
5		-
6	4		"github.com/bnema/purego-ffmpeg/ffmpeg"
7	5		"github.com/rs/zerolog"
8	6

@@ -90,7 +88,7 @@ func probeProfile(p hwProfile, logger *zerolog.Logger) port.HWCapabilities {
90	88		}
91	89		// Free the probing device context; a long-lived one will be
92	90		// created per transcode session (or shared across sessions).
93		-	defer ffmpeg.BufferUnref(unsafe.Pointer(&deviceCtx))
	91	+	defer ffmpeg.BufferUnref(&deviceCtx)
94	92
95	93		logger.Info().
96	94		Str("api", p.api).

 package transcoder
 import (
-	"unsafe"
 	"github.com/bnema/purego-ffmpeg/ffmpeg"
 	"github.com/rs/zerolog"
 	}
 	// Free the probing device context; a long-lived one will be
 	// created per transcode session (or shared across sessions).
-	defer ffmpeg.BufferUnref(unsafe.Pointer(&deviceCtx))
+	defer ffmpeg.BufferUnref(&deviceCtx)
 	logger.Info().
 		Str("api", p.api).

M internal/infrastructure/transcoder/pipeline.go +28 -11


@@ -9,7 +9,11 @@ import (
9	9		"strings"
10	10		"unsafe"
11	11
	12	+	"net"
	13	+	"time"
	14	+
12	15		"github.com/ebitengine/purego"
	16	+	"github.com/rs/zerolog"
13	17
14	18		"github.com/bnema/purego-ffmpeg/ffmpeg"
15	19

@@ -31,19 +35,32 @@ type pipeline struct {
31	35		quality string
32	36		pw *io.PipeWriter
33	37		cancel context.CancelFunc
	38	+	logger zerolog.Logger
34	39
35	40		// Prevent GC of callback closures and uintptrs.
36	41		readCb uintptr
37	42		writeCb uintptr
38	43		}
39	44
40		-	func newPipeline(hwCaps port.HWCapabilities, sourceURL string, headers map[string]string, quality string, pw io.PipeWriter) pipeline {
	45	+	// httpClient is used for fetching source media. It has connect/TLS/header
	46	+	// timeouts but no overall Timeout so that long-lived streaming responses
	47	+	// are not killed. The session context handles cancellation.
	48	+	var httpClient = &http.Client{
	49	+	Transport: &http.Transport{
	50	+	DialContext: (&net.Dialer{Timeout: 10 * time.Second}).DialContext,
	51	+	TLSHandshakeTimeout: 10 * time.Second,
	52	+	ResponseHeaderTimeout: 15 * time.Second,
	53	+	},
	54	+	}
	55	+
	56	+	func newPipeline(hwCaps port.HWCapabilities, sourceURL string, headers map[string]string, quality string, pw io.PipeWriter, logger zerolog.Logger) pipeline {
41	57		return &pipeline{
42	58		hwCaps: hwCaps,
43	59		sourceURL: sourceURL,
44	60		headers: headers,
45	61		quality: quality,
46	62		pw: pw,
	63	+	logger: logger,
47	64		}
48	65		}
49	66

@@ -70,7 +87,7 @@ func (p *pipeline) doRun(ctx context.Context) error {
70	87		req.Header.Set(k, v)
71	88		}
72	89
73		-	resp, err := http.DefaultClient.Do(req)
	90	+	resp, err := httpClient.Do(req)
74	91		if err != nil {
75	92		return fmt.Errorf("fetch source: %w", err)
76	93		}

@@ -150,7 +167,7 @@ func (p *pipeline) doRun(ctx context.Context) error {
150	167		return fmt.Errorf("open video encoder: %w", err)
151	168		}
152	169		if hwDeviceCtx != nil {
153		-	defer ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
	170	+	defer ffmpeg.BufferUnref(&hwDeviceCtx)
154	171		}
155	172		defer ffmpeg.CodecFreeContext(unsafe.Pointer(&vidEncCtx))
156	173

@@ -302,7 +319,7 @@ func (p *pipeline) doRun(ctx context.Context) error {
302	319
303	320		// Flush audio.
304	321		if audioTx != nil {
305		-	audioTx.flush(outFmtCtx)
	322	+	audioTx.flush(outFmtCtx, p.logger)
306	323		}
307	324
308	325		// --- Step 15: Write trailer ---

@@ -388,13 +405,13 @@ func (p *pipeline) openVideoEncoder(decCtx unsafe.Pointer) (hwDeviceCtx unsafe.P
388	405		}
389	406		}
390	407		if enc == nil {
391		-	ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
	408	+	ffmpeg.BufferUnref(&hwDeviceCtx)
392	409		return nil, nil, errors.New("no usable encoder codec found")
393	410		}
394	411
395	412		encCtx = ffmpeg.CodecAllocContext3(enc)
396	413		if encCtx == nil {
397		-	ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
	414	+	ffmpeg.BufferUnref(&hwDeviceCtx)
398	415		return nil, nil, errors.New("failed to allocate video encoder context")
399	416		}
400	417

@@ -434,7 +451,7 @@ func (p *pipeline) openVideoEncoder(decCtx unsafe.Pointer) (hwDeviceCtx unsafe.P
434	451		hwDeviceRef := ffmpeg.BufferRef(hwDeviceCtx)
435	452		if hwDeviceRef == nil {
436	453		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
437		-	ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
	454	+	ffmpeg.BufferUnref(&hwDeviceCtx)
438	455		return nil, nil, errors.New("failed to ref HW device context for encoder")
439	456		}
440	457		ffmpeg.CodecCtxSetHwDeviceCtx(encCtx, hwDeviceRef)

@@ -444,7 +461,7 @@ func (p *pipeline) openVideoEncoder(decCtx unsafe.Pointer) (hwDeviceCtx unsafe.P
444	461		framesRef := ffmpeg.HWFrameCtxAlloc(hwDeviceCtx)
445	462		if framesRef == nil {
446	463		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
447		-	ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
	464	+	ffmpeg.BufferUnref(&hwDeviceCtx)
448	465		return nil, nil, errors.New("failed to allocate HW frames context")
449	466		}
450	467

@@ -478,9 +495,9 @@ func (p *pipeline) openVideoEncoder(decCtx unsafe.Pointer) (hwDeviceCtx unsafe.P
478	495		}
479	496
480	497		if ret := ffmpeg.HWFrameCtxInit(framesRef); ret < 0 {
481		-	ffmpeg.BufferUnref(unsafe.Pointer(&framesRef))
	498	+	ffmpeg.BufferUnref(&framesRef)
482	499		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
483		-	ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
	500	+	ffmpeg.BufferUnref(&hwDeviceCtx)
484	501		return nil, nil, fmt.Errorf("init HW frames context: %d", ret)
485	502		}
486	503

@@ -489,7 +506,7 @@ func (p *pipeline) openVideoEncoder(decCtx unsafe.Pointer) (hwDeviceCtx unsafe.P
489	506
490	507		if ret := ffmpeg.CodecOpen2(encCtx, enc, nil); ret < 0 {
491	508		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
492		-	ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
	509	+	ffmpeg.BufferUnref(&hwDeviceCtx)
493	510		return nil, nil, fmt.Errorf("open video encoder %s: %d", encoderName, ret)
494	511		}
495	512

 	"strings"
 	"unsafe"
+	"net"
+	"time"
 	"github.com/ebitengine/purego"
+	"github.com/rs/zerolog"
 	"github.com/bnema/purego-ffmpeg/ffmpeg"
 	quality   string
 	pw        *io.PipeWriter
 	cancel    context.CancelFunc
+	logger    zerolog.Logger
 	// Prevent GC of callback closures and uintptrs.
 	readCb  uintptr
 	writeCb uintptr
 }
-func newPipeline(hwCaps port.HWCapabilities, sourceURL string, headers map[string]string, quality string, pw *io.PipeWriter) *pipeline {
+// httpClient is used for fetching source media. It has connect/TLS/header
+// timeouts but no overall Timeout so that long-lived streaming responses
+// are not killed. The session context handles cancellation.
+var httpClient = &http.Client{
+	Transport: &http.Transport{
+		DialContext:           (&net.Dialer{Timeout: 10 * time.Second}).DialContext,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ResponseHeaderTimeout: 15 * time.Second,
+	},
+}
+func newPipeline(hwCaps port.HWCapabilities, sourceURL string, headers map[string]string, quality string, pw *io.PipeWriter, logger zerolog.Logger) *pipeline {
 	return &pipeline{
 		hwCaps:    hwCaps,
 		sourceURL: sourceURL,
 		headers:   headers,
 		quality:   quality,
 		pw:        pw,
+		logger:    logger,
 	}
 }
 		req.Header.Set(k, v)
 	}
-	resp, err := http.DefaultClient.Do(req)
+	resp, err := httpClient.Do(req)
 	if err != nil {
 		return fmt.Errorf("fetch source: %w", err)
 	}
 		return fmt.Errorf("open video encoder: %w", err)
 	}
 	if hwDeviceCtx != nil {
-		defer ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
+		defer ffmpeg.BufferUnref(&hwDeviceCtx)
 	}
 	defer ffmpeg.CodecFreeContext(unsafe.Pointer(&vidEncCtx))
 	// Flush audio.
 	if audioTx != nil {
-		audioTx.flush(outFmtCtx)
+		audioTx.flush(outFmtCtx, p.logger)
 	}
 	// --- Step 15: Write trailer ---
 		}
 	}
 	if enc == nil {
-		ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
+		ffmpeg.BufferUnref(&hwDeviceCtx)
 		return nil, nil, errors.New("no usable encoder codec found")
 	}
 	encCtx = ffmpeg.CodecAllocContext3(enc)
 	if encCtx == nil {
-		ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
+		ffmpeg.BufferUnref(&hwDeviceCtx)
 		return nil, nil, errors.New("failed to allocate video encoder context")
 	}
 	hwDeviceRef := ffmpeg.BufferRef(hwDeviceCtx)
 	if hwDeviceRef == nil {
 		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
-		ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
+		ffmpeg.BufferUnref(&hwDeviceCtx)
 		return nil, nil, errors.New("failed to ref HW device context for encoder")
 	}
 	ffmpeg.CodecCtxSetHwDeviceCtx(encCtx, hwDeviceRef)
 		framesRef := ffmpeg.HWFrameCtxAlloc(hwDeviceCtx)
 		if framesRef == nil {
 			ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
-			ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
+			ffmpeg.BufferUnref(&hwDeviceCtx)
 			return nil, nil, errors.New("failed to allocate HW frames context")
 		}
 		}
 		if ret := ffmpeg.HWFrameCtxInit(framesRef); ret < 0 {
-			ffmpeg.BufferUnref(unsafe.Pointer(&framesRef))
+			ffmpeg.BufferUnref(&framesRef)
 			ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
-			ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
+			ffmpeg.BufferUnref(&hwDeviceCtx)
 			return nil, nil, fmt.Errorf("init HW frames context: %d", ret)
 		}
 	if ret := ffmpeg.CodecOpen2(encCtx, enc, nil); ret < 0 {
 		ffmpeg.CodecFreeContext(unsafe.Pointer(&encCtx))
-		ffmpeg.BufferUnref(unsafe.Pointer(&hwDeviceCtx))
+		ffmpeg.BufferUnref(&hwDeviceCtx)
 		return nil, nil, fmt.Errorf("open video encoder %s: %d", encoderName, ret)
 	}

M internal/infrastructure/transcoder/pool.go +3 -3


@@ -49,7 +49,7 @@ var errPoolAtCapacity = errors.New("transcoder: session pool at capacity")
49	49
50	50		// sessionPool tracks active transcode sessions with a concurrency limit.
51	51		type sessionPool struct {
52		-	mu sync.Mutex
	52	+	mu sync.RWMutex
53	53		sessions map[string]*session
54	54		maxConc int
55	55		}

@@ -83,8 +83,8 @@ func (p *sessionPool) remove(id string) {
83	83
84	84		// count returns the number of active sessions.
85	85		func (p *sessionPool) count() int {
86		-	p.mu.Lock()
87		-	defer p.mu.Unlock()
	86	+	p.mu.RLock()
	87	+	defer p.mu.RUnlock()
88	88		return len(p.sessions)
89	89		}
90	90


@@ -49,7 +49,7 @@ var errPoolAtCapacity = errors.New("transcoder: session pool at capacity")
49			49
50		// sessionPool tracks active transcode sessions with a concurrency limit.	50		// sessionPool tracks active transcode sessions with a concurrency limit.
51		type sessionPool struct {	51		type sessionPool struct {
52	-	mu sync.Mutex	52	+	mu sync.RWMutex
53		sessions map[string]*session	53		sessions map[string]*session
54		maxConc int	54		maxConc int
55		}	55		}

@@ -83,8 +83,8 @@ func (p *sessionPool) remove(id string) {
83			83
84		// count returns the number of active sessions.	84		// count returns the number of active sessions.
85		func (p *sessionPool) count() int {	85		func (p *sessionPool) count() int {
86	-	p.mu.Lock()	86	+	p.mu.RLock()
87	-	defer p.mu.Unlock()	87	+	defer p.mu.RUnlock()
88		return len(p.sessions)	88		return len(p.sessions)
89		}	89		}
90			90

M internal/infrastructure/transcoder/transcoder.go +10 -5


@@ -111,7 +111,7 @@ func (t *Transcoder) Start(ctx context.Context, sourceURL string, headers map[st
111	111		return nil, fmt.Errorf("transcoder: %w", err)
112	112		}
113	113
114		-	p := newPipeline(t.hwCaps, sourceURL, headers, t.cfg.Quality, pw)
	114	+	p := newPipeline(t.hwCaps, sourceURL, headers, t.cfg.Quality, pw, t.logger)
115	115
116	116		t.logger.Info().
117	117		Str("session_id", id).

@@ -120,11 +120,16 @@ func (t *Transcoder) Start(ctx context.Context, sourceURL string, headers map[st
120	120		Msg("starting transcode session")
121	121
122	122		go func() {
	123	+	defer func() {
	124	+	t.pool.remove(id)
	125	+	t.logger.Info().Str("session_id", id).Msg("transcode session finished")
	126	+	}()
	127	+	defer func() {
	128	+	if r := recover(); r != nil {
	129	+	t.logger.Error().Str("session_id", id).Interface("panic", r).Msg("transcode session panicked")
	130	+	}
	131	+	}()
123	132		p.run(sessionCtx)
124		-	t.pool.remove(id)
125		-	t.logger.Info().
126		-	Str("session_id", id).
127		-	Msg("transcode session finished")
128	133		}()
129	134
130	135		return s, nil

 		return nil, fmt.Errorf("transcoder: %w", err)
 	}
-	p := newPipeline(t.hwCaps, sourceURL, headers, t.cfg.Quality, pw)
+	p := newPipeline(t.hwCaps, sourceURL, headers, t.cfg.Quality, pw, t.logger)
 	t.logger.Info().
 		Str("session_id", id).
 		Msg("starting transcode session")
 	go func() {
+		defer func() {
+			t.pool.remove(id)
+			t.logger.Info().Str("session_id", id).Msg("transcode session finished")
+		}()
+		defer func() {
+			if r := recover(); r != nil {
+				t.logger.Error().Str("session_id", id).Interface("panic", r).Msg("transcode session panicked")
+			}
+		}()
 		p.run(sessionCtx)
-		t.pool.remove(id)
-		t.logger.Info().
-			Str("session_id", id).
-			Msg("transcode session finished")
 	}()
 	return s, nil