From d433cfb9255b2544ae70e0d46aed827f95b7fb94 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Mon, 1 Jun 2026 19:17:38 -0700 Subject: [PATCH] media: wire the detached zero-downtime ingest path end to end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Connect the production entry points to the detached worker machinery. - Worker body de-framing: WorkerInput reconstructs the raw media off the fd-passed push connection — prepend the bytes main read past the headers (Prebuf) then de-chunk if the push used chunked transfer-encoding (httputil.NewChunkedReader). The socket-mode subcommand applies it. - MKVIngestDetached: main's hijacked push connection → fd-pass to a detached worker on a per-session unix socket → consume its frames into ValidateMP4 with reconnect; reap on clean end, leave running on a main-shutdown ctx cancel. - ResumeDetachedWorkers: at startup, discover sockets of workers that outlived a restart and resume draining them. Wired into runMain under --isolated-ingest. - handleIncomingStream: with --isolated-ingest, hijack the authed push and route to MKVIngestDetached (falling back to the fd-4-pipe isolation when the connection can't be hijacked). TestWorkerInputDeframes covers the prebuf + chunked de-framing deterministically. The detached lifecycle (spawn/discover/reconnect/serve) is proven by TestDetachedWorkerZeroDowntime; the buffer-across-restart by the frameServer tests. The api hijack glue is thin over those tested pieces and is the one bit that still needs a real-push integration check (no in-container harness for it). Co-Authored-By: Claude Opus 4.8 --- pkg/api/api_internal.go | 22 ++++++++ pkg/cmd/streamplace.go | 11 +++- pkg/media/ingest_daemon.go | 86 +++++++++++++++++++++++++++++ pkg/media/ingest_subprocess_test.go | 6 +- pkg/media/ingest_worker.go | 25 +++++++++ pkg/media/ingest_worker_test.go | 21 +++++++ 6 files changed, 165 insertions(+), 6 deletions(-) diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index f3f5a529..67b54ae3 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -263,6 +263,28 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err } if a.CLI.IsolatedIngest { + // Zero-downtime path: hijack the authed push connection and hand it to a + // DETACHED worker that owns the connection (so it survives a main + // restart) and serves signed segments back over its socket. + if hj, ok := w.(http.Hijacker); ok { + conn, bufrw, herr := hj.Hijack() + if herr != nil { + log.Error(reqCtx, "ingest hijack failed", "error", herr) + return + } + var prebuf []byte + if n := bufrw.Reader.Buffered(); n > 0 { + prebuf = make([]byte, n) + _, _ = io.ReadFull(bufrw.Reader, prebuf) + } + chunked := len(httpReq.TransferEncoding) > 0 && httpReq.TransferEncoding[0] == "chunked" + if derr := a.MediaManager.MKVIngestDetached(reqCtx, conn, prebuf, chunked, mediaSigner); derr != nil { + log.Log(reqCtx, "isolated stream ended", "error", derr) + } + return // connection hijacked; the HTTP response is ours now + } + // No hijack support (HTTP/2, some proxies) → fd-4-pipe isolation: still + // fault-isolated, just no restart-survival. err = a.MediaManager.MKVIngestIsolated(reqCtx, r, mediaSigner) } else { err = a.MediaManager.MKVIngest(reqCtx, r, mediaSigner) diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index b0faf281..8b3f1f29 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -263,6 +263,11 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu if err != nil { return err } + if cli.IsolatedIngest { + // Reconnect to any ingest workers still running from before this restart + // and drain whatever they buffered while we were down (zero-downtime). + mm.ResumeDetachedWorkers(ctx) + } ms, err := media.MakeMediaSigner(ctx, cli, cli.StreamerName, signer, mod) if err != nil { @@ -869,16 +874,16 @@ func makeIngestWorkerCommand(build *config.BuildFlags) *urfavecli.Command { // Media comes from the fd-passed ingest connection (InputFD) when main // handed one off, else stdin. if cfg.SocketPath != "" { - input := io.Reader(os.Stdin) + raw := io.Reader(os.Stdin) if cfg.InputFD > 0 { f := os.NewFile(uintptr(cfg.InputFD), "ingest-input") if f == nil { return fmt.Errorf("ingest-worker: bad input fd %d", cfg.InputFD) } defer f.Close() - input = f + raw = f } - return media.ServeMKVIngestWorkerSocket(ctx, cfg, input) + return media.ServeMKVIngestWorkerSocket(ctx, cfg, media.WorkerInput(cfg, raw)) } framesFile := os.NewFile(4, "ingest-frames") diff --git a/pkg/media/ingest_daemon.go b/pkg/media/ingest_daemon.go index 00ea09ed..0af289fe 100644 --- a/pkg/media/ingest_daemon.go +++ b/pkg/media/ingest_daemon.go @@ -13,6 +13,7 @@ import ( "syscall" "time" + "github.com/google/uuid" "stream.place/streamplace/pkg/log" ) @@ -104,6 +105,91 @@ func (mm *MediaManager) ConsumeWorkerSocket(ctx context.Context, socketPath, str } } +// ingestWorkerSocketDir returns (creating it) the directory of per-session +// worker frame sockets — the set a restarting main scans to resume. +func (mm *MediaManager) ingestWorkerSocketDir() (string, error) { + dir := mm.cli.DataFilePath([]string{"ingest-workers"}) + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", err + } + return dir, nil +} + +// MKVIngestDetached is the production zero-downtime entry: main has authed the +// push and hijacked its connection; this fd-passes that connection to a DETACHED +// worker (own session, survives a main restart) which ingests the media directly +// and serves signed segments over a per-session unix socket, and then consumes +// those frames into ValidateMP4 with reconnect. prebuf is any body bytes main +// already read past the headers; chunked says the push body is chunked. +// +// Because the worker owns the connection and is detached, a main restart neither +// breaks the ingest nor loses output: the worker keeps signing into its buffer, +// and the restarted main rediscovers the socket (DiscoverWorkerSockets) and +// drains it. +func (mm *MediaManager) MKVIngestDetached(ctx context.Context, conn net.Conn, prebuf []byte, chunked bool, ms MediaSigner) error { + cfg, err := mm.buildWorkerConfig(ctx, ms) + if err != nil { + return err + } + dir, err := mm.ingestWorkerSocketDir() + if err != nil { + return err + } + cfg.SocketPath = filepath.Join(dir, uuid.NewString()+".sock") + cfg.InputFD = 4 + cfg.Prebuf = prebuf + cfg.Chunked = chunked + + tcp, ok := conn.(*net.TCPConn) + if !ok { + return fmt.Errorf("isolated ingest requires a TCP connection, got %T", conn) + } + connFile, err := tcp.File() // dup the fd to hand to the worker + if err != nil { + return fmt.Errorf("dup ingest connection: %w", err) + } + + proc, err := SpawnIngestWorkerDetached(cfg, connFile) + connFile.Close() // the worker holds its own dup + conn.Close() // main is out of the media path now + if err != nil { + return fmt.Errorf("spawn detached worker: %w", err) + } + + err = mm.ConsumeWorkerSocket(ctx, cfg.SocketPath, ms.Streamer(), mm.validateSegment(ctx)) + if err == nil { + go func() { _, _ = proc.Wait() }() // clean end: reap the exiting worker + } + // On ctx cancel (main shutting down) we deliberately leave the detached worker + // running; a restarting main reconnects via discovery. + return err +} + +// ResumeDetachedWorkers reconnects to any ingest workers still running from +// before a main restart and resumes consuming their frames (draining whatever +// they buffered while main was down). Intended to run once at main startup. +func (mm *MediaManager) ResumeDetachedWorkers(ctx context.Context) { + dir, err := mm.ingestWorkerSocketDir() + if err != nil { + log.Error(ctx, "resume ingest workers: socket dir", "error", err) + return + } + socks, err := DiscoverWorkerSockets(dir) + if err != nil { + log.Error(ctx, "resume ingest workers: discover", "error", err) + return + } + for _, sock := range socks { + sock := sock + log.Log(ctx, "resuming detached ingest worker", "socket", sock) + go func() { + if cerr := mm.ConsumeWorkerSocket(ctx, sock, "resumed", mm.validateSegment(ctx)); cerr != nil { + log.Error(ctx, "resumed ingest worker ended", "socket", sock, "error", cerr) + } + }() + } +} + // DiscoverWorkerSockets lists the worker frame sockets under dir — the running // workers a restarting main should reconnect to and resume consuming. func DiscoverWorkerSockets(dir string) ([]string, error) { diff --git a/pkg/media/ingest_subprocess_test.go b/pkg/media/ingest_subprocess_test.go index ea9d48dd..04d518d0 100644 --- a/pkg/media/ingest_subprocess_test.go +++ b/pkg/media/ingest_subprocess_test.go @@ -41,16 +41,16 @@ func runIngestWorkerHelper() int { // Socket mode (Stage 4): serve frames over the unix socket; media comes from // the fd-passed ingest connection (InputFD) when present, else stdin. if cfg.SocketPath != "" { - input := io.Reader(os.Stdin) + raw := io.Reader(os.Stdin) if cfg.InputFD > 0 { f := os.NewFile(uintptr(cfg.InputFD), "ingest-input") if f == nil { return 2 } defer f.Close() - input = f + raw = f } - if err := ServeMKVIngestWorkerSocket(context.Background(), cfg, input); err != nil { + if err := ServeMKVIngestWorkerSocket(context.Background(), cfg, WorkerInput(cfg, raw)); err != nil { return 1 } return 0 diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index 3c7b59da..9a9321bc 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -1,9 +1,11 @@ package media import ( + "bytes" "context" "fmt" "io" + "net/http/httputil" "github.com/go-gst/go-gst/gst" "stream.place/streamplace/pkg/config" @@ -51,6 +53,29 @@ type IngestWorkerConfig struct { // of stdin, so main is out of the media path and the worker keeps ingesting // across a main restart. 0 → read media from stdin. InputFD int `json:"input_fd,omitempty"` + + // Prebuf is the HTTP body bytes main had already read past the request headers + // when it hijacked the push connection — prepended to the fd stream so none are + // lost. Chunked says the body uses chunked transfer-encoding, so the worker + // de-chunks the (prebuf+fd) stream to recover the raw media. Both are unset for + // stdin / raw fd input. + Prebuf []byte `json:"prebuf,omitempty"` + Chunked bool `json:"chunked,omitempty"` +} + +// WorkerInput reconstructs the raw media stream the gst pipeline reads from the +// fd-passed push connection: prepend any bytes main already read past the headers +// (Prebuf), then de-chunk if the push used chunked transfer-encoding. For stdin +// or a raw fd (no prebuf, not chunked) it returns raw unchanged. +func WorkerInput(cfg IngestWorkerConfig, raw io.Reader) io.Reader { + r := raw + if len(cfg.Prebuf) > 0 { + r = io.MultiReader(bytes.NewReader(cfg.Prebuf), r) + } + if cfg.Chunked { + r = httputil.NewChunkedReader(r) + } + return r } // RunMKVIngestWorker is the body of the `ingest-worker` subcommand. It reads an diff --git a/pkg/media/ingest_worker_test.go b/pkg/media/ingest_worker_test.go index efb699c9..c1d73299 100644 --- a/pkg/media/ingest_worker_test.go +++ b/pkg/media/ingest_worker_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "io" "strings" "testing" @@ -18,6 +19,26 @@ import ( "stream.place/streamplace/pkg/muxl" ) +// TestWorkerInputDeframes checks the body-deframing the worker applies to the +// fd-passed push connection: prebuf (bytes main read past the headers) is +// prepended, and a chunked transfer-encoding is decoded back to the raw media. +func TestWorkerInputDeframes(t *testing.T) { + payload := []byte("the-actual-media-bytes-pretend-this-is-mkv-data") + // A textbook chunked body: one chunk then the zero terminator. + body := []byte(fmt.Sprintf("%x\r\n%s\r\n0\r\n\r\n", len(payload), payload)) + + // prebuf = the slice main already read; the rest is still on the fd. + cfg := IngestWorkerConfig{Chunked: true, Prebuf: append([]byte(nil), body[:5]...)} + got, err := io.ReadAll(WorkerInput(cfg, bytes.NewReader(body[5:]))) + require.NoError(t, err) + require.Equal(t, payload, got, "prebuf + chunked fd de-frames to the original media") + + // Raw (no prebuf, not chunked) passes through unchanged. + got, err = io.ReadAll(WorkerInput(IngestWorkerConfig{}, bytes.NewReader(payload))) + require.NoError(t, err) + require.Equal(t, payload, got) +} + // makeH264AACMKV builds a clean, single-track, streamable H264+AAC MKV from an // H264+Opus MP4 fixture (video passed through, audio transcoded Opus→AAC). The // repo's only AAC fixture (sample-stream.mkv) carries four audio tracks, which -- 2.51.2