Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
11 kB · 300 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301package media
import ( "bufio" "bytes" "context" "crypto/ecdsa" "encoding/json" "errors" "fmt" "io" "os" "os/exec" "sync" "time"
"stream.place/streamplace/pkg/crypto/signers" "stream.place/streamplace/pkg/ingestframe" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/spmetrics")
// ingestWorkerWatchdog bounds how long an isolated worker may go without// producing a segment frame before it's presumed wedged (a native pipeline gst// can't drain — e.g. a pathological stream) and killed. A healthy stream emits a// segment every GoP (~1–2s), so this is generous. Var (not const) so tests can// shorten it.var ingestWorkerWatchdog = 30 * time.Second
// MP4IngestIsolated is the process-isolated counterpart to MP4Ingest. Instead of// running the demux + sign pipeline in this process — where a native gst fault,// OOM, or deadlock would take the whole node down — it spawns a dedicated// `ingest-worker` subprocess that owns the pipeline and streams signed canonical// .m4s segments back. This process only reads frames and runs ValidateMP4// (memory-safe Go + wasm), so a single failing stream can at worst kill its own// worker; the node survives.//// Per the locked design the worker signs everything, so main hands it the// streamer key + cert + a once-built manifest over a dedicated config fd (kept// off argv/env). See buildWorkerConfig for the interim key-custody note.func (mm *MediaManager) MP4IngestIsolated(ctx context.Context, input io.Reader, ms MediaSigner) error { cfg, err := mm.buildWorkerConfig(ctx, ms) if err != nil { return err } cfgJSON, err := json.Marshal(cfg) if err != nil { return fmt.Errorf("marshal worker config: %w", err) }
// Debug recording now happens inside the worker (cfg.Record), uniformly across // the isolated paths — main stays out of the data path here too.
exe, err := os.Executable() if err != nil { return fmt.Errorf("locate self: %w", err) } ctx, cancel := context.WithCancel(ctx) defer cancel()
// The streamer DID rides argv (public, non-sensitive) so a worker is // identifiable in a process listing; key material stays on fd 3. cmd := exec.CommandContext(ctx, exe, "ingest-worker", cfg.StreamerDID)
// Dedicated pipes: fd 3 carries the config in, fd 4 carries the frame stream // out. Keeping frames off stdout means nothing the worker (or gst, or the // self-test) writes to stdout/stderr can corrupt them; those stay plain logs. cfgR, cfgW, err := os.Pipe() if err != nil { return fmt.Errorf("config pipe: %w", err) } defer cfgW.Close() framesR, framesW, err := os.Pipe() if err != nil { cfgR.Close() return fmt.Errorf("frames pipe: %w", err) } defer framesR.Close() cmd.ExtraFiles = []*os.File{cfgR, framesW} // → child fd 3, fd 4
stdin, err := cmd.StdinPipe() if err != nil { cfgR.Close() framesW.Close() return err } stdout, err := cmd.StdoutPipe() if err != nil { cfgR.Close() framesW.Close() return err } stderr, err := cmd.StderrPipe() if err != nil { cfgR.Close() framesW.Close() return err }
if err := cmd.Start(); err != nil { cfgR.Close() framesW.Close() return fmt.Errorf("start ingest worker: %w", err) } cfgR.Close() // the child holds its own copy now framesW.Close() // ditto; the parent only reads framesR spmetrics.IngestWorkerStarts.WithLabelValues("mp4-fd").Inc()
go func() { _, _ = cfgW.Write(cfgJSON) cfgW.Close() // EOF so the worker's config read completes }()
// Pump the media to the worker; closing stdin on input EOF is the worker's // end-of-stream. Managed here (not cmd.Stdin) so a worker exit can't wedge // cmd.Wait on a still-blocked body read. go func() { defer stdin.Close() _, _ = io.Copy(stdin, input) }()
// Forward stdout + stderr to the node logger. Drain both fully before Wait. var logsWG sync.WaitGroup logsWG.Add(2) go func() { defer logsWG.Done(); streamWorkerLogs(ctx, stdout, ms.Streamer()) }() go func() { defer logsWG.Done(); streamWorkerLogs(ctx, stderr, ms.Streamer()) }()
// Watchdog: a worker that stops producing frames is presumed wedged and // killed (cancel → CommandContext SIGKILLs it), so a single bad stream can't // hang its session forever. Reset on every frame. watchdog := time.AfterFunc(ingestWorkerWatchdog, func() { log.Warn(ctx, "ingest worker watchdog fired (no frames); killing worker", "streamer", ms.Streamer(), "timeout", ingestWorkerWatchdog) cancel() }) defer watchdog.Stop()
// Ban / key revocation: the worker has no bus or model, so main watches on its // behalf and kills it (cancel → CommandContext SIGKILLs the worker), matching // the in-process HandleKeyRevocation. Without this a banned user keeps // streaming under --isolated-ingest. go mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { log.Warn(ctx, "ingest worker: ending stream", "reason", reason, "streamer", ms.Streamer()) cancel() })
// Read signed-segment frames and feed each into the normal chokepoint. sawEnd, readErr := mm.consumeWorkerFrames(ctx, ingestframe.NewReader(framesR), ms.Streamer(), mm.validateSegment(ctx), func() { watchdog.Reset(ingestWorkerWatchdog) }) logsWG.Wait() werr := cmd.Wait()
// Classify the exit for the lifecycle metrics: a crash is a read error or a // nonzero exit without a clean End; a clean End (even with a nonzero exit) is // clean. ctx cancel (main shutdown SIGKILLing the worker) isn't counted. var exitErr error if readErr != nil { exitErr = readErr } else if werr != nil && !sawEnd { exitErr = werr } recordWorkerExit("mp4-fd", exitErr, ctx.Err())
switch { case readErr != nil: return fmt.Errorf("ingest worker stream: %w", readErr) case werr != nil && !sawEnd: // A non-zero exit without a clean End frame means the worker died — that // failure is contained to the subprocess; the node is unaffected. return fmt.Errorf("ingest worker exited: %w", werr) case werr != nil: log.Warn(ctx, "ingest worker signalled clean end but exited nonzero", "streamer", ms.Streamer(), "error", werr) } return nil}
// recordWorkerExit folds a worker's terminal state into the lifecycle metrics:// a clean end vs a crash (any non-nil error). A ctx cancel (main shutting down,// the worker forcibly stopped or deliberately left detached) is not a fault we// count.func recordWorkerExit(transport string, exitErr, ctxErr error) { switch { case ctxErr != nil: // main shutdown — not a stream-level fault case exitErr == nil: spmetrics.IngestWorkerExits.WithLabelValues(transport, "clean").Inc() default: spmetrics.IngestWorkerExits.WithLabelValues(transport, "crash").Inc() }}
// consumeWorkerFrames reads framed segments from the worker and runs ValidateMP4// over each. It returns whether a clean End frame was seen and the terminal read// error: nil on a clean close (End then EOF), or io.ErrUnexpectedEOF / a desync// error when the worker died mid-frame.func (mm *MediaManager) consumeWorkerFrames(ctx context.Context, fr *ingestframe.Reader, streamer string, onSegment func([]byte) error, onProgress func()) (sawEnd bool, _ error) { for { typ, payload, err := fr.ReadFrame() if err != nil { if errors.Is(err, io.EOF) { return sawEnd, nil } return sawEnd, err } if onProgress != nil { onProgress() } switch typ { case ingestframe.Segment: if onSegment != nil { if serr := onSegment(payload); serr != nil { // Per-segment failures are logged, not fatal to the stream — a // bad GoP shouldn't tear down an otherwise-healthy ingest. log.Error(ctx, "ingest worker: segment handler failed", "streamer", streamer, "error", serr) } } case ingestframe.End: sawEnd = true case ingestframe.Error: log.Error(ctx, "ingest worker: reported error", "streamer", streamer, "error", string(payload)) } }}
// validateSegment is the onSegment handler for ingested worker frames: it folds// each signed segment into the normal ValidateMP4 chokepoint (verify → archive →// live-HLS → notify).func (mm *MediaManager) validateSegment(ctx context.Context) func([]byte) error { return func(seg []byte) error { return mm.ValidateMP4(ctx, bytes.NewReader(seg), true) }}
// streamWorkerLogs forwards the worker's stderr lines into the node logger.func streamWorkerLogs(ctx context.Context, stderr io.Reader, streamer string) { scan := bufio.NewScanner(stderr) scan.Buffer(make([]byte, 0, 64*1024), 1024*1024) for scan.Scan() { if line := scan.Text(); line != "" { log.Log(ctx, "[ingest-worker] "+line, "streamer", streamer) } }}
// buildWorkerConfig extracts the handshake the worker needs to sign on main's// behalf. INTERIM key custody: requires a software MediaSignerLocal — the// Mist-pull/RTMP path always provides one; anything else errors so the caller can// fall back to the in-process path.func (mm *MediaManager) buildWorkerConfig(ctx context.Context, ms MediaSigner) (IngestWorkerConfig, error) { local, ok := ms.(*MediaSignerLocal) if !ok { return IngestWorkerConfig{}, fmt.Errorf("isolated ingest requires a local signer, got %T", ms) } if _, ok := local.Signer.(*ecdsa.PrivateKey); !ok { return IngestWorkerConfig{}, fmt.Errorf("isolated ingest requires a software signer, got %T", local.Signer) } keyPEM, err := signers.MarshalES256KPrivateKeyPEM(local.Signer) if err != nil { return IngestWorkerConfig{}, fmt.Errorf("marshal streamer key: %w", err) } manifest, err := local.buildManifest(ctx, time.Now().UnixMilli()) if err != nil { return IngestWorkerConfig{}, fmt.Errorf("build manifest: %w", err) } cfg := IngestWorkerConfig{ StreamerDID: ms.Streamer(), KeyPEM: keyPEM, CertPEM: local.Cert, Manifest: manifest, BroadcasterHost: mm.cli.BroadcasterHost, } // Debug recording: main owns the per-stream setting (it needs the DB); the // worker carries out the recording (it owns the data path). A lookup failure // is non-fatal — just don't record. if rec, rerr := mm.shouldRecord(ctx, ms.Streamer()); rerr != nil { log.Warn(ctx, "could not read debug-recording setting; not recording", "streamer", ms.Streamer(), "error", rerr) } else if rec { cfg.Record = true cfg.DataDir = mm.cli.DataDir // The worker writes the recording, so it needs main's S3 destination too — // without it, DebugRecordingCreate inside the worker would silently fall // back to local disk under DataDir. Only sent when recording, to keep the // S3 secret out of handshakes that don't need it. if mm.cli.S3Configured() { s3cfg := mm.cli.S3Config() cfg.S3 = &s3cfg } } // Node transcode signer lets the worker complete to dual-codec itself. If it's // unavailable, the worker emits single-codec (the node doesn't re-transcode the // worker's output) — an acceptable, logged fallback rather than a hard failure. if nodeCert, nodeKeyPEM, serr := mm.transcodeSigner(); serr == nil { cfg.NodeCertPEM = nodeCert cfg.NodeKeyPEM = nodeKeyPEM } else { log.Warn(ctx, "node transcode signer unavailable; isolated worker will emit single-codec", "error", serr) } return cfg, nil}