Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
8.3 kB · 244 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245package media
import ( "bytes" "context" "fmt" "io" "net" "os" "sync" "time"
"stream.place/streamplace/pkg/ingestframe" "stream.place/streamplace/pkg/log")
// workerFrameBuffer bounds how many signed segments a worker holds while main is// disconnected — ~10 min at one segment per ~1s GoP. Beyond this the oldest are// dropped (a main outage longer than this loses the oldest tail, loudly).const workerFrameBuffer = 600
// workerDrainGrace bounds how long a worker lingers after its stream ends waiting// for main to drain the buffer. Generous enough for a main restart/upgrade; an// orphaned worker (main never returns) exits after it.const workerDrainGrace = 60 * time.Second
// FrameWriter is the worker's segment sink. Stage 1 uses a direct framed pipe// (*ingestframe.Writer); the zero-downtime path uses *frameServer, which buffers// across a disconnected main and replays on reconnect. Structurally satisfied by// *ingestframe.Writer, so RunMP4IngestWorker is agnostic to which it gets.type FrameWriter interface { Segment(seg []byte) error End() error Error(msg string) error}
type bufferedFrame struct { typ ingestframe.Type payload []byte}
// frameServer delivers worker frames to the main process over a reconnectable// transport (a per-session unix socket), buffering — bounded, drop-oldest —// whenever no client is attached. This is the heart of the zero-downtime upgrade// path: main disconnects for a restart, the worker keeps signing segments into// the buffer, and the reconnecting main drains the buffer before going live, so// segments produced during a brief restart are not lost. Drops are bounded and// counted (a main outage longer than the buffer window loses the oldest tail,// loudly, rather than growing without limit).//// Safe for concurrent push (the worker) vs attach/detach (the socket accept// loop). A push to an attached-but-dead client fails the write, auto-detaches,// and re-buffers that frame, so a hard main disconnect degrades to buffering.type frameServer struct { mu sync.Mutex pending []bufferedFrame conn net.Conn w *ingestframe.Writer maxBuf int dropped int everAttached bool}
// newFrameServer creates a server that buffers up to maxBuf frames while no// client is attached.func newFrameServer(maxBuf int) *frameServer { return &frameServer{maxBuf: maxBuf}}
func (s *frameServer) push(typ ingestframe.Type, payload []byte) { s.mu.Lock() defer s.mu.Unlock() if s.w != nil { if err := s.w.WriteFrame(typ, payload); err == nil { return } // Client gone; drop it and buffer this frame instead. s.conn, s.w = nil, nil } s.pending = append(s.pending, bufferedFrame{typ, bytes.Clone(payload)}) for len(s.pending) > s.maxBuf { s.pending = s.pending[1:] s.dropped++ }}
func (s *frameServer) Segment(seg []byte) error { s.push(ingestframe.Segment, seg); return nil }func (s *frameServer) End() error { s.push(ingestframe.End, nil); return nil }func (s *frameServer) Error(msg string) error { s.push(ingestframe.Error, []byte(msg)); return nil }func (s *frameServer) Answer(sdp string) error { s.push(ingestframe.Answer, []byte(sdp)); return nil }
// dropped reports how many buffered frames were discarded because the buffer// overflowed (main was disconnected longer than the buffer window).func (s *frameServer) droppedCount() int { s.mu.Lock() defer s.mu.Unlock() return s.dropped}
// attach binds a freshly-connected client, replaying buffered frames in order// before going live. On a replay error the client is dropped and the buffer kept// intact for the next reconnect.func (s *frameServer) attach(conn net.Conn) { s.mu.Lock() defer s.mu.Unlock() w := ingestframe.NewWriter(conn) for _, f := range s.pending { if err := w.WriteFrame(f.typ, f.payload); err != nil { return } } s.pending = nil s.conn, s.w = conn, w s.everAttached = true}
// waitDrained blocks until a client has attached and the buffer is fully written// out to it (so main has the whole stream, including the trailing End), or until// ctx is cancelled or grace elapses. The worker calls this after the stream ends// so it doesn't exit — discarding the in-memory buffer — before a reconnecting// main has drained it. grace bounds an orphaned worker whose main never returns.func (s *frameServer) waitDrained(ctx context.Context, grace time.Duration) { deadline := time.NewTimer(grace) defer deadline.Stop() tick := time.NewTicker(100 * time.Millisecond) defer tick.Stop() for { s.mu.Lock() drained := s.everAttached && len(s.pending) == 0 s.mu.Unlock() if drained { return } select { case <-ctx.Done(): return case <-deadline.C: log.Warn(ctx, "ingest worker: main never drained the frame buffer; exiting", "pending", len(s.pending)) return case <-tick.C: } }}
// closeConn closes the current client connection, giving main a clean EOF after// the trailing End — the signal that the worker is done and won't reconnect.func (s *frameServer) closeConn() { s.mu.Lock() c := s.conn s.conn, s.w = nil, nil s.mu.Unlock() if c != nil { c.Close() }}
// detachConn drops the named client if it's still the current one (a stale// connection's teardown must not clobber a newer one that already reattached).// The buffer and drop count are preserved.func (s *frameServer) detachConn(conn net.Conn) { s.mu.Lock() defer s.mu.Unlock() if s.conn == conn { s.conn, s.w = nil, nil }}
// ServeMP4IngestWorkerSocket runs the ingest worker, delivering its signed// segments to main over a per-session unix socket at cfg.SocketPath with// buffered reconnect — the zero-downtime path. It listens, serves the frame// stream (buffering across any main disconnect), runs the ingest, frames a// trailing End/Error, then lingers until main has drained the buffer before// removing the socket and returning.func ServeMP4IngestWorkerSocket(ctx context.Context, cfg IngestWorkerConfig, stdin io.Reader) error { if cfg.SocketPath == "" { return fmt.Errorf("ServeMP4IngestWorkerSocket: empty socket path") } ctx, cancel := context.WithCancel(ctx) defer cancel()
_ = os.Remove(cfg.SocketPath) // clear any stale socket from a prior worker ln, err := net.Listen("unix", cfg.SocketPath) if err != nil { return fmt.Errorf("listen %s: %w", cfg.SocketPath, err) } defer func() { ln.Close() removeWorkerFiles(cfg.SocketPath) }()
srv := newFrameServer(workerFrameBuffer) manifest := newManifestHolder(cfg.Manifest) go serveFrameSocket(ctx, ln, srv, manifest)
runErr := RunMP4IngestWorker(ctx, cfg, stdin, srv, manifest.get) if runErr != nil { _ = srv.Error(runErr.Error()) } else { _ = srv.End() }
// Don't exit (and drop the in-memory buffer) until main has the whole stream, // including the trailing End — so a brief main restart loses nothing. srv.waitDrained(ctx, workerDrainGrace) // Close the connection so main reads End then a clean EOF (worker is done). srv.closeConn() return runErr}
// serveFrameSocket accepts client connections on ln and attaches each to the// server, replacing any prior client (main reconnecting after a restart). The// reverse direction carries control frames from main: it reads them and applies// a Manifest update to the holder; any other frame is ignored. The read also// unblocks when main disconnects, at which point the server reverts to// buffering. Returns when ctx is cancelled or the listener is closed.func serveFrameSocket(ctx context.Context, ln net.Listener, s *frameServer, manifest *manifestHolder) { go func() { <-ctx.Done() ln.Close() }() for { conn, err := ln.Accept() if err != nil { return // listener closed } log.Log(ctx, "ingest worker: main attached to frame socket") s.attach(conn) go func(c net.Conn) { fr := ingestframe.NewReader(c) for { typ, payload, rerr := fr.ReadFrame() if rerr != nil { break // main disconnected (or sent garbage); revert to buffering } if typ == ingestframe.Manifest { manifest.set(payload) log.Log(ctx, "ingest worker: manifest updated by main", "bytes", len(payload)) } } s.detachConn(c) log.Log(ctx, "ingest worker: main detached from frame socket") }(conn) }}