Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
19 kB · 519 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520package websocketrep
import ( "bytes" "context" "crypto/tls" "fmt" "net/url" "strings" "sync" "time"
"github.com/gorilla/websocket" glex "github.com/streamplace/glex/runtime" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/appbsky" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/media" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/moq" "stream.place/streamplace/pkg/muxl" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/spmetrics" "stream.place/streamplace/pkg/statedb")
type WebsocketReplicator struct { bus *bus.Bus cli *config.CLI mod model.Model conns map[string]bool connsMutex sync.RWMutex // state is the station's shared statedb, whose broadcast_origins table // is the second (and, with a shared database, the dependable) way to // learn about origins; nil in tests. state *statedb.StatefulDB // latest is the newest origin heard per streamer (see latestOrigin). latest map[string]latestOrigin latestMutex sync.RWMutex group *errgroup.Group mm *media.MediaManager // moqCert is the self-signed certificate this node's MoQ listener // serves, whose hash goes into the advertised URL for peers to pin; // nil when the listener has a trusted certificate (or is off). moqCert *moq.SelfSignedCert // MoqTLS is the certificate the MoQ listener serves: set before Start // when --moq-addr is on (see serveMoQ). MoqTLS *tls.Config // publisher serves the segment bus to subscribers; pushPublisher is the // same with this node's live streams announced, for sessions to --peers. publisher *moqPublisher pushPublisher *pushPublisher // pushed counts the sessions currently pushing each streamer here. pushedMu sync.Mutex pushed map[string]int}
func NewWebsocketReplicator(cli *config.CLI, bus *bus.Bus, mod model.Model, mm *media.MediaManager, state *statedb.StatefulDB, moqCert *moq.SelfSignedCert) *WebsocketReplicator { publisher := &moqPublisher{bus: bus, cli: cli} return &WebsocketReplicator{ bus: bus, cli: cli, mod: mod, state: state, conns: make(map[string]bool), connsMutex: sync.RWMutex{}, latest: make(map[string]latestOrigin), mm: mm, moqCert: moqCert, publisher: publisher, pushPublisher: &pushPublisher{moqPublisher: publisher, live: newLiveSet(cli.StreamSessionTimeout)}, pushed: map[string]int{}, }}
// latestOrigin is the newest broadcast origin heard for a streamer and when// it was heard. The ingest node re-publishes its origin every 30s or so// while segments flow, so an origin nobody has refreshed for originStaleAfter// belongs to a stream that ended (or an ingest node that died) and is not// worth dialling any more.type latestOrigin struct { view *placestream.BroadcastDefs_BroadcastOriginView seen time.Time}
var ( pullBackoffMin = time.Second pullBackoffMax = 30 * time.Second // A pull whose origin nobody has refreshed for originQuietAfter is // probably a stream that ended; it is still retried (an ingest node // that comes back is picked up without anyone doing anything), just // at pullBackoffQuiet rather than pullBackoffMax. originQuietAfter = 10 * time.Minute pullBackoffQuiet = 5 * time.Minute // A pull that lasted this long counts as having worked: the next failure // starts the backoff over rather than continuing where it left off. pullHealthyAfter = 30 * time.Second // How often the shared broadcast_origins table is read, and how far // back a row still counts as an origin worth pulling from. originPollInterval = 10 * time.Second originPollWindow = 5 * time.Minute)
func (r *WebsocketReplicator) rememberOrigin(view *placestream.BroadcastDefs_BroadcastOriginView, streamer string) { r.latestMutex.Lock() defer r.latestMutex.Unlock() r.latest[streamer] = latestOrigin{view: view, seen: time.Now()}}
func (r *WebsocketReplicator) latestFor(streamer string) (latestOrigin, bool) { r.latestMutex.RLock() defer r.latestMutex.RUnlock() l, ok := r.latest[streamer] return l, ok}
func (r *WebsocketReplicator) Start(ctx context.Context, cli *config.CLI) error { _ = r.getMyWebsocketURL() // panic check _ = r.getMyMoqURL() // panic check r.group, ctx = errgroup.WithContext(ctx) if r.state != nil { r.group.Go(func() error { r.pollOrigins(ctx) return nil }) } if cli.MoqAddr != "" { r.group.Go(func() error { return r.serveMoQ(ctx) }) } if len(cli.Peers) > 0 { // What this node is ingesting, for the peers it pushes to: every // validated segment, like the director sees them. r.group.Go(func() error { segments := r.mm.NewSegment() ticker := time.NewTicker(time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return nil case not := <-segments: _ = r.SendSegment(ctx, not) case <-ticker.C: r.pushPublisher.live.expire() } } }) for _, peer := range cli.Peers { r.group.Go(func() error { r.push(ctx, peer) return nil }) } } r.group.Go(func() error { return r.startBusSubscribe(ctx) }) return r.group.Wait()}
// pollOrigins learns about origins from the station's shared statedb rather// than the firehose: the ingest node upserts (streamer, its server DID) on// every local segment, so the table says who is ingesting whom right now// regardless of which relays this node follows. (The origin record on the// firehose lives in the streamer's own repo, which a peer's firehose never// carries — a station whose nodes relay only each other would otherwise// never hear of any origin.)func (r *WebsocketReplicator) pollOrigins(ctx context.Context) { ticker := time.NewTicker(originPollInterval) defer ticker.Stop() for { rows, err := r.state.ListBroadcastOriginsSince(time.Now().Add(-originPollWindow)) if err != nil { log.Error(ctx, "could not list broadcast origins", "error", err) } seen := map[string]bool{} for _, row := range rows { if seen[row.StreamerRepoDID] { continue // newest row per streamer wins } seen[row.StreamerRepoDID] = true view := r.originViewForRow(row) if err := r.handleOriginMessage(ctx, view); err != nil { log.Error(ctx, "could not handle broadcast origin row", "streamer", row.StreamerRepoDID, "server", row.ServerDID, "error", err) } } select { case <-ctx.Done(): return case <-ticker.C: } }}
// originViewForRow shapes a statedb row like the firehose's origin view, with// the websocket URL a node at that server DID advertises (same scheme rule// as our own, see getMyWebsocketURL — the station's nodes share a config).func (r *WebsocketReplicator) originViewForRow(row statedb.BroadcastOrigin) *placestream.BroadcastDefs_BroadcastOriginView { u := url.URL{Scheme: "ws", Host: strings.TrimPrefix(row.ServerDID, "did:web:"), Path: "/xrpc/place.stream.live.subscribeSegments"} if r.cli.HasHTTPS() { u.Scheme = "wss" } u.RawQuery = url.Values{"streamer": []string{row.StreamerRepoDID}}.Encode() wsURL := u.String() origin := &placestream.BroadcastOrigin{ Streamer: row.StreamerRepoDID, Server: row.ServerDID, WebsocketURL: &wsURL, UpdatedAt: row.UpdatedAt.UTC().Format(time.RFC3339), } if row.MoqURL != "" { origin.MoqURL = &row.MoqURL } return &placestream.BroadcastDefs_BroadcastOriginView{ Author: appbsky.ActorDefs_ProfileViewBasic{Did: row.StreamerRepoDID}, Record: &glex.LexiconTypeDecoder{Val: origin}, }}
func (r *WebsocketReplicator) startBusSubscribe(ctx context.Context) error { // start subscription first so we're buffering new origins busCh := r.bus.Subscribe("") originViews, err := r.mod.GetRecentBroadcastOrigins(ctx) if err != nil { return fmt.Errorf("failed to get recent broadcast origins: %w", err) } for _, view := range originViews { err = r.handleOriginMessage(ctx, &view) if err != nil { log.Error(ctx, "could not check origin", "error", err) } } log.Log(ctx, "Resumed recent broadcast origins", "count", len(originViews)) for { select { case <-ctx.Done(): return ctx.Err() case msg := <-busCh: if view, ok := msg.(*placestream.BroadcastDefs_BroadcastOriginView); ok { log.Debug(ctx, "got broadcast origin view", "view", view) err = r.handleOriginMessage(ctx, view) if err != nil { log.Error(ctx, "could not handle origin message", "error", err) } } } }}
func (r *WebsocketReplicator) handleOriginMessage(ctx context.Context, view *placestream.BroadcastDefs_BroadcastOriginView) error { origin, ok := view.Record.Val.(*placestream.BroadcastOrigin) if !ok { return fmt.Errorf("record is not a BroadcastOrigin") } ctx = log.WithLogValues(ctx, "streamer", view.Author.Did) if origin.WebsocketURL == nil { spmetrics.BroadcastOriginsTotal.WithLabelValues("no_url").Inc() return fmt.Errorf("origin has no websocket URL author=%s", view.Author.Did) } if !r.cli.ShouldSyndicate(origin.Streamer) { spmetrics.BroadcastOriginsTotal.WithLabelValues("not_syndicated").Inc() log.Debug(ctx, "not replicating streamer", "streamer", origin.Streamer) return nil } myURL := r.getMyWebsocketURL() u, err := url.Parse(*origin.WebsocketURL) if err != nil { spmetrics.BroadcastOriginsTotal.WithLabelValues("no_url").Inc() return fmt.Errorf("could not parse origin websocket URL: %w", err) } if u.Host == myURL.Host { spmetrics.BroadcastOriginsTotal.WithLabelValues("self").Inc() log.Debug(ctx, "origin websocket URL is on this node, skipping") return nil } // Remembered before the connected check: a running pull re-reads the // newest origin on every attempt (the stream may have moved nodes) and // each refresh keeps it from concluding the stream is over. r.rememberOrigin(view, origin.Streamer) if r.hasConnection(origin.Streamer) { spmetrics.BroadcastOriginsTotal.WithLabelValues("already_connected").Inc() log.Debug(ctx, "already has connection") return nil } spmetrics.BroadcastOriginsTotal.WithLabelValues("connect").Inc() log.Log(ctx, "syndicating: pulling from origin", "streamer", origin.Streamer, "origin", *origin.WebsocketURL) r.group.Go(func() error { r.pull(ctx, origin.Streamer) return nil }) return nil}
// pull keeps one streamer's segments flowing from wherever its origin is:// dial, read until the connection drops, back off, dial again, for as long// as the node runs. It never gives up: a stream that is merely taking a// while to come back is indistinguishable from one that ended, and a// periodic dial for a stream that ought to be live costs nothing. Each// attempt re-reads the newest origin for the streamer, so a stream that// moves to another ingest node is followed. Before this a failed dial was// logged once and forgotten.func (r *WebsocketReplicator) pull(ctx context.Context, streamer string) { if err := r.tryConnection(streamer); err != nil { return } defer r.removeConnection(streamer) ctx = log.WithLogValues(ctx, "streamer", streamer) backoff := pullBackoffMin attempts := 0 for { latest, ok := r.latestFor(streamer) if !ok { return } attempts++ started := time.Now() err := r.open(ctx, latest.view) if ctx.Err() != nil { return } if time.Since(started) > pullHealthyAfter { backoff = pullBackoffMin } cap := pullBackoffMax if time.Since(latest.seen) > originQuietAfter { cap = pullBackoffQuiet } if backoff > cap { backoff = cap } // Every attempt while it's fresh; once it's a slow ping, a line now // and then so the log says the stream is still being watched. if backoff < cap || attempts%10 == 0 { log.Warn(ctx, "syndication: pull from origin ended, retrying", "error", err, "attempt", attempts, "retry_in", backoff, "origin_age", time.Since(latest.seen).Round(time.Second)) } else { log.Debug(ctx, "syndication: pull from origin ended, retrying", "error", err, "attempt", attempts, "retry_in", backoff) } select { case <-ctx.Done(): return case <-time.After(backoff): } backoff *= 2 }}
// open pulls from the origin over MoQ when it advertises a MoQ URL, else// over its websocket, until the connection ends; it always returns an// error saying why.func (r *WebsocketReplicator) open(ctx context.Context, view *placestream.BroadcastDefs_BroadcastOriginView) error { origin, ok := view.Record.Val.(*placestream.BroadcastOrigin) if !ok { return fmt.Errorf("record is not a BroadcastOrigin") } if origin.MoqURL != nil { return r.openMoq(ctx, origin) } return r.openWebsocket(ctx, origin)}
// openMoq subscribes to the origin's source and rendition tracks over// Media over QUIC and feeds them through the same validation as// openWebsocket. Anything that fails validation ends the session: the// wire carries MUXL and nothing else.func (r *WebsocketReplicator) openMoq(ctx context.Context, origin *placestream.BroadcastOrigin) error { sess, err := r.dialMoq(ctx, *origin.MoqURL, nil) if err != nil { return err } defer sess.Close() log.Log(ctx, "syndication: connected to origin (moq)", "origin", *origin.MoqURL) return r.consume(ctx, sess, origin.Streamer, nil)}
// feedRenditions folds a rendition addendum the origin minted into this// node's live window and playback bus, once its signatures check out.func (r *WebsocketReplicator) feedRenditions(ctx context.Context, streamer string, addendum []byte) error { // Rendition tracks the origin minted for this stream: checked (every // track is a signed transcoded asset) and folded into our live window // as HLS variants. Not a segment, so not validated or archived as one. if _, err := muxl.RunMuxlVerify(context.Background(), bytes.NewReader(addendum)); err != nil { return fmt.Errorf("rendition addendum failed verification: %w", err) } // Only public streams are syndicated, so the ordering check that // wants the start time does not apply here. r.mm.FeedLiveRenditions(context.WithoutCancel(ctx), streamer, addendum, time.Time{}, true) // And onto this node's bus for WebRTC viewers who pick a rendition here. r.mm.PublishRenditionsForPlayback(context.WithoutCancel(ctx), streamer, addendum, true) return nil}
// openWebsocket dials the origin and feeds its segments through validation// until the connection ends; it always returns an error saying why.func (r *WebsocketReplicator) openWebsocket(ctx context.Context, origin *placestream.BroadcastOrigin) error { if origin.WebsocketURL == nil { return fmt.Errorf("origin has no websocket URL") } dialCtx, cancel := context.WithTimeout(ctx, 15*time.Second) conn, _, err := websocket.DefaultDialer.DialContext(dialCtx, *origin.WebsocketURL, nil) cancel() if err != nil { spmetrics.ReplicationConnectErrorsTotal.Inc() return fmt.Errorf("could not dial websocket (%s): %w", *origin.WebsocketURL, err) } defer conn.Close() // Drop the connection when the pull is cancelled, so ReadMessage returns. // The closer lives exactly as long as this connection: the pull loop // reconnects for as long as the node runs, and a closer waiting on the // node's context would pile up one goroutine per reconnect. connCtx, connDone := context.WithCancel(ctx) defer connDone() go func() { <-connCtx.Done() conn.Close() }() log.Log(ctx, "syndication: connected to origin", "origin", *origin.WebsocketURL) spmetrics.ReplicationOutboundOpen.WithLabelValues(origin.Streamer).Inc() defer spmetrics.ReplicationOutboundOpen.WithLabelValues(origin.Streamer).Dec() for { typ, msg, err := conn.ReadMessage() if err != nil { spmetrics.ReplicationConnectErrorsTotal.Inc() return fmt.Errorf("could not read message: %w", err) } if typ != websocket.BinaryMessage { log.Error(ctx, "expected binary message", "type", typ) return fmt.Errorf("expected binary message") } log.Debug(ctx, "received message", "type", typ, "length", len(msg)) if addendum, ok := media.UnframeRenditions(msg); ok { if err := r.feedRenditions(ctx, origin.Streamer, addendum); err != nil { log.Warn(ctx, "syndication: rendition addendum dropped", "error", err) } continue } err = r.mm.ValidateMP4(context.Background(), bytes.NewReader(msg), false) if err != nil { return fmt.Errorf("could not validate segment: %w", err) } }}
func (r *WebsocketReplicator) hasConnection(origin string) bool { r.connsMutex.RLock() defer r.connsMutex.RUnlock() return r.conns[origin]}
func (r *WebsocketReplicator) tryConnection(origin string) error { r.connsMutex.Lock() defer r.connsMutex.Unlock() if _, ok := r.conns[origin]; ok { return fmt.Errorf("connection already exists") } r.conns[origin] = true return nil}
func (r *WebsocketReplicator) removeConnection(origin string) { r.connsMutex.Lock() defer r.connsMutex.Unlock() delete(r.conns, origin)}
// SendSegment notes a stream this node is ingesting, for the peers it// pushes to; replication itself is pull-based. A stream is announced once// its first segment is on the bus: a peer subscribes the moment it hears// of a stream, and a subscription to one the bus does not have yet is// refused, the notification running ahead of the director publishing it.func (r *WebsocketReplicator) SendSegment(ctx context.Context, seg *media.NewSegmentNotification) error { if seg.Local && r.bus.HasSegments(seg.Segment.RepoDID, MoqSourceTrack) { r.pushPublisher.live.touch(seg.Segment.RepoDID) } return nil}
func (r *WebsocketReplicator) BuildOriginRecord(origin *placestream.BroadcastOrigin) error { u := r.getMyWebsocketURL() u.Path = "/xrpc/place.stream.live.subscribeSegments" u.RawQuery = url.Values{ "streamer": []string{origin.Streamer}, }.Encode()
urlStr := u.String() origin.WebsocketURL = &urlStr if moqURL := r.getMyMoqURL(); moqURL != "" { origin.MoqURL = &moqURL } return nil}
// getMyMoqURL is the moqt:// URL peers reach this node's MoQ listener at;// "" with no listener.func (r *WebsocketReplicator) getMyMoqURL() string { return AdvertisedMoqURL(r.cli, r.moqCert)}
func (r *WebsocketReplicator) getMyWebsocketURL() *url.URL { if r.cli.WebsocketURL != "" { u, err := url.Parse(r.cli.WebsocketURL) // chill to panic, we're going to check this on boot if err != nil { panic("invalid websocket override URL: " + r.cli.WebsocketURL) } return u } u := url.URL{ Scheme: "ws", Host: r.cli.ServerHost, } if r.cli.HasHTTPS() { u.Scheme = "wss" } return &u}