diff --git a/docs/moq.md b/docs/moq.md index ba163073f..d1b5e0a6e 100644 --- a/docs/moq.md +++ b/docs/moq.md @@ -2,10 +2,12 @@ Streamplace nodes hand live streams to each other over [moq-lite](https://datatracker.ietf.org/doc/html/draft-lcurley-moq-lite-06) -(draft 06). The implementation is `pkg/moq`; the node's listener and the -segment-bus publisher are `pkg/api/moq.go`; the pull side is the -`websocketrep` replicator, which prefers an origin's MoQ URL over its -websocket when one is advertised. +(draft 06). The implementation is `pkg/moq`; the node's listener, the +segment-bus publisher and both directions of transfer live in the +`websocketrep` replicator (`pkg/replication/websocketrep/moq.go`). A node +pulls a stream from its origin, preferring the origin's MoQ URL over its +websocket when one is advertised, and pushes its own streams to the nodes +named in `--peers`. ## Wire convention @@ -43,10 +45,11 @@ One UDP port serves both bindings, selected by ALPN: ## Configuration -| Flag | Env | Default | Meaning | -| ------------ | ------------- | -------- | -------------------------------------------------------------------------------------------- | -| `--moq-addr` | `SP_MOQ_ADDR` | `:38443` | UDP listen address; empty disables the listener. | -| `--moq-url` | `SP_MOQ_URL` | | Override for the advertised `moqt://` URL (default: `--server-host` on `--moq-addr`'s port). | +| Flag | Env | Default | Meaning | +| ------------ | ------------- | -------- | -------------------------------------------------------------------------------------------------------------- | +| `--moq-addr` | `SP_MOQ_ADDR` | `:38443` | UDP listen address; empty disables the listener. | +| `--moq-url` | `SP_MOQ_URL` | | Override for the advertised `moqt://` URL (default: `--server-host` on `--moq-addr`'s port). | +| `--peers` | `SP_PEERS` | | Nodes to push this node's live streams to: each an `https://` node origin or a `moqt://` URL, comma-separated. | The listener terminates TLS itself, even on a node behind an HTTPS proxy: @@ -66,3 +69,38 @@ The advertised URL travels two ways, like the websocket URL: - the `moq_url` column of the station's shared `broadcast_origins` table, written by the ingest node on every local segment, so nodes sharing a statedb find each other without a relay. + +## Push + +A node that cannot be reached, one beside a streamer's encoder behind NAT, +pushes instead of waiting to be pulled from. For each `--peers` entry it +keeps one session open, dialled with the publisher Role in SETUP +(draft §7.3.3), for as long as it runs. The peer answers its +`ANNOUNCE_REQUEST` with the streams the pusher is ingesting, one route per +streamer DID, as they start and stop; the peer subscribes to each one's +`source` and `renditions` tracks and validates them exactly as it would a +pull. A pushed stream is a replicated stream on the peer (`Local: false`); +while it flows the peer writes itself into the station's `broadcast_origins` +table for it, so the nodes sharing its statedb pull the stream from the peer +rather than from a pusher they cannot reach. + +An `https://` peer is asked for its current MoQ URL on every dial, through +`place.stream.broadcast.getBroadcaster`'s `moqURL`: a node behind an HTTPS +proxy serves MoQ with a self-signed certificate it rotates, so a fixed +`moqt://…?certhash=` entry would go stale. A `moqt://` entry is dialled as +given. A listener accepts any pushed stream its `--allowed-streams` would +accept over RTMP; a stream that fails that check, or a frame that fails +validation, ends the session. + +## Who records, multistreams and transcodes + +Recording a stream to S3, driving its multistream targets and transcoding +its renditions must each happen on exactly one node, however many carry the +stream. Every node with a session for a stream, whether it ingested the +segments or received them from a peer, contends for each job it is equipped +to do (S3 configured, a Livepeer gateway, always for multistream) through a +lease in the station's shared statedb (`leases`; `pkg/director/roles.go`). +The holder renews the lease every 2 seconds; a node that dies stops, and +the next contender takes the job over once the 6 second TTL has lapsed. A +node that ends a session cleanly releases its leases, so a peer takes over +at once. A node on its own statedb simply holds every lease. diff --git a/hack/lib/e2e-harness.sh b/hack/lib/e2e-harness.sh index 4687df129..8854eba89 100644 --- a/hack/lib/e2e-harness.sh +++ b/hack/lib/e2e-harness.sh @@ -7,7 +7,8 @@ # # The caller's cwd must be the repo root. On return the harness is up and its # variables are exported: SERVER_URL, SERVER2_URL (a second node that pulls -# the stream from the first over MoQ), ACCOUNT_HANDLE, ACCOUNT_DID, +# the stream from the first over MoQ), SERVER3_URL (a third node the first +# pushes the stream to over MoQ), ACCOUNT_HANDLE, ACCOUNT_DID, # ACCOUNT_PASSWORD, VIDEO_URI (the account's test VOD), E2E_FIXTURE_MP4 (the # local file the harness streams into the node), and in HTTPS mode also # SERVER_HTTPS_URL, PDS_HTTPS_URL, @@ -80,7 +81,7 @@ e2e_harness_start() { fi # shellcheck disable=SC1090 . "$E2E_ENVFILE" - export SERVER_URL SERVER2_URL ACCOUNT_HANDLE ACCOUNT_DID ACCOUNT_PASSWORD VIDEO_URI + export SERVER_URL SERVER2_URL SERVER3_URL ACCOUNT_HANDLE ACCOUNT_DID ACCOUNT_PASSWORD VIDEO_URI export E2E_FIXTURE_MP4 # only set in HTTPS mode export SERVER_HTTPS_URL PDS_HTTPS_URL E2E_PROXY_URL E2E_TLS_SPKI E2E_TLS_CA diff --git a/js/docs/src/content/docs/lex-reference/broadcast/place-stream-broadcast-getbroadcaster.md b/js/docs/src/content/docs/lex-reference/broadcast/place-stream-broadcast-getbroadcaster.md index 99c060202..0294e0529 100644 --- a/js/docs/src/content/docs/lex-reference/broadcast/place-stream-broadcast-getbroadcaster.md +++ b/js/docs/src/content/docs/lex-reference/broadcast/place-stream-broadcast-getbroadcaster.md @@ -24,11 +24,12 @@ Get information about a Streamplace broadcaster. **Schema Type:** `object` -| Name | Type | Req'd | Description | Constraints | -| ------------- | ----------------- | ----- | --------------------------------------------------------------- | ------------- | -| `broadcaster` | `string` | ✅ | DID of the Streamplace broadcaster to which this server belongs | Format: `did` | -| `server` | `string` | ❌ | DID of this particular Streamplace server | Format: `did` | -| `admins` | Array of `string` | ❌ | Array of DIDs authorized as admins | | +| Name | Type | Req'd | Description | Constraints | +| ------------- | ----------------- | ----- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------- | +| `broadcaster` | `string` | ✅ | DID of the Streamplace broadcaster to which this server belongs | Format: `did` | +| `server` | `string` | ❌ | DID of this particular Streamplace server | Format: `did` | +| `admins` | Array of `string` | ❌ | Array of DIDs authorized as admins | | +| `moqURL` | `string` | ❌ | moqt:// URL of this server's Media over QUIC listener, which peers push streams to and pull them from; a certhash query parameter pins the server certificate (SHA-256, hex) when it is self-signed. Absent when the listener is off. | Format: `uri` | --- @@ -70,6 +71,11 @@ Get information about a Streamplace broadcaster. "format": "did" }, "description": "Array of DIDs authorized as admins" + }, + "moqURL": { + "type": "string", + "format": "uri", + "description": "moqt:// URL of this server's Media over QUIC listener, which peers push streams to and pull them from; a certhash query parameter pins the server certificate (SHA-256, hex) when it is self-signed. Absent when the listener is off." } } } diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index 3e4268ca5..55410fa4d 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -4176,6 +4176,11 @@ "type": "string", "format": "did" } + }, + "moqURL": { + "type": "string", + "description": "moqt:// URL of this server's Media over QUIC listener, which peers push streams to and pull them from; a certhash query parameter pins the server certificate (SHA-256, hex) when it is self-signed. Absent when the listener is off.", + "format": "uri" } }, "required": ["broadcaster"] diff --git a/js/e2e-web/README.md b/js/e2e-web/README.md index 26af36723..0e01b4bf3 100644 --- a/js/e2e-web/README.md +++ b/js/e2e-web/README.md @@ -78,9 +78,10 @@ and checks that later deletions still run while the commit watermark advances only after successful reprocessing. `syndication` plays the test stream from the harness's second node, which pulls -it from the first over Media over QUIC (`SERVER2_URL`; see `docs/moq.md`). It is -web-only: node-to-node transfer has no mobile surface, and the mobile flows play -from the ingest node. +it from the first over Media over QUIC (`SERVER2_URL`; see `docs/moq.md`). +`push` plays it from the third node, which the first node pushes it to +(`SERVER3_URL`, the first node's `--peers`). Both are web-only: node-to-node +transfer has no mobile surface, and the mobile flows play from the ingest node. ## OAuth over real HTTPS diff --git a/js/e2e-web/flows/push.spec.ts b/js/e2e-web/flows/push.spec.ts new file mode 100644 index 000000000..668f43764 --- /dev/null +++ b/js/e2e-web/flows/push.spec.ts @@ -0,0 +1,53 @@ +import { expect, test } from "@playwright/test"; +import { pointAppAtNode } from "../server-setup"; + +// The harness runs a third node that syndicates nothing itself: the first +// node is configured to push its live streams to it (--peers, see +// pkg/cmd/e2e.go and docs/moq.md), the way a node beside a streamer's encoder +// hands a stream off to a server it cannot be reached from. Playback from +// that node is the push end to end: the pusher looking up the peer's MoQ URL, +// dialling it as a publisher, announcing the stream, the peer subscribing, +// and the MUXL segments validating and going live there. +// +// Web only: node-to-node transfer has no mobile surface. +test("push: stream plays from the node it is pushed to over MoQ", async ({ + page, +}) => { + const SERVER3_URL = process.env.SERVER3_URL; + const did = process.env.ACCOUNT_DID; + if (!SERVER3_URL || !did) { + throw new Error( + "SERVER3_URL / ACCOUNT_DID are not set — start the harness first", + ); + } + + await pointAppAtNode(page, SERVER3_URL); + + // The stream is live on the node once its first pushed segment validates. + await expect + .poll( + async () => { + const res = await fetch( + `${SERVER3_URL}/xrpc/place.stream.live.getLiveUsers?limit=50`, + ); + if (!res.ok) return 0; + const body = (await res.json()) as { streams?: unknown[] }; + return body.streams?.length ?? 0; + }, + { timeout: 90_000, intervals: [2_000] }, + ) + .toBeGreaterThan(0); + + await page.goto(`${SERVER3_URL}/${did}`); + await expect( + page.getByText("Now streaming - e2e test stream").first(), + ).toBeVisible({ timeout: 30_000 }); + const video = page.locator("video").first(); + await expect(video).toBeVisible(); + // Frames are actually arriving over the push, not just a player mounted. + await expect + .poll(() => video.evaluate((v) => (v as HTMLVideoElement).currentTime), { + timeout: 30_000, + }) + .toBeGreaterThan(0); +}); diff --git a/lexicons/place/stream/broadcast/getBroadcaster.json b/lexicons/place/stream/broadcast/getBroadcaster.json index 91d9258c7..88a03711d 100644 --- a/lexicons/place/stream/broadcast/getBroadcaster.json +++ b/lexicons/place/stream/broadcast/getBroadcaster.json @@ -33,6 +33,11 @@ "format": "did" }, "description": "Array of DIDs authorized as admins" + }, + "moqURL": { + "type": "string", + "format": "uri", + "description": "moqt:// URL of this server's Media over QUIC listener, which peers push streams to and pull them from; a certhash query parameter pins the server certificate (SHA-256, hex) when it is self-signed. Absent when the listener is off." } } } diff --git a/pkg/api/api.go b/pkg/api/api.go index 7dff34bcf..2fb84a486 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -2,7 +2,6 @@ package api import ( "context" - "crypto/tls" "encoding/json" "errors" "fmt" @@ -80,9 +79,10 @@ type StreamplaceAPI struct { // ACME, when set, supplies TLS certificates for every TLS listener and // answers HTTP-01 challenges on the redirect listener. ACME *acme.Manager - // MoqTLS is the certificate the Media over QUIC listener serves; see - // ServeMoQ. nil when --moq-addr is empty. - MoqTLS *tls.Config + // MoqURL is the moqt:// URL peers reach this node's Media over QUIC + // listener at, as place.stream.broadcast.getBroadcaster reports it; + // nil when --moq-addr is empty. + MoqURL func() string // not thread-safe yet Aliases map[string]string Bus *bus.Bus @@ -182,6 +182,7 @@ func (a *StreamplaceAPI) Handler(ctx context.Context) (http.Handler, error) { return nil, err } a.XRPCServer = xrpc.(*spxrpc.Server) + a.XRPCServer.MoqURL = a.MoqURL router := httprouter.New() // Create our middleware factory with the default settings. diff --git a/pkg/api/moq.go b/pkg/api/moq.go deleted file mode 100644 index 6f6be0f23..000000000 --- a/pkg/api/moq.go +++ /dev/null @@ -1,114 +0,0 @@ -package api - -import ( - "context" - "fmt" - "net" - "time" - - "stream.place/streamplace/pkg/bus" - "stream.place/streamplace/pkg/config" - "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/media" - "stream.place/streamplace/pkg/moq" -) - -// MoqSourceTrack is the MoQ track carrying a stream's signed source -// segments; the rendition addenda ride media.RenditionsChannel. The -// broadcast path is the streamer's DID. One group per segment, one frame -// per group: the bare canonical MUXL segment, exactly the bytes the -// websocket path ships (see docs/moq.md). -const MoqSourceTrack = "source" - -// ServeMoQ publishes every live stream's MUXL segments over Media over QUIC -// on --moq-addr, to raw QUIC and WebTransport subscribers alike. It is the -// transport peers prefer to place.stream.live.subscribeSegments. -func (a *StreamplaceAPI) ServeMoQ(ctx context.Context) error { - pc, err := net.ListenPacket("udp", a.CLI.MoqAddr) - if err != nil { - return fmt.Errorf("listening for moq: %w", err) - } - defer pc.Close() - log.Log(ctx, "moq server starting", "addr", pc.LocalAddr()) - srv := &moq.Server{TLSConfig: a.MoqTLS, Publisher: &moqPublisher{bus: a.Bus, cli: a.CLI}} - return srv.Serve(ctx, pc) -} - -// moqPublisher serves the segment bus: a subscriber to a live streamer's -// tracks gets what the bus publishes from then on, primed with the last -// few segments like the websocket path. -type moqPublisher struct { - bus *bus.Bus - cli *config.CLI -} - -func (p *moqPublisher) Track(ctx context.Context, broadcast, name string) (moq.Track, error) { - if p.cli.DisableSyndication { - return nil, moq.ErrNotFound - } - var buf int - switch name { - case MoqSourceTrack: - buf = 2 - case media.RenditionsChannel: - buf = 4 - default: - return nil, moq.ErrNotFound - } - // A stream that is not live here is refused rather than waited on; - // renditions are checked against the source, since a stream need not - // have any. - if !p.bus.HasSegments(broadcast, MoqSourceTrack) { - return nil, moq.ErrNotFound - } - ctx, cancel := context.WithCancel(context.WithoutCancel(ctx)) - t := &moqTrack{bus: p.bus, broadcast: broadcast, name: name, cancel: cancel} - t.ch = p.bus.SubscribeSegmentBuf(ctx, broadcast, name, buf) - return t, nil -} - -type moqTrack struct { - bus *bus.Bus - broadcast string - name string - ch *bus.SegChan - cancel context.CancelFunc - seq uint64 -} - -func (t *moqTrack) Info() moq.TrackInfo { - return moq.TrackInfo{MaxAge: 10 * time.Second, Timescale: 1000} -} - -// Next waits for the next published segment. A segment that is not MUXL is -// an error, which resets the subscription: this transport carries nothing -// else. -func (t *moqTrack) Next(ctx context.Context) (*moq.Group, error) { - for { - select { - case <-ctx.Done(): - return nil, ctx.Err() - case seg := <-t.ch.C: - if !seg.Published { - continue - } - if !isMuxl(seg.Muxl) { - return nil, fmt.Errorf("segment %s of %s/%s is not a MUXL segment", seg.Filepath, t.broadcast, t.name) - } - g := &moq.Group{Sequence: t.seq, Frames: []moq.Frame{{Timestamp: time.Now().UnixMilli(), Payload: seg.Muxl}}} - t.seq++ - return g, nil - } - } -} - -func (t *moqTrack) Close() { - t.cancel() - t.bus.UnsubscribeSegment(context.Background(), t.broadcast, t.name, t.ch) -} - -// isMuxl recognises a bare canonical segment: it leads with the c2pa or -// muxl uuid box, never with ftyp/moov (a presentation MP4) or anything else. -func isMuxl(b []byte) bool { - return len(b) >= 8 && string(b[4:8]) == "uuid" -} diff --git a/pkg/cmd/e2e.go b/pkg/cmd/e2e.go index be8d45de9..dd6693cc7 100644 --- a/pkg/cmd/e2e.go +++ b/pkg/cmd/e2e.go @@ -359,6 +359,29 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri return fmt.Errorf("break debug-recording sink: %w", err) } + // A third node, fed by push: the first node pushes its streams to it + // over MoQ (--peers, see docs/moq.md), the way a node beside a + // streamer's encoder hands off to a server. Its ports are picked now so + // the first node can be told where it will be; it starts after the + // first, which retries the push until it is up. + pushHTTPPort, err := freePort() + if err != nil { + return err + } + pushInternalPort, err := freePort() + if err != nil { + return err + } + pushRTMPPort, err := freePort() + if err != nil { + return err + } + pushMoqPort, err := freeUDPPort() + if err != nil { + return err + } + pushAddr := fmt.Sprintf("127.0.0.1:%d", pushHTTPPort) + nodeEnv := []string{ fmt.Sprintf("SP_HTTP_ADDR=%s", httpAddr), fmt.Sprintf("SP_HTTP_INTERNAL_ADDR=127.0.0.1:%d", internalPort), @@ -371,6 +394,7 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri fmt.Sprintf("SP_BROADCASTER_HOST=%s", broadcasterHost), fmt.Sprintf("SP_WEBSOCKET_URL=ws://%s", httpAddr), fmt.Sprintf("SP_MOQ_URL=moqt://127.0.0.1:%d", moqPort), + fmt.Sprintf("SP_PEERS=http://%s", pushAddr), } if tlsEnv != nil { nodeEnv = append(nodeEnv, tlsEnv.NodeEnv()...) @@ -438,6 +462,33 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri return fmt.Errorf("peer: %w", err) } + // The push-fed node syndicates nothing itself: the only way the stream + // reaches it is the first node pushing it. + pushDataDir, err := os.MkdirTemp("", "streamplace-e2e-push-*") + if err != nil { + return err + } + defer os.RemoveAll(pushDataDir) //nolint:errcheck + pushEnv := []string{ + fmt.Sprintf("SP_HTTP_ADDR=%s", pushAddr), + fmt.Sprintf("SP_HTTP_INTERNAL_ADDR=127.0.0.1:%d", pushInternalPort), + fmt.Sprintf("SP_RTMP_ADDR=127.0.0.1:%d", pushRTMPPort), + fmt.Sprintf("SP_MOQ_ADDR=127.0.0.1:%d", pushMoqPort), + fmt.Sprintf("SP_BROADCASTER_HOST=%s", pushAddr), + fmt.Sprintf("SP_WEBSOCKET_URL=ws://%s", pushAddr), + } + if tlsEnv != nil { + pushEnv = append(pushEnv, tlsEnv.TrustEnv()...) + } + pushCmd, err := startNode(ctx, self, env, pushDataDir, pushEnv) + if err != nil { + return fmt.Errorf("push node: %w", err) + } + defer killNode(ctx, pushCmd, out.Did) + if err := waitForNode(ctx, pushAddr); err != nil { + return fmt.Errorf("push node: %w", err) + } + // Register a stream key for the test account. priv, pub, err := spkey.GenerateStreamKeyForDID(out.Did) if err != nil { @@ -531,8 +582,8 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri // poll for SERVER_URL and then read the whole file. E2E_FIXTURE_MP4 is // the local file the ingest loop is streaming, which tests upload through // the app to get themselves a playable VOD. - vars := fmt.Sprintf("SERVER_URL=http://%s\nSERVER2_URL=http://%s\nACCOUNT_HANDLE=%s\nACCOUNT_DID=%s\nACCOUNT_PASSWORD=%s\nVIDEO_URI=%s\nE2E_FIXTURE_MP4=%s\n", - httpAddr, peerAddr, out.Handle, out.Did, password, videoURI, fixture) + vars := fmt.Sprintf("SERVER_URL=http://%s\nSERVER2_URL=http://%s\nSERVER3_URL=http://%s\nACCOUNT_HANDLE=%s\nACCOUNT_DID=%s\nACCOUNT_PASSWORD=%s\nVIDEO_URI=%s\nE2E_FIXTURE_MP4=%s\n", + httpAddr, peerAddr, pushAddr, out.Handle, out.Did, password, videoURI, fixture) if tlsEnv != nil { // The same node over HTTPS at its public name, plus what clients // need to reach and trust it (see e2e_https.go): a browser pins the diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 44f4dc915..db0196f8e 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -427,8 +427,10 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu return err } } + var wsrep *websocketrep.WebsocketReplicator if slices.Contains(cli.Replicators, config.ReplicatorWebsocket) { - replicator = websocketrep.NewWebsocketReplicator(b, mod, mm, state, moqSelfSigned) + wsrep = websocketrep.NewWebsocketReplicator(cli, b, mod, mm, state, moqSelfSigned) + replicator = wsrep } d := director.NewDirector(mm, mod, cli, b, op, state, replicator, ldb, atsync) @@ -633,22 +635,22 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu }) } - if cli.MoqAddr != "" { + // The MoQ listener is the websocket replicator's: node-to-node + // transfer in both directions runs through it (see docs/moq.md). + if cli.MoqAddr != "" && wsrep != nil { switch { case a.ACME != nil: - a.MoqTLS = a.ACME.TLSConfig() + wsrep.MoqTLS = a.ACME.TLSConfig() case cli.Secure: cert, err := tls.LoadX509KeyPair(cli.TLSCertPath, cli.TLSKeyPath) if err != nil { return fmt.Errorf("loading tls certificate for moq: %w", err) } - a.MoqTLS = &tls.Config{Certificates: []tls.Certificate{cert}} + wsrep.MoqTLS = &tls.Config{Certificates: []tls.Certificate{cert}} default: - a.MoqTLS = moqSelfSigned.TLSConfig() + wsrep.MoqTLS = moqSelfSigned.TLSConfig() } - group.Go(func() error { - return a.ServeMoQ(ctx) - }) + a.MoqURL = func() string { return websocketrep.AdvertisedMoqURL(cli, moqSelfSigned) } } group.Go(func() error { diff --git a/pkg/config/config.go b/pkg/config/config.go index 7c13392b3..513b0ae00 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -594,7 +594,7 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { }, &urfavecli.StringFlag{ Name: "peers", - Usage: `other streamplace nodes to replicate to (default: "")`, + Usage: `comma-separated nodes to push this node's live streams to over Media over QUIC, each a node's https:// origin (its MoQ URL is looked up) or a moqt:// URL (default: "")`, Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { if s == "" { return nil diff --git a/pkg/director/roles.go b/pkg/director/roles.go new file mode 100644 index 000000000..cbfaa298e --- /dev/null +++ b/pkg/director/roles.go @@ -0,0 +1,123 @@ +package director + +import ( + "context" + "time" + + "stream.place/streamplace/pkg/log" +) + +// The jobs a live stream needs doing exactly once however many nodes carry +// it: recording it to S3 for live-to-VOD, driving its multistream targets, +// transcoding its renditions. Every node with a session for the stream, +// whether it ingested the segments or received them from a peer, competes +// for each job it is equipped to do through a lease in the station's shared +// statedb. The holder renews the lease; a holder that dies stops renewing, +// and the next node to ask after roleLeaseTTL takes the job over. A node on +// its own statedb simply holds every lease. +const ( + roleRecord = "record" + roleMultistream = "multistream" + roleTranscode = "transcode" +) + +var ( + // roleLeaseTTL is how long a dead holder keeps a job from the others. + roleLeaseTTL = 6 * time.Second + // roleLeaseRenew is how often a holder renews, and a contender asks. + roleLeaseRenew = 2 * time.Second +) + +// role is one job: what taking it up and putting it down means. start gets +// a context that ends when the role is lost or the session ends. +type role struct { + name string + start func(ctx context.Context) + stop func() +} + +// startRoles enters the session into contention for the stream's jobs. +// Once per session, on the first segment of either provenance. For the +// session's first segment the director runs Start in the background and +// NewSegment right after it, so the two calls can race; sync.Once lets one +// of them do the work and the other return. +func (ss *StreamSession) startRoles(ctx context.Context, repoDID string) { + ss.roles.Do(func() { + if ss.cli.S3Configured() { + ss.contend(ctx, repoDID, role{ + name: roleRecord, + start: func(context.Context) { ss.startS3Upload(ctx, repoDID) }, + stop: func() { ss.s3Close(ctx) }, + }) + } + ss.contend(ctx, repoDID, role{ + name: roleMultistream, + start: func(rctx context.Context) { + ss.Go(ctx, func() error { return ss.HandleMultistreamTargets(rctx) }) + }, + stop: func() {}, + }) + if ss.cli.LivepeerGatewayURL != "" { + ss.contend(ctx, repoDID, role{ + name: roleTranscode, + start: func(context.Context) { ss.transcoding.Store(true) }, + stop: func() { ss.transcoding.Store(false) }, + }) + } + }) +} + +// contend runs one role's lease loop for the life of the session: take the +// lease when it is free, renew it while held, give the job up when it is +// lost, and release it on the way out so a peer takes over at once. +func (ss *StreamSession) contend(ctx context.Context, repoDID string, r role) { + name := "stream:" + r.name + ":" + repoDID + holder := ss.cli.ServerDID() + ctx = log.WithLogValues(ctx, "role", r.name) + ss.Go(ctx, func() error { + var held bool + var renewed time.Time + var cancel context.CancelFunc + drop := func() { + held = false + cancel() + r.stop() + } + defer func() { + if held { + drop() + if err := ss.statefulDB.ReleaseLease(name, holder); err != nil { + log.Warn(ctx, "could not release stream role lease", "error", err) + } + } + }() + for { + ok, err := ss.statefulDB.AcquireLease(name, holder, roleLeaseTTL) + if err != nil { + // A database we cannot reach cannot have handed the + // lease to anyone either, unless this node alone is cut + // off: keep the job until the lease would have lapsed. + log.Warn(ctx, "stream role lease check failed", "error", err) + ok = held && time.Since(renewed) < roleLeaseTTL + } else if ok { + renewed = time.Now() + } + switch { + case ok && !held: + held = true + rctx, stop := context.WithCancel(ctx) + cancel = stop + log.Log(ctx, "taking up stream role") + r.start(rctx) + case !ok && held: + log.Log(ctx, "stream role taken over by another node") + drop() + } + select { + case <-ctx.Done(): + return nil + case <-time.After(roleLeaseRenew): + } + } + }) +} diff --git a/pkg/director/roles_test.go b/pkg/director/roles_test.go new file mode 100644 index 000000000..9685f275d --- /dev/null +++ b/pkg/director/roles_test.go @@ -0,0 +1,95 @@ +package director + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/statedb" +) + +// contender is a session on one node, contending for a role, with its +// start/stop calls recorded. +type contender struct { + ss *StreamSession + cancel context.CancelFunc + events chan string +} + +func newContender(t *testing.T, state *statedb.StatefulDB, node string) *contender { + ctx, cancel := context.WithCancel(context.Background()) + g, ctx := errgroup.WithContext(ctx) + started := make(chan struct{}) + close(started) + c := &contender{ + ss: &StreamSession{cli: &config.CLI{ServerHost: node}, statefulDB: state, g: g, started: started}, + cancel: cancel, + events: make(chan string, 16), + } + c.ss.contend(ctx, "did:plc:streamer", role{ + name: "record", + start: func(context.Context) { c.events <- node + " start" }, + stop: func() { c.events <- node + " stop" }, + }) + t.Cleanup(func() { + cancel() + _ = g.Wait() + }) + return c +} + +func (c *contender) expect(t *testing.T, event string, within time.Duration) { + t.Helper() + select { + case got := <-c.events: + require.Equal(t, event, got) + case <-time.After(within): + t.Fatalf("no %q within %s", event, within) + } +} + +func (c *contender) quiet(t *testing.T, d time.Duration) { + t.Helper() + select { + case got := <-c.events: + t.Fatalf("unexpected %q", got) + case <-time.After(d): + } +} + +// Two nodes carrying one stream: the first to ask does the job and keeps +// it while it renews; when it goes away the other takes over, at once on +// a clean exit (the lease is released) and within the TTL otherwise. +func TestRoleFailsOverBetweenNodes(t *testing.T) { + roleLeaseTTL, roleLeaseRenew = 150*time.Millisecond, 30*time.Millisecond + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + state, err := statedb.MakeDB(t.Context(), &config.CLI{DBURL: ":memory:"}, nil, mod) + require.NoError(t, err) + + a := newContender(t, state, "a.example") + a.expect(t, "a.example start", time.Second) + b := newContender(t, state, "b.example") + b.quiet(t, 3*roleLeaseTTL) + + // a ends cleanly: its stop runs and b has the job before the TTL. + a.cancel() + a.expect(t, "a.example stop", time.Second) + b.expect(t, "b.example start", roleLeaseTTL) + + // b dies without releasing: its lease stays on the books until it + // lapses, and a new node gets the job then, not before. (A clean + // cancel would release it, so the dead node's lease is put back.) + b.cancel() + b.expect(t, "b.example stop", time.Second) + held, err := state.AcquireLease("stream:record:did:plc:streamer", "did:web:b.example", roleLeaseTTL) + require.NoError(t, err) + require.True(t, held) + c := newContender(t, state, "c.example") + c.quiet(t, roleLeaseTTL/2) + c.expect(t, "c.example start", 3*roleLeaseTTL) +} diff --git a/pkg/director/s3_upload.go b/pkg/director/s3_upload.go index 4e158e0e3..69f5eb3c7 100644 --- a/pkg/director/s3_upload.go +++ b/pkg/director/s3_upload.go @@ -66,14 +66,8 @@ func (ss *StreamSession) shouldRecordLivestream(ctx context.Context, repoDID str return spsettings.LivestreamRecording == nil || *spsettings.LivestreamRecording } -func (ss *StreamSession) maybeStartS3Upload(ctx context.Context, repoDID string) { - if !ss.cli.S3Configured() { - // Debug: the no-S3 dev default, so one line per stream start would be - // pure noise there — but it's flippable at runtime when an operator is - // asking "why isn't this node recording anything?". - log.Debug(ctx, "live recording skipped: S3 not configured", "repoDID", repoDID) - return - } +// startS3Upload takes up recording the stream: the record role's start. +func (ss *StreamSession) startS3Upload(ctx context.Context, repoDID string) { if !ss.shouldRecordLivestream(ctx, repoDID) { // Info, not Debug: this fires once per stream session, and it's the // answer to "why does this streamer have no recording?" — a question @@ -85,7 +79,7 @@ func (ss *StreamSession) maybeStartS3Upload(ctx context.Context, repoDID string) // live-rec/ namespaces the in-progress livestream recordings away from the // finalized VOD blobs (blobs/) and anything else in the bucket. keyPrefix := liveRecPrefix + repoDID + "/" - ss.s3Uploader = s3.NewS3Uploader(cfg, repoDID, keyPrefix, s3.DefaultCutoverEvery, ss.statefulDB) + u := s3.NewS3Uploader(cfg, repoDID, keyPrefix, s3.DefaultCutoverEvery, ss.statefulDB) // Best-effort initial resolve of the livestream URI so the very first // object is tagged. The director treats "latest livestream for repo" as // the current stream everywhere (notification blast, idle finalize), so we @@ -94,19 +88,30 @@ func (ss *StreamSession) maybeStartS3Upload(ctx context.Context, repoDID string) if ls, err := ss.mod.GetLatestLivestreamForRepo(repoDID); err != nil { log.Warn(ctx, "live recording: failed to resolve initial livestream URI; first object starts untagged", "error", err, "repoDID", repoDID) } else if ls != nil { - ss.s3Uploader.SetLivestreamURI(ls.URI) + u.SetLivestreamURI(ls.URI) } + ss.s3Mu.Lock() + ss.s3Uploader = u + ss.s3Mu.Unlock() log.Log(ctx, "S3 upload enabled", "bucket", ss.cli.S3Bucket, "endpoint", ss.cli.S3Endpoint, "repoDID", repoDID) } +// uploader is the live recording, or nil while this node is not recording. +func (ss *StreamSession) uploader() *s3.S3Uploader { + ss.s3Mu.Lock() + defer ss.s3Mu.Unlock() + return ss.s3Uploader +} + func (ss *StreamSession) s3Upload(ctx context.Context, notif *media.NewSegmentNotification) { - if ss.s3Uploader == nil { + u := ss.uploader() + if u == nil { return } ss.Go(ctx, func() error { // notif.Muxl is the bare canonical segment; it concatenates directly // (the S3 uploader synthesizes one init and prepends it per object). - return ss.s3Uploader.AddSegment(ctx, notif.Muxl) + return u.AddSegment(ctx, notif.Muxl) }) } @@ -116,20 +121,28 @@ func (ss *StreamSession) s3Upload(ctx context.Context, notif *media.NewSegmentNo // the recording is closed out promptly rather than lingering un-completed until // stream teardown. func (ss *StreamSession) s3Cutover(ctx context.Context) { - if ss.s3Uploader == nil { + u := ss.uploader() + if u == nil { return } ss.Go(ctx, func() error { - return ss.s3Uploader.Cutover(ctx) + return u.Cutover(ctx) }) } +// s3Close puts recording down, flushing the current object so finalize can +// find it: the record role's stop, whether the session ended or another +// node took the job over. func (ss *StreamSession) s3Close(ctx context.Context) { - if ss.s3Uploader == nil { + ss.s3Mu.Lock() + u := ss.s3Uploader + ss.s3Uploader = nil + ss.s3Mu.Unlock() + if u == nil { return } - // this context is already canceled, so we need a new one - err := ss.s3Uploader.Close(context.Background()) + // the session context may already be canceled, so we need a new one + err := u.Close(context.Background()) if err != nil { log.Error(ctx, "error closing S3 upload", "error", err) } diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index 3e561a06d..d2fb040e0 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -9,6 +9,7 @@ import ( "net/url" "strings" "sync" + "sync/atomic" "time" "github.com/bluesky-social/indigo/atproto/syntax" @@ -74,12 +75,12 @@ type StreamSession struct { lastLivestreamTime time.Time lastViewCountTime time.Time + s3Mu sync.Mutex s3Uploader *s3.S3Uploader - // localRole runs once this node takes up the ingest node's jobs for the - // session (recording, multistream targets): on the first local segment, - // whether that is the session's first segment or one that arrives after - // the node syndicated the stream for a while. - localRole sync.Once + // roles enters the session into the stream's jobs once; see startRoles. + roles sync.Once + // transcoding is whether this node holds the transcode role. + transcoding atomic.Bool } // bitrateMargin is the wiggle room over the configured maximum before a stream @@ -139,14 +140,10 @@ func (ss *StreamSession) Start(ctx context.Context, notif *media.NewSegmentNotif allRenditions = append(allRenditions, renditions.AudioRendition) close(ss.started) - - // Live recording and multistream targets are the ingest node's jobs - // alone: a node that merely syndicates this stream must not also write - // it to S3, or every node records the same stream and they fight over - // the finalize. Taken up on the first local segment; see startLocalRole. - if notif.Local { - ss.startLocalRole(ctx, notif.Segment.RepoDID) - } + // Live recording, multistream targets and transcoding are jobs for + // exactly one node: every node carrying the stream contends for them, + // and the leases decide (see roles.go). + ss.startRoles(ctx, notif.Segment.RepoDID) // Start background workers for status, origin, and livestream updates ss.g.Go(func() error { @@ -167,13 +164,9 @@ func (ss *StreamSession) Start(ctx context.Context, notif *media.NewSegmentNotif case <-ss.segmentChan: // reset timer case <-ctx.Done(): - // Drain all in-flight session goroutines (including the per-segment - // AddSegment senders) BEFORE closing the uploader, so no segment - // send can race the uploader's channel close. s3Close then flushes - // the final object on the uploader's own (still-live) context. - err := ss.g.Wait() - ss.s3Close(ctx) - return err + // Every in-flight session goroutine finishes here, the role + // loops that put the stream's jobs down included. + return ss.g.Wait() // case <-time.After(time.Minute * 1): case <-time.After(ss.cli.StreamSessionTimeout): log.Log(ctx, "stream session timeout, shutting down", "timeout", ss.cli.StreamSessionTimeout) @@ -192,21 +185,6 @@ func (ss *StreamSession) Start(ctx context.Context, notif *media.NewSegmentNotif } } -// startLocalRole takes up what only the ingest node does for a stream: -// recording it to S3 for live-to-VOD and driving its multistream targets. -// Once per session, on the first local segment. For the session's first -// segment the director runs Start in the background and NewSegment right -// after it, so the two calls can race; sync.Once lets one of them do the -// work and the other return. -func (ss *StreamSession) startLocalRole(ctx context.Context, repoDID string) { - ss.localRole.Do(func() { - ss.maybeStartS3Upload(ctx, repoDID) - ss.Go(ctx, func() error { - return ss.HandleMultistreamTargets(ctx) - }) - }) -} - // Execute a goroutine in the context of the stream session. Errors are // non-fatal; if you actually want to melt the universe on an error you // should panic() @@ -270,11 +248,10 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment } // A session that began with replicated segments (this node syndicated - // the stream) and now receives local ones has become the ingest node: - // take up recording and multistream targets now rather than never. - if notif.Local { - ss.startLocalRole(ctx, spseg.Creator) - } + // the stream) and then receives local ones is already in the running + // for the stream's jobs: contention starts on the first segment of + // either kind. + ss.startRoles(ctx, spseg.Creator) // stream.received is enqueued once per StreamSession when the first local media segment is accepted. if notif.Local && !ss.streamReceivedNotified { @@ -298,9 +275,8 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment // got absorbed into the recording). The moment publishing stops we complete // the current object so it's immediately finalize-able instead of lingering // un-completed (which made finalize report "no recorded S3 segments"). - if !notif.Local { - // replicated segment: the origin node records it - } else if notif.Metadata.Published { + // Both are no-ops on a node that does not hold the record role. + if notif.Metadata.Published { ss.s3Upload(ctx, notif) } else { ss.s3Cutover(ctx) @@ -352,10 +328,9 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment ss.UpdateLivestream(ctx, spseg.Creator) } ss.UpdateViewCount(ctx) - - // Transcoding is the ingest node's job, like recording: a node that - // merely syndicates the stream receives the renditions with it. - if ss.cli.LivepeerGatewayURL != "" && notif.Local { + // Transcoding is a job for the node holding the transcode role; every + // other node carrying the stream receives the renditions with it. + if ss.cli.LivepeerGatewayURL != "" && ss.transcoding.Load() { // Claimed in arrival order: the renditions publish in it however // the gateway's round trips finish. renditionTurn := ss.renditions.turn() @@ -386,8 +361,8 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment // Refresh the S3 uploader's livestream tag now that this stream's // own record is indexed (it may not have been when the uploader // started), so all objects are attributed to the right stream. - if ss.s3Uploader != nil { - ss.s3Uploader.SetLivestreamURI(livestreamModel.URI) + if u := ss.uploader(); u != nil { + u.SetLivestreamURI(livestreamModel.URI) } lsv, err := livestreamModel.ToLivestreamView() if err != nil { diff --git a/pkg/moq/announce.go b/pkg/moq/announce.go new file mode 100644 index 000000000..9164d1c46 --- /dev/null +++ b/pkg/moq/announce.go @@ -0,0 +1,209 @@ +package moq + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "slices" + "sync" +) + +// Announced is the peer's answer to an ANNOUNCE_REQUEST: the broadcasts +// under a prefix as the peer advertises and retracts them. +type Announced struct { + s *Session + st bidiStream + prefix string + + mu sync.Mutex + err error + paths []string + out chan Announcement + done chan struct{} + closed chan struct{} + once sync.Once +} + +// Announced asks the peer for the broadcasts under prefix. It does not +// wait for the answer: a refusal (a reset) surfaces from Next. +func (s *Session) Announced(ctx context.Context, prefix string) (*Announced, error) { + st, err := s.conn.OpenStreamSync(ctx) + if err != nil { + return nil, fmt.Errorf("opening announce stream: %w", err) + } + if err := writeAll(st, msg(nil).str(prefix).frame(streamAnnounce)); err != nil { + reset(st, errCancelled) + return nil, fmt.Errorf("writing announce request: %w", err) + } + a := &Announced{s: s, st: st, prefix: prefix, out: make(chan Announcement, 16), done: make(chan struct{}), closed: make(chan struct{})} + go a.read(bufio.NewReader(st)) + return a, nil +} + +// Next returns the next broadcast appearing or going away, blocking until +// one does, the peer ends the announcements (io.EOF: every route is then +// gone), the session ends, or ctx does. +func (a *Announced) Next(ctx context.Context) (Announcement, error) { + select { + case ann := <-a.out: + return ann, nil + case <-a.done: + // Drain what was queued before the end was noticed. + select { + case ann := <-a.out: + return ann, nil + default: + } + a.mu.Lock() + defer a.mu.Unlock() + return Announcement{}, a.err + case <-a.s.ctx.Done(): + return Announcement{}, a.s.Err() + case <-ctx.Done(): + return Announcement{}, ctx.Err() + } +} + +// Close ends the request; the peer sees the stream go away. +func (a *Announced) Close() { + a.once.Do(func() { + close(a.closed) + reset(a.st, errCancelled) + }) +} + +// emit hands an announcement to Next, unless nobody will call it again. +func (a *Announced) emit(ann Announcement) bool { + select { + case a.out <- ann: + return true + case <-a.closed: + return false + case <-a.s.ctx.Done(): + return false + } +} + +func (a *Announced) finish(err error) { + a.mu.Lock() + a.err = err + a.mu.Unlock() + close(a.done) +} + +// read consumes the response side: ANNOUNCE_OK, then announcements until +// the peer ends the stream. +func (a *Announced) read(br *bufio.Reader) { + m, err := readMessage(br, maxControlMessage) + if err != nil { + a.finish(fmt.Errorf("announce refused: %w", err)) + return + } + b := body{b: m} + peerHop, _ := b.varint(), b.varint() // hop id, active count + if err := b.done(); err != nil { + a.s.fail(sessionProtocolViolation, fmt.Errorf("decoding announce ok: %w", err)) + a.finish(err) + return + } + for { + typ, err := readVarint(br) + if err != nil { + if errors.Is(err, io.EOF) { + a.finish(io.EOF) + } else { + a.finish(fmt.Errorf("announce stream reset: %w", err)) + } + return + } + m, err := readMessage(br, maxControlMessage) + if err != nil { + a.s.fail(sessionProtocolViolation, fmt.Errorf("reading announcement: %w", err)) + a.finish(err) + return + } + b := body{b: m} + switch typ { + case announceStart: + suffix := b.str() + hops, err := a.hops(&b) + if err != nil { + a.s.fail(sessionProtocolViolation, err) + a.finish(err) + return + } + path := a.prefix + suffix + // A route advertised twice is a protocol violation; one that + // looped back through us is discarded, id and all + // (draft §5.1.1.1). + if slices.Contains(a.paths, path) { + err := fmt.Errorf("route %q announced twice", path) + a.s.fail(sessionProtocolViolation, err) + a.finish(err) + return + } + if slices.Contains(hops, a.s.hop) || peerHop == a.s.hop { + a.paths = append(a.paths, "") + continue + } + a.paths = append(a.paths, path) + if !a.emit(Announcement{Path: path, Active: true}) { + return + } + case announceEnd: + id := b.varint() + if err := b.done(); err != nil { + a.s.fail(sessionProtocolViolation, err) + a.finish(err) + return + } + if id >= uint64(len(a.paths)) { + err := fmt.Errorf("announce end for unknown id %d", id) + a.s.fail(sessionProtocolViolation, err) + a.finish(err) + return + } + path := a.paths[id] + a.paths[id] = "" + if path != "" && !a.emit(Announcement{Path: path}) { + return + } + case announceUpdate: + // New hops or cost for a route we already have: nothing to + // act on, but it must still decode. + b.varint() + if _, err := a.hops(&b); err != nil { + a.s.fail(sessionProtocolViolation, err) + a.finish(err) + return + } + default: + // Something newer: nothing to act on. + } + } +} + +// hops decodes the Hop Count, Hop IDs and the two Route Costs that end an +// ANNOUNCE_START or ANNOUNCE_UPDATE, checking the message is consumed. +func (a *Announced) hops(b *body) ([]uint64, error) { + n := b.varint() + if n > uint64(len(b.b)) { + return nil, fmt.Errorf("announcement claims %d hops", n) + } + hops := make([]uint64, 0, n) + for i := uint64(0); i < n; i++ { + h := b.varint() + if h != 0 && slices.Contains(hops, h) { + return nil, fmt.Errorf("hop %d repeated in announcement", h) + } + hops = append(hops, h) + } + b.varint() // warm route cost + b.varint() // cold route cost + if err := b.done(); err != nil { + return nil, fmt.Errorf("decoding announcement: %w", err) + } + return hops, nil +} diff --git a/pkg/moq/dial.go b/pkg/moq/dial.go index 547ce0979..fcde877f9 100644 --- a/pkg/moq/dial.go +++ b/pkg/moq/dial.go @@ -17,11 +17,15 @@ import ( type DialOptions struct { // TLS verifies the server. nil uses the system roots. TLS *tls.Config + // Publisher, when set, makes this a publishing session: the server + // subscribes to its tracks (a push), and the Role in SETUP says so. + // Without it the session subscribes. + Publisher Publisher } -// Dial opens a subscriber session. A moqt:// (or moq://, moql://, moqs://) -// URL dials raw QUIC with the moq-lite ALPN, carrying the URL's path and -// query in SETUP; an https:// URL dials WebTransport with moq-lite as the +// Dial opens a session. A moqt:// (or moq://, moql://, moqs://) URL dials +// raw QUIC with the moq-lite ALPN, carrying the URL's path and query in +// SETUP; an https:// URL dials WebTransport with moq-lite as the // subprotocol. The default port is 443. func Dial(ctx context.Context, rawURL string, opts DialOptions) (*Session, error) { u, err := url.Parse(rawURL) @@ -38,14 +42,24 @@ func Dial(ctx context.Context, rawURL string, opts DialOptions) (*Session, error } switch u.Scheme { case "moqt", "moq", "moql", "moqs": - return dialQUIC(ctx, u, tlsConf) + return dialQUIC(ctx, u, tlsConf, opts.Publisher) case "https": - return dialWebTransport(ctx, u, tlsConf) + return dialWebTransport(ctx, u, tlsConf, opts.Publisher) default: return nil, fmt.Errorf("unsupported moq url scheme %q", u.Scheme) } } +// setupParams is the client's SETUP: Role narrows the session to what we +// do with it (draft §7.3.3). +func setupParams(pub Publisher) map[uint64][]byte { + role := uint64(roleSubscriber) + if pub != nil { + role = rolePublisher + } + return map[uint64][]byte{setupParamRole: appendVarint(nil, role)} +} + func quicConfig() *quic.Config { return &quic.Config{ MaxIdleTimeout: 30 * time.Second, @@ -55,7 +69,7 @@ func quicConfig() *quic.Config { } } -func dialQUIC(ctx context.Context, u *url.URL, tlsConf *tls.Config) (*Session, error) { +func dialQUIC(ctx context.Context, u *url.URL, tlsConf *tls.Config, pub Publisher) (*Session, error) { port := u.Port() if port == "" { port = "443" @@ -70,15 +84,15 @@ func dialQUIC(ctx context.Context, u *url.URL, tlsConf *tls.Config) (*Session, e return nil, fmt.Errorf("server negotiated %q, want %s", proto, ALPN) } // The request target rides SETUP on a binding with no request URI - // (draft §7.3.2); Role narrows the session to what we do with it. - params := map[uint64][]byte{setupParamRole: appendVarint(nil, roleSubscriber)} + // (draft §7.3.2). + params := setupParams(pub) if path := u.RequestURI(); path != "" && path != "/" { params[setupParamPath] = []byte(path) } - return newSession(ctx, quicConn{c}, nil, setupMessage(params)) + return newSession(ctx, quicConn{c}, pub, false, setupMessage(params)) } -func dialWebTransport(ctx context.Context, u *url.URL, tlsConf *tls.Config) (*Session, error) { +func dialWebTransport(ctx context.Context, u *url.URL, tlsConf *tls.Config, pub Publisher) (*Session, error) { tlsConf.NextProtos = []string{http3.NextProtoH3} t := &webtransport.Transport{ TLSClientConfig: tlsConf, @@ -93,6 +107,5 @@ func dialWebTransport(ctx context.Context, u *url.URL, tlsConf *tls.Config) (*Se _ = sess.CloseWithError(sessionProtocolViolation, "subprotocol") return nil, fmt.Errorf("server negotiated %q, want %s", proto, ALPN) } - params := map[uint64][]byte{setupParamRole: appendVarint(nil, roleSubscriber)} - return newSession(ctx, wtConn{sess}, nil, setupMessage(params)) + return newSession(ctx, wtConn{sess}, pub, false, setupMessage(setupParams(pub))) } diff --git a/pkg/moq/moq_test.go b/pkg/moq/moq_test.go index 351843ddf..45d7f7237 100644 --- a/pkg/moq/moq_test.go +++ b/pkg/moq/moq_test.go @@ -109,14 +109,21 @@ func selfSigned(t *testing.T) (*tls.Config, *tls.Config) { &tls.Config{RootCAs: pool} } -// serve starts a Server on a loopback UDP port for the test's lifetime and -// returns its port. +// serve starts a Server for pub on a loopback UDP port for the test's +// lifetime and returns its port. func serve(t *testing.T, ctx context.Context, pub Publisher) (int, *tls.Config) { + t.Helper() + return serveWith(t, ctx, &Server{Publisher: pub}) +} + +// serveWith is serve for a Server the test configured; its TLSConfig is +// supplied here. +func serveWith(t *testing.T, ctx context.Context, srv *Server) (int, *tls.Config) { t.Helper() serverTLS, clientTLS := selfSigned(t) + srv.TLSConfig = serverTLS pc, err := net.ListenPacket("udp", "127.0.0.1:0") require.NoError(t, err) - srv := &Server{TLSConfig: serverTLS, Publisher: pub} ctx, cancel := context.WithCancel(ctx) done := make(chan error, 1) go func() { done <- srv.Serve(ctx, pc) }() diff --git a/pkg/moq/publish.go b/pkg/moq/publish.go index 00411625b..a44e8b8a7 100644 --- a/pkg/moq/publish.go +++ b/pkg/moq/publish.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "strings" "stream.place/streamplace/pkg/log" ) @@ -205,10 +206,13 @@ func (s *Session) serveTrack(br *bufio.Reader, st bidiStream) { _ = st.Close() } -// serveAnnounce advertises one route covering the whole requested prefix: -// a claim of capability, not inventory (draft §5.1.1). Whether a path -// under it names a broadcast is answered by SUBSCRIBE. The stream stays -// open, as the peer treats its end as every route going away. +// serveAnnounce answers an ANNOUNCE_REQUEST. An Announcer's broadcasts are +// advertised one route each as they come and go, so the peer can enumerate +// them (draft §3.2); any other Publisher is advertised as one route +// covering the whole requested prefix, a claim of capability rather than +// inventory, with SUBSCRIBE answering whether a path under it is live. The +// stream stays open either way, as the peer treats its end as every route +// going away. func (s *Session) serveAnnounce(br *bufio.Reader, st bidiStream) { m, err := readMessage(br, maxControlMessage) if err != nil { @@ -216,17 +220,62 @@ func (s *Session) serveAnnounce(br *bufio.Reader, st bidiStream) { return } b := body{b: m} - b.str() // prefix + prefix := b.str() if err := b.done(); err != nil { s.fail(sessionProtocolViolation, fmt.Errorf("decoding announce request: %w", err)) return } - ok := msg(nil).varint(s.hop).varint(1).frame() - start := msg(nil).str("").varint(0).varint(0).varint(0).frame(announceStart) - if err := writeAll(st, append(ok, start...)); err != nil { + ann, ok := s.pub.(Announcer) + if !ok { + ok := msg(nil).varint(s.hop).varint(1).frame() + start := msg(nil).str("").varint(0).varint(0).varint(0).frame(announceStart) + if err := writeAll(st, append(ok, start...)); err != nil { + st.CancelWrite(errInternal) + return + } + _, _ = io.Copy(io.Discard, br) + _ = st.Close() + return + } + // The peer ending its side (FIN or reset) ends the announcements. + ctx, cancel := context.WithCancel(s.ctx) + defer cancel() + go func() { + defer cancel() + _, _ = io.Copy(io.Discard, br) + }() + // Active Count 0: the live set arrives as updates, which spares + // counting it up front (draft §7.5). + if err := writeAll(st, msg(nil).varint(s.hop).varint(0).frame()); err != nil { st.CancelWrite(errInternal) return } - _, _ = io.Copy(io.Discard, br) + // Announce IDs are implicit: the n-th ANNOUNCE_START on the stream is + // id n, and ANNOUNCE_END names one (draft §5.1.1). + ids := map[string]uint64{} + var next uint64 + for a := range ann.Announce(ctx, prefix) { + var out []byte + if a.Active { + if _, dup := ids[a.Path]; dup { + continue + } + ids[a.Path] = next + next++ + // Suffix only: the peer prepends the prefix it asked for. + out = msg(nil).str(strings.TrimPrefix(a.Path, prefix)).varint(0).varint(0).varint(0).frame(announceStart) + } else { + id, ok := ids[a.Path] + if !ok { + continue + } + delete(ids, a.Path) + out = msg(nil).varint(id).frame(announceEnd) + } + if err := writeAll(st, out); err != nil { + st.CancelWrite(errInternal) + return + } + } _ = st.Close() } diff --git a/pkg/moq/push_test.go b/pkg/moq/push_test.go new file mode 100644 index 000000000..8a5cf981c --- /dev/null +++ b/pkg/moq/push_test.go @@ -0,0 +1,171 @@ +package moq + +import ( + "context" + "fmt" + "io" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// testAnnouncer is a publisher whose broadcasts come and go. +type testAnnouncer struct { + testPublisher + mu sync.Mutex + live map[string]bool + listeners []chan Announcement +} + +func (p *testAnnouncer) Announce(ctx context.Context, prefix string) <-chan Announcement { + ch := make(chan Announcement, 16) + p.mu.Lock() + for path := range p.live { + ch <- Announcement{Path: path, Active: true} + } + p.listeners = append(p.listeners, ch) + p.mu.Unlock() + go func() { + <-ctx.Done() + p.mu.Lock() + defer p.mu.Unlock() + for i, l := range p.listeners { + if l == ch { + p.listeners = append(p.listeners[:i], p.listeners[i+1:]...) + close(ch) + return + } + } + }() + return ch +} + +func (p *testAnnouncer) set(path string, active bool) { + p.mu.Lock() + defer p.mu.Unlock() + if active { + p.live[path] = true + } else { + delete(p.live, path) + } + for _, l := range p.listeners { + l <- Announcement{Path: path, Active: active} + } +} + +// A client that dials as a publisher is handed to the server's Publishers +// hook, which learns its broadcasts by ANNOUNCE and subscribes to them +// like any publisher's: the whole of a push, over both bindings. +func TestPushAnnouncesAndServes(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + for _, binding := range []string{"moqt", "https"} { + t.Run(binding, func(t *testing.T) { + tr := &testTrack{groups: groups(3), closed: make(chan struct{})} + pub := &testAnnouncer{ + testPublisher: testPublisher{tracks: map[string]*testTrack{"did:plc:a/source": tr}}, + live: map[string]bool{"did:plc:a": true}, + } + type got struct { + anns []Announcement + frames []string + err error + } + results := make(chan got, 1) + port, clientTLS := serveWith(t, ctx, &Server{Publishers: func(ctx context.Context, sess *Session) { + g := got{} + g.err = func() error { + ann, err := sess.Announced(ctx, "") + if err != nil { + return err + } + a, err := ann.Next(ctx) + if err != nil { + return err + } + g.anns = append(g.anns, a) + sub, err := sess.Subscribe(ctx, a.Path, "source") + if err != nil { + return err + } + for { + f, err := sub.Next(ctx) + if err == io.EOF { + break + } + if err != nil { + return err + } + g.frames = append(g.frames, string(f.Payload)) + } + // A broadcast ending is announced too. + a, err = ann.Next(ctx) + if err != nil { + return err + } + g.anns = append(g.anns, a) + return nil + }() + results <- g + <-ctx.Done() + }}) + + sess, err := Dial(ctx, fmt.Sprintf("%s://127.0.0.1:%d/push", binding, port), DialOptions{TLS: clientTLS, Publisher: pub}) + require.NoError(t, err) + defer sess.Close() + // The receiver drains the track; then the broadcast goes away. + select { + case <-tr.closed: + case <-time.After(5 * time.Second): + t.Fatal("receiver never subscribed to the pushed track") + } + pub.set("did:plc:a", false) + select { + case g := <-results: + require.NoError(t, g.err) + require.Equal(t, []Announcement{{Path: "did:plc:a", Active: true}, {Path: "did:plc:a"}}, g.anns) + require.Equal(t, []string{"segment-0", "segment-1", "segment-2"}, g.frames) + case <-time.After(5 * time.Second): + t.Fatal("receiver did not get the pushed broadcast") + } + }) + } +} + +// A server with no Publishers hook refuses a publishing client outright, +// as unauthorized, rather than holding a session it will never read from. +func TestPushRefusedWithoutAcceptor(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + port, clientTLS := serve(t, ctx, &testPublisher{}) + sess, err := Dial(ctx, fmt.Sprintf("moqt://127.0.0.1:%d", port), DialOptions{TLS: clientTLS, Publisher: &testPublisher{}}) + require.NoError(t, err) + defer sess.Close() + select { + case <-sess.Done(): + require.Error(t, sess.Err()) + case <-time.After(5 * time.Second): + t.Fatal("publishing session was not refused") + } +} + +// A subscriber asking a plain Publisher for announcements gets the one +// covering route, and the announcements end with the session. +func TestAnnouncedCoveringRoute(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + port, clientTLS := serve(t, ctx, &testPublisher{}) + sess, err := Dial(ctx, fmt.Sprintf("moqt://127.0.0.1:%d", port), DialOptions{TLS: clientTLS}) + require.NoError(t, err) + defer sess.Close() + ann, err := sess.Announced(ctx, "did:") + require.NoError(t, err) + a, err := ann.Next(ctx) + require.NoError(t, err) + require.Equal(t, Announcement{Path: "did:", Active: true}, a) + ann.Close() + _, err = ann.Next(ctx) + require.Error(t, err) +} diff --git a/pkg/moq/server.go b/pkg/moq/server.go index 1e13cf899..6f676fa75 100644 --- a/pkg/moq/server.go +++ b/pkg/moq/server.go @@ -22,8 +22,15 @@ type Server struct { // TLSConfig supplies the certificate; its NextProtos are replaced. TLSConfig *tls.Config Publisher Publisher + // Publishers, when set, is handed every session whose client declared + // the publisher Role: a peer pushing broadcasts here, which it + // discovers with Session.Announced and subscribes to. It runs for the + // session's life; the session is closed when it returns. Without it + // such a session is refused as unauthorized. + Publishers func(ctx context.Context, sess *Session) - wt *webtransport.Server + ctx context.Context + wt *webtransport.Server } // Serve accepts sessions on conn until ctx ends. Every live session is @@ -45,6 +52,7 @@ func (s *Server) Serve(ctx context.Context, conn net.PacketConn) error { s.wt = wt ctx, cancel := context.WithCancel(ctx) defer cancel() + s.ctx = ctx go func() { <-ctx.Done() _ = ln.Close() @@ -63,16 +71,12 @@ func (s *Server) Serve(ctx context.Context, conn net.PacketConn) error { // Off the accept loop: setup waits on the peer for stream // credit, and one slow peer must not hold up the rest. go func() { - sess, err := newSession(ctx, quicConn{c}, s.Publisher, setupMessage(nil)) + sess, err := newSession(ctx, quicConn{c}, s.Publisher, true, setupMessage(nil)) if err != nil { log.Warn(ctx, "moq: session setup failed", "remote", c.RemoteAddr(), "error", err) return } - select { - case <-ctx.Done(): - _ = sess.Close() - case <-sess.Done(): - } + s.run(sess) }() case http3.NextProtoH3: go func() { @@ -86,6 +90,34 @@ func (s *Server) Serve(ctx context.Context, conn net.PacketConn) error { } } +// run sees an accepted session out: closed when the server stops, and in +// between either served by Publisher (the peer subscribes) or, when the +// peer declared itself a publisher, handed to Publishers to subscribe to. +func (s *Server) run(sess *Session) { + go func() { + select { + case <-s.ctx.Done(): + _ = sess.Close() + case <-sess.Done(): + } + }() + select { + case <-sess.Done(): + return + case <-sess.setupDone: + } + if sess.peerRole != rolePublisher { + <-sess.Done() + return + } + if s.Publishers == nil { + sess.fail(sessionUnauthorized, errors.New("publishing is not accepted here")) + return + } + s.Publishers(sess.ctx, sess) + _ = sess.Close() +} + // upgrade turns an HTTP/3 CONNECT into a moq-lite session. The subprotocol // must have been negotiated: a client that offered none would expect the // pre-ALPN version handshake, which this package does not speak. @@ -99,7 +131,10 @@ func (s *Server) upgrade(w http.ResponseWriter, r *http.Request) { _ = sess.CloseWithError(sessionProtocolViolation, "subprotocol "+ALPN+" required") return } - if _, err := newSession(r.Context(), wtConn{sess}, s.Publisher, setupMessage(nil)); err != nil { + ms, err := newSession(r.Context(), wtConn{sess}, s.Publisher, true, setupMessage(nil)) + if err != nil { log.Warn(r.Context(), "moq: webtransport session setup failed", "remote", r.RemoteAddr, "error", err) + return } + go s.run(ms) } diff --git a/pkg/moq/session.go b/pkg/moq/session.go index b74ae170e..c85510bc8 100644 --- a/pkg/moq/session.go +++ b/pkg/moq/session.go @@ -28,6 +28,25 @@ type Publisher interface { Track(ctx context.Context, broadcast, name string) (Track, error) } +// Announcer is a Publisher that also enumerates its broadcasts, so a peer +// can discover them by ANNOUNCE instead of knowing their paths: the +// "common convention" of draft §3.2, each broadcast's exact path as its own +// route. A Publisher without it is advertised as one route covering +// whatever prefix is asked for. +type Announcer interface { + Publisher + // Announce reports the broadcasts whose path starts with prefix: those + // live now, then each one that starts or ends, until ctx ends. The + // channel is closed when ctx ends. + Announce(ctx context.Context, prefix string) <-chan Announcement +} + +// Announcement is one broadcast appearing (Active) or going away. +type Announcement struct { + Path string + Active bool +} + // Track is one subscribable series of groups. type Track interface { Info() TrackInfo @@ -71,6 +90,13 @@ type Session struct { cancel context.CancelCauseFunc // hop is this endpoint's Hop ID, stamped on announcements. hop uint64 + // server is set on an accepted session; the peer is then a client, + // which is the only endpoint that may send a Role. + server bool + // setupDone is closed once the peer's SETUP has been read; peerRole is + // the Role it declared (0, Both, when it declared none). + setupDone chan struct{} + peerRole uint64 mu sync.Mutex subs map[uint64]*Subscription @@ -84,7 +110,7 @@ const setupTimeout = 10 * time.Second // newSession starts accepting the peer's streams and sends SETUP. Both // roles run the same loops; what is accepted differs by whether pub is // set. ctx bounds the setup only; the session outlives it. -func newSession(ctx context.Context, c conn, pub Publisher, setup []byte) (*Session, error) { +func newSession(ctx context.Context, c conn, pub Publisher, server bool, setup []byte) (*Session, error) { sctx, cancel := context.WithCancelCause(c.Context()) var hop [8]byte if _, err := rand.Read(hop[:]); err != nil { @@ -92,12 +118,14 @@ func newSession(ctx context.Context, c conn, pub Publisher, setup []byte) (*Sess return nil, err } s := &Session{ - conn: c, - pub: pub, - ctx: sctx, - cancel: cancel, - hop: binary.BigEndian.Uint64(hop[:])&maxVarint | 1, - subs: map[uint64]*Subscription{}, + conn: c, + pub: pub, + ctx: sctx, + cancel: cancel, + hop: binary.BigEndian.Uint64(hop[:])&maxVarint | 1, + server: server, + setupDone: make(chan struct{}), + subs: map[uint64]*Subscription{}, } ctx, cancelSetup := context.WithTimeout(ctx, setupTimeout) defer cancelSetup() @@ -133,6 +161,48 @@ func setupMessage(params map[uint64][]byte) []byte { return m.frame(streamSetup) } +// readSetup decodes the peer's SETUP parameters. Only the Role matters +// here, and only a client may send one (draft §7.3.3); a second SETUP +// stream is one too many (draft §3.1). +func (s *Session) readSetup(m []byte) error { + select { + case <-s.setupDone: + return errors.New("second setup stream") + default: + } + b := body{b: m} + n := b.varint() + for i := uint64(0); i < n && b.err == nil; i++ { + id, size := b.varint(), b.varint() + if b.err != nil || size > uint64(len(b.b)) { + return errors.New("decoding setup: malformed parameter") + } + v := body{b: b.b[:size]} + b.b = b.b[size:] + if id != setupParamRole { + continue + } + if !s.server { + return errors.New("server sent a role") + } + role := v.varint() + if err := v.done(); err != nil { + return fmt.Errorf("decoding setup role: %w", err) + } + switch role { + case rolePublisher, roleSubscriber: + s.peerRole = role + default: + // Unknown values mean Both (draft §7.3.3). + } + } + if err := b.done(); err != nil { + return fmt.Errorf("decoding setup: %w", err) + } + close(s.setupDone) + return nil +} + // Close ends the session; every subscription fails and every served track // is closed. func (s *Session) Close() error { @@ -184,8 +254,8 @@ func (s *Session) acceptBidi() { } } -// handleUni routes a data stream: the peer's SETUP (read and, carrying no -// capability we act on, discarded) or a group for one of our subscriptions. +// handleUni routes a data stream: the peer's SETUP or a group for one of +// our subscriptions. func (s *Session) handleUni(st recvStream) { br := bufio.NewReader(st) typ, err := readVarint(br) @@ -195,8 +265,13 @@ func (s *Session) handleUni(st recvStream) { } switch typ { case streamSetup: - if _, err := readMessage(br, maxControlMessage); err != nil { + m, err := readMessage(br, maxControlMessage) + if err != nil { s.fail(sessionProtocolViolation, fmt.Errorf("reading setup: %w", err)) + return + } + if err := s.readSetup(m); err != nil { + s.fail(sessionProtocolViolation, err) } case streamGroup: m, err := readMessage(br, maxControlMessage) diff --git a/pkg/moq/wire.go b/pkg/moq/wire.go index 569c46285..7227d16c0 100644 --- a/pkg/moq/wire.go +++ b/pkg/moq/wire.go @@ -43,7 +43,12 @@ const ( setupParamRole = 0x3 ) -const roleSubscriber = 2 +// SETUP Role parameter values (draft §7.3.3): the direction the client +// intends to use the session for. +const ( + rolePublisher = 1 + roleSubscriber = 2 +) // Subscribe stream response types (publisher → subscriber). const ( @@ -53,11 +58,16 @@ const ( ) // Announce stream message types (publisher → subscriber). -const announceStart = 0x0 +const ( + announceStart = 0x0 + announceEnd = 0x1 + announceUpdate = 0x2 +) // Session error codes (CONNECTION_CLOSE). const ( sessionNoError = 0x0 + sessionUnauthorized = 0x2 sessionProtocolViolation = 0x3 ) diff --git a/pkg/placestream/broadcastgetbroadcaster.go b/pkg/placestream/broadcastgetbroadcaster.go index a85afd72e..a342f5905 100644 --- a/pkg/placestream/broadcastgetbroadcaster.go +++ b/pkg/placestream/broadcastgetbroadcaster.go @@ -18,6 +18,8 @@ type BroadcastGetBroadcaster_Output struct { Admins []string `json:"admins,omitempty"` // broadcaster: DID of the Streamplace broadcaster to which this server belongs Broadcaster string `json:"broadcaster"` + // moqURL: moqt:// URL of this server's Media over QUIC listener, which peers push streams to and pull them from; a certhash query parameter pins the server certificate (SHA-256, hex) when it is self-signed. Absent when the listener is off. + MoqURL *string `json:"moqURL,omitempty"` // server: DID of this particular Streamplace server Server *string `json:"server,omitempty"` } diff --git a/pkg/replication/websocketrep/moq.go b/pkg/replication/websocketrep/moq.go new file mode 100644 index 000000000..f48a50205 --- /dev/null +++ b/pkg/replication/websocketrep/moq.go @@ -0,0 +1,498 @@ +package websocketrep + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net" + "net/http" + "net/url" + "strings" + "sync" + "time" + + "golang.org/x/sync/errgroup" + "stream.place/streamplace/pkg/aqhttp" + "stream.place/streamplace/pkg/bus" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/media" + "stream.place/streamplace/pkg/moq" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/spmetrics" +) + +// MoqSourceTrack is the MoQ track carrying a stream's signed source +// segments; the rendition addenda ride media.RenditionsChannel. The +// broadcast path is the streamer's DID. One group per segment, one frame +// per group: the bare canonical MUXL segment, exactly the bytes the +// websocket path ships (see docs/moq.md). +const MoqSourceTrack = "source" + +// AdvertisedMoqURL is the moqt:// URL peers reach this node's Media over +// QUIC listener at: the --moq-url override or this host on --moq-addr's +// port, carrying the self-signed certificate's hash for the peer to pin; +// "" with no listener. +func AdvertisedMoqURL(cli *config.CLI, cert *moq.SelfSignedCert) string { + if cli.MoqAddr == "" { + return "" + } + var u *url.URL + if cli.MoqURL != "" { + var err error + u, err = url.Parse(cli.MoqURL) + // chill to panic, we're going to check this on boot + if err != nil { + panic("invalid moq override URL: " + cli.MoqURL) + } + } else { + host := cli.ServerHost + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + _, port, err := net.SplitHostPort(cli.MoqAddr) + if err != nil { + panic("invalid moq listen address: " + cli.MoqAddr) + } + u = &url.URL{Scheme: "moqt", Host: net.JoinHostPort(host, port)} + } + if cert != nil { + q := u.Query() + q.Set(moq.CertHashParam, cert.Hash()) + u.RawQuery = q.Encode() + } + return u.String() +} + +// serveMoQ runs the node's Media over QUIC listener on --moq-addr: every +// live stream's MUXL segments for raw QUIC and WebTransport subscribers +// alike, and the other way, peers pushing their streams here. +func (r *WebsocketReplicator) serveMoQ(ctx context.Context) error { + pc, err := net.ListenPacket("udp", r.cli.MoqAddr) + if err != nil { + return fmt.Errorf("listening for moq: %w", err) + } + defer pc.Close() + log.Log(ctx, "moq server starting", "addr", pc.LocalAddr()) + srv := &moq.Server{TLSConfig: r.MoqTLS, Publisher: r.publisher, Publishers: r.acceptPush} + return srv.Serve(ctx, pc) +} + +// moqPublisher serves the segment bus: a subscriber to a live streamer's +// tracks gets what the bus publishes from then on, primed with the last +// few segments like the websocket path. +type moqPublisher struct { + bus *bus.Bus + cli *config.CLI +} + +func (p *moqPublisher) Track(ctx context.Context, broadcast, name string) (moq.Track, error) { + if p.cli.DisableSyndication { + return nil, moq.ErrNotFound + } + var buf int + switch name { + case MoqSourceTrack: + buf = 2 + case media.RenditionsChannel: + buf = 4 + default: + return nil, moq.ErrNotFound + } + // A stream that is not live here is refused rather than waited on; + // renditions are checked against the source, since a stream need not + // have any. + if !p.bus.HasSegments(broadcast, MoqSourceTrack) { + return nil, moq.ErrNotFound + } + ctx, cancel := context.WithCancel(context.WithoutCancel(ctx)) + t := &moqTrack{bus: p.bus, broadcast: broadcast, name: name, cancel: cancel} + t.ch = p.bus.SubscribeSegmentBuf(ctx, broadcast, name, buf) + return t, nil +} + +type moqTrack struct { + bus *bus.Bus + broadcast string + name string + ch *bus.SegChan + cancel context.CancelFunc + seq uint64 +} + +func (t *moqTrack) Info() moq.TrackInfo { + return moq.TrackInfo{MaxAge: 10 * time.Second, Timescale: 1000} +} + +// Next waits for the next published segment. A segment that is not MUXL is +// an error, which resets the subscription: this transport carries nothing +// else. +func (t *moqTrack) Next(ctx context.Context) (*moq.Group, error) { + for { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case seg := <-t.ch.C: + if !seg.Published { + continue + } + if !isMuxl(seg.Muxl) { + return nil, fmt.Errorf("segment %s of %s/%s is not a MUXL segment", seg.Filepath, t.broadcast, t.name) + } + g := &moq.Group{Sequence: t.seq, Frames: []moq.Frame{{Timestamp: time.Now().UnixMilli(), Payload: seg.Muxl}}} + t.seq++ + return g, nil + } + } +} + +func (t *moqTrack) Close() { + t.cancel() + t.bus.UnsubscribeSegment(context.Background(), t.broadcast, t.name, t.ch) +} + +// isMuxl recognises a bare canonical segment: it leads with the c2pa or +// muxl uuid box, never with ftyp/moov (a presentation MP4) or anything else. +func isMuxl(b []byte) bool { + return len(b) >= 8 && string(b[4:8]) == "uuid" +} + +// pushPublisher is what a push session serves a peer: the bus, plus the +// streams this node is ingesting as its announced inventory. The listener +// keeps the plain publisher, so a browser asking it for routes gets one +// covering everything it serves, replicas included. +type pushPublisher struct { + *moqPublisher + live *liveSet +} + +func (p *pushPublisher) Announce(ctx context.Context, prefix string) <-chan moq.Announcement { + return p.live.announce(ctx, prefix) +} + +// liveSet is the streams this node is ingesting right now. A streamer +// enters on its first local segment and leaves once none has arrived for +// ttl, the stream session timeout: the moment the director ends the +// session and the bus forgets the stream. +type liveSet struct { + ttl time.Duration + + mu sync.Mutex + seen map[string]time.Time + listeners map[chan moq.Announcement]string // the prefix each asked for +} + +func newLiveSet(ttl time.Duration) *liveSet { + return &liveSet{ttl: ttl, seen: map[string]time.Time{}, listeners: map[chan moq.Announcement]string{}} +} + +// touch notes a local segment of streamer's. +func (l *liveSet) touch(streamer string) { + l.mu.Lock() + defer l.mu.Unlock() + _, known := l.seen[streamer] + l.seen[streamer] = time.Now() + if !known { + l.notify(moq.Announcement{Path: streamer, Active: true}) + } +} + +// expire retracts the streams that have gone quiet. +func (l *liveSet) expire() { + l.mu.Lock() + defer l.mu.Unlock() + for streamer, last := range l.seen { + if time.Since(last) > l.ttl { + delete(l.seen, streamer) + l.notify(moq.Announcement{Path: streamer}) + } + } +} + +// notify hands an announcement to every listener whose prefix it matches. +// A listener that is not keeping up is closed rather than left behind: its +// announce stream ends, the peer drops the session, and the next one +// starts from the current set. The caller holds mu. +func (l *liveSet) notify(a moq.Announcement) { + for ch, prefix := range l.listeners { + if !strings.HasPrefix(a.Path, prefix) { + continue + } + select { + case ch <- a: + default: + delete(l.listeners, ch) + close(ch) + } + } +} + +// announce implements moq.Announcer for the set: the live streams under +// prefix now, then changes, until ctx ends. +func (l *liveSet) announce(ctx context.Context, prefix string) <-chan moq.Announcement { + l.mu.Lock() + ch := make(chan moq.Announcement, len(l.seen)+64) + for streamer := range l.seen { + if strings.HasPrefix(streamer, prefix) { + ch <- moq.Announcement{Path: streamer, Active: true} + } + } + l.listeners[ch] = prefix + l.mu.Unlock() + go func() { + <-ctx.Done() + l.mu.Lock() + defer l.mu.Unlock() + if _, ok := l.listeners[ch]; ok { + delete(l.listeners, ch) + close(ch) + } + }() + return ch +} + +// push keeps a publishing session open to one --peers entry for as long as +// the node runs: dial, serve the peer's subscriptions until the session +// drops, back off, dial again. The peer learns which streams to subscribe +// to from the announcements, so an idle session (nothing live here) is +// kept too; it is what makes the next stream reach the peer at once. +func (r *WebsocketReplicator) push(ctx context.Context, peer string) { + ctx = log.WithLogValues(ctx, "peer", peer) + backoff := pullBackoffMin + for { + started := time.Now() + err := r.openPush(ctx, peer) + if ctx.Err() != nil { + return + } + if time.Since(started) > pullHealthyAfter { + backoff = pullBackoffMin + } + if backoff > pullBackoffMax { + backoff = pullBackoffMax + } + log.Warn(ctx, "moq push: session to peer ended, retrying", "error", err, "retry_in", backoff) + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + backoff *= 2 + } +} + +// openPush dials the peer as a publisher and holds the session until it +// ends; it always returns an error saying why. +func (r *WebsocketReplicator) openPush(ctx context.Context, peer string) error { + moqURL, err := r.resolvePeer(ctx, peer) + if err != nil { + return err + } + sess, err := r.dialMoq(ctx, moqURL, r.pushPublisher) + if err != nil { + return err + } + defer sess.Close() + log.Log(ctx, "moq push: connected to peer", "url", moqURL) + select { + case <-ctx.Done(): + return ctx.Err() + case <-sess.Done(): + return sess.Err() + } +} + +// resolvePeer turns a --peers entry into the MoQ URL to dial. A moqt:// +// URL is dialled as given. An http(s):// URL is a node's web origin, asked +// for its current MoQ URL on every dial: a node behind an HTTPS proxy +// serves MoQ with a self-signed certificate it rotates, so the hash a +// fixed URL would carry goes stale. +func (r *WebsocketReplicator) resolvePeer(ctx context.Context, peer string) (string, error) { + u, err := url.Parse(peer) + if err != nil { + return "", fmt.Errorf("could not parse peer %q: %w", peer, err) + } + switch u.Scheme { + case "moqt", "moq", "moql", "moqs": + return peer, nil + case "http", "https": + default: + return "", fmt.Errorf("peer %q: want a moqt:// or http(s):// URL", peer) + } + u.Path, u.RawQuery = "/xrpc/place.stream.broadcast.getBroadcaster", "" + req, err := http.NewRequest(http.MethodGet, u.String(), nil) + if err != nil { + return "", err + } + ctx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + res, err := aqhttp.Do(ctx, req) + if err != nil { + return "", fmt.Errorf("asking peer for its moq url: %w", err) + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + return "", fmt.Errorf("asking peer for its moq url: %s", res.Status) + } + var out placestream.BroadcastGetBroadcaster_Output + if err := json.NewDecoder(res.Body).Decode(&out); err != nil { + return "", fmt.Errorf("decoding peer's getBroadcaster: %w", err) + } + if out.MoqURL == nil || *out.MoqURL == "" { + return "", fmt.Errorf("peer %s has no moq listener", u.Host) + } + return *out.MoqURL, nil +} + +// dialMoq opens a session to a moqt:// URL, pinning the certificate its +// certhash parameter names when it has one; pub, if set, is what the +// session serves (a push), else it subscribes (a pull). +func (r *WebsocketReplicator) dialMoq(ctx context.Context, rawURL string, pub moq.Publisher) (*moq.Session, error) { + u, err := url.Parse(rawURL) + if err != nil { + return nil, fmt.Errorf("could not parse moq URL: %w", err) + } + opts := moq.DialOptions{Publisher: pub} + if q := u.Query(); q.Has(moq.CertHashParam) { + opts.TLS = moq.PinnedTLSConfig(q.Get(moq.CertHashParam)) + q.Del(moq.CertHashParam) + u.RawQuery = q.Encode() + } + dialCtx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + sess, err := moq.Dial(dialCtx, u.String(), opts) + if err != nil { + spmetrics.ReplicationConnectErrorsTotal.Inc() + return nil, fmt.Errorf("could not dial moq (%s): %w", rawURL, err) + } + return sess, nil +} + +// acceptPush is the receiving end of push: a peer has dialled in as a +// publisher, and its announcements say which streams it is ingesting. Each +// one is subscribed to and fed through the same validation as a pull; +// anything that fails validation ends the session, the wire carrying MUXL +// and nothing else. While a pushed stream flows this node also puts +// itself in the station's broadcast_origins for it, so peers sharing the +// statedb pull it from here: the pusher is typically behind NAT, and its +// own origin record names a listener nobody can reach. +func (r *WebsocketReplicator) acceptPush(ctx context.Context, sess *moq.Session) { + ann, err := sess.Announced(ctx, "") + if err != nil { + log.Warn(ctx, "moq push: could not request announcements", "error", err) + return + } + defer ann.Close() + log.Log(ctx, "moq push: peer connected") + g, gctx := errgroup.WithContext(ctx) + running := map[string]context.CancelFunc{} + for { + a, err := ann.Next(gctx) + if err != nil { + if ctx.Err() == nil { + log.Log(ctx, "moq push: peer session ended", "error", err) + } + break + } + stop, ok := running[a.Path] + if !a.Active { + if ok { + stop() + delete(running, a.Path) + } + continue + } + if ok { + continue + } + if err := r.cli.StreamIsAllowed(a.Path); err != nil { + log.Warn(ctx, "moq push: refusing pushed stream", "streamer", a.Path, "error", err) + continue + } + bctx, cancel := context.WithCancel(gctx) + running[a.Path] = cancel + streamer := a.Path + g.Go(func() error { + err := r.consume(bctx, sess, streamer) + if bctx.Err() != nil { + // Retracted by the peer, or the session is over. + return nil + } + return err + }) + if r.state != nil { + g.Go(func() error { + r.advertiseOrigin(bctx, streamer) + return nil + }) + } + } + for _, stop := range running { + stop() + } + if err := g.Wait(); err != nil { + log.Warn(ctx, "moq push: pushed stream failed, dropping the session", "error", err) + } +} + +// originRefresh is how often a pushed stream's broadcast_origins row is +// touched; peers read the table every originPollInterval. +var originRefresh = 5 * time.Second + +func (r *WebsocketReplicator) advertiseOrigin(ctx context.Context, streamer string) { + for { + if err := r.state.UpsertBroadcastOrigin(streamer, r.cli.ServerDID(), r.getMyMoqURL(), time.Now()); err != nil { + log.Warn(ctx, "moq push: could not record this node as origin", "streamer", streamer, "error", err) + } + select { + case <-ctx.Done(): + return + case <-time.After(originRefresh): + } + } +} + +// consume subscribes to one streamer's source and rendition tracks on a +// session and feeds them through validation until the session or ctx ends; +// it always returns an error saying why. +func (r *WebsocketReplicator) consume(ctx context.Context, sess *moq.Session, streamer string) error { + source, err := sess.Subscribe(ctx, streamer, MoqSourceTrack) + if err != nil { + spmetrics.ReplicationConnectErrorsTotal.Inc() + return fmt.Errorf("could not open source subscription: %w", err) + } + defer source.Close() + renditions, err := sess.Subscribe(ctx, streamer, media.RenditionsChannel) + if err != nil { + spmetrics.ReplicationConnectErrorsTotal.Inc() + return fmt.Errorf("could not open renditions subscription: %w", err) + } + defer renditions.Close() + spmetrics.ReplicationOutboundOpen.WithLabelValues(streamer).Inc() + defer spmetrics.ReplicationOutboundOpen.WithLabelValues(streamer).Dec() + g, gctx := errgroup.WithContext(ctx) + g.Go(func() error { + for { + f, err := source.Next(gctx) + if err != nil { + return fmt.Errorf("source track: %w", err) + } + if err := r.mm.ValidateMP4(context.Background(), bytes.NewReader(f.Payload), false); err != nil { + return fmt.Errorf("could not validate segment: %w", err) + } + } + }) + g.Go(func() error { + for { + f, err := renditions.Next(gctx) + if err != nil { + return fmt.Errorf("renditions track: %w", err) + } + if err := r.feedRenditions(gctx, streamer, f.Payload); err != nil { + return err + } + } + }) + return g.Wait() +} diff --git a/pkg/replication/websocketrep/moq_test.go b/pkg/replication/websocketrep/moq_test.go new file mode 100644 index 000000000..8da7d0c67 --- /dev/null +++ b/pkg/replication/websocketrep/moq_test.go @@ -0,0 +1,74 @@ +package websocketrep + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/moq" + "stream.place/streamplace/pkg/placestream" +) + +// The announced inventory is the streams with recent local segments: a +// streamer appears on its first segment, once, and is retracted when its +// segments stop; a listener joining late gets the current set first. +func TestLiveSetAnnounces(t *testing.T) { + l := newLiveSet(50 * time.Millisecond) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + early := l.announce(ctx, "") + l.touch("did:plc:a") + l.touch("did:plc:a") + late := l.announce(ctx, "did:plc:") + other := l.announce(ctx, "did:web:") + + require.Equal(t, moq.Announcement{Path: "did:plc:a", Active: true}, <-early) + require.Equal(t, moq.Announcement{Path: "did:plc:a", Active: true}, <-late) + require.Empty(t, other, "a listener only hears paths under its prefix") + + time.Sleep(80 * time.Millisecond) + l.expire() + require.Equal(t, moq.Announcement{Path: "did:plc:a"}, <-early) + require.Equal(t, moq.Announcement{Path: "did:plc:a"}, <-late) + require.Empty(t, early, "one announcement per change") + + cancel() + _, open := <-early + require.False(t, open, "the channel closes with the listener's context") +} + +// A --peers entry naming a node's web origin resolves to the MoQ URL that +// node currently reports; one with no listener is an error to retry, not a +// session to open. +func TestResolvePeer(t *testing.T) { + var moqURL string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + require.Equal(t, "/xrpc/place.stream.broadcast.getBroadcaster", req.URL.Path) + out := placestream.BroadcastGetBroadcaster_Output{Broadcaster: "did:web:peer"} + if moqURL != "" { + out.MoqURL = &moqURL + } + _ = json.NewEncoder(w).Encode(out) + })) + defer srv.Close() + r := &WebsocketReplicator{} + + _, err := r.resolvePeer(context.Background(), srv.URL) + require.ErrorContains(t, err, "no moq listener") + + moqURL = "moqt://127.0.0.1:38443?certhash=ab" + got, err := r.resolvePeer(context.Background(), srv.URL+"/ignored?x=1") + require.NoError(t, err) + require.Equal(t, moqURL, got) + + got, err = r.resolvePeer(context.Background(), "moqt://peer.example:38443?certhash=cd") + require.NoError(t, err) + require.Equal(t, "moqt://peer.example:38443?certhash=cd", got) + + _, err = r.resolvePeer(context.Background(), "ftp://peer.example") + require.Error(t, err) +} diff --git a/pkg/replication/websocketrep/websocket_replicator.go b/pkg/replication/websocketrep/websocket_replicator.go index b72432465..fe264ffdc 100644 --- a/pkg/replication/websocketrep/websocket_replicator.go +++ b/pkg/replication/websocketrep/websocket_replicator.go @@ -3,8 +3,8 @@ package websocketrep import ( "bytes" "context" + "crypto/tls" "fmt" - "net" "net/url" "strings" "sync" @@ -45,18 +45,29 @@ type WebsocketReplicator struct { // 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 } -func NewWebsocketReplicator(bus *bus.Bus, mod model.Model, mm *media.MediaManager, state *statedb.StatefulDB, moqCert *moq.SelfSignedCert) *WebsocketReplicator { +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, - mod: mod, - state: state, - conns: make(map[string]bool), - connsMutex: sync.RWMutex{}, - latest: make(map[string]latestOrigin), - mm: mm, - moqCert: moqCert, + 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)}, } } @@ -102,7 +113,6 @@ func (r *WebsocketReplicator) latestFor(streamer string) (latestOrigin, bool) { } func (r *WebsocketReplicator) Start(ctx context.Context, cli *config.CLI) error { - r.cli = cli _ = r.getMyWebsocketURL() // panic check _ = r.getMyMoqURL() // panic check r.group, ctx = errgroup.WithContext(ctx) @@ -112,7 +122,40 @@ func (r *WebsocketReplicator) Start(ctx context.Context, cli *config.CLI) error return nil }) } - return r.startBusSubscribe(ctx) + 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 @@ -320,61 +363,13 @@ func (r *WebsocketReplicator) open(ctx context.Context, view *placestream.Broadc // 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 { - u, err := url.Parse(*origin.MoqURL) + sess, err := r.dialMoq(ctx, *origin.MoqURL, nil) if err != nil { - return fmt.Errorf("could not parse origin moq URL: %w", err) - } - var opts moq.DialOptions - if q := u.Query(); q.Has(moq.CertHashParam) { - opts.TLS = moq.PinnedTLSConfig(q.Get(moq.CertHashParam)) - q.Del(moq.CertHashParam) - u.RawQuery = q.Encode() - } - dialCtx, cancel := context.WithTimeout(ctx, 15*time.Second) - sess, err := moq.Dial(dialCtx, u.String(), opts) - cancel() - if err != nil { - spmetrics.ReplicationConnectErrorsTotal.Inc() - return fmt.Errorf("could not dial moq (%s): %w", *origin.MoqURL, err) + return err } defer sess.Close() - source, err := sess.Subscribe(ctx, origin.Streamer, "source") - if err != nil { - spmetrics.ReplicationConnectErrorsTotal.Inc() - return fmt.Errorf("could not open source subscription: %w", err) - } - renditions, err := sess.Subscribe(ctx, origin.Streamer, media.RenditionsChannel) - if err != nil { - spmetrics.ReplicationConnectErrorsTotal.Inc() - return fmt.Errorf("could not open renditions subscription: %w", err) - } log.Log(ctx, "syndication: connected to origin (moq)", "origin", *origin.MoqURL) - spmetrics.ReplicationOutboundOpen.WithLabelValues(origin.Streamer).Inc() - defer spmetrics.ReplicationOutboundOpen.WithLabelValues(origin.Streamer).Dec() - g, gctx := errgroup.WithContext(ctx) - g.Go(func() error { - for { - f, err := source.Next(gctx) - if err != nil { - return fmt.Errorf("source track: %w", err) - } - if err := r.mm.ValidateMP4(context.Background(), bytes.NewReader(f.Payload), false); err != nil { - return fmt.Errorf("could not validate segment: %w", err) - } - } - }) - g.Go(func() error { - for { - f, err := renditions.Next(gctx) - if err != nil { - return fmt.Errorf("renditions track: %w", err) - } - if err := r.feedRenditions(gctx, origin.Streamer, f.Payload); err != nil { - return err - } - } - }) - return g.Wait() + return r.consume(ctx, sess, origin.Streamer) } // feedRenditions folds a rendition addendum the origin minted into this @@ -467,8 +462,15 @@ func (r *WebsocketReplicator) removeConnection(origin string) { delete(r.conns, origin) } -// we're pull-based, nothing to do here +// 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 } @@ -487,38 +489,10 @@ func (r *WebsocketReplicator) BuildOriginRecord(origin *placestream.BroadcastOri return nil } -// getMyMoqURL is the moqt:// URL peers pull this node's streams from: the -// --moq-url override or this host on --moq-addr's port, carrying the -// self-signed certificate's hash for the peer to pin; "" with no listener. +// getMyMoqURL is the moqt:// URL peers reach this node's MoQ listener at; +// "" with no listener. func (r *WebsocketReplicator) getMyMoqURL() string { - if r.cli.MoqAddr == "" { - return "" - } - var u *url.URL - if r.cli.MoqURL != "" { - var err error - u, err = url.Parse(r.cli.MoqURL) - // chill to panic, we're going to check this on boot - if err != nil { - panic("invalid moq override URL: " + r.cli.MoqURL) - } - } else { - host := r.cli.ServerHost - if h, _, err := net.SplitHostPort(host); err == nil { - host = h - } - _, port, err := net.SplitHostPort(r.cli.MoqAddr) - if err != nil { - panic("invalid moq listen address: " + r.cli.MoqAddr) - } - u = &url.URL{Scheme: "moqt", Host: net.JoinHostPort(host, port)} - } - if r.moqCert != nil { - q := u.Query() - q.Set(moq.CertHashParam, r.moqCert.Hash()) - u.RawQuery = q.Encode() - } - return u.String() + return AdvertisedMoqURL(r.cli, r.moqCert) } func (r *WebsocketReplicator) getMyWebsocketURL() *url.URL { diff --git a/pkg/s3/s3.go b/pkg/s3/s3.go index 6bad8e1a5..ffa6be84f 100644 --- a/pkg/s3/s3.go +++ b/pkg/s3/s3.go @@ -69,6 +69,11 @@ type S3Uploader struct { closeOnce sync.Once closeErr error closed atomic.Bool + // closing orders a send on segCh before Close's close of it, so a + // segment that arrives as the uploader is being closed is refused + // rather than panicking: the recorder role moves between nodes while + // segments are still flowing. + closing sync.RWMutex } // SetLivestreamURI records the livestream this stream's objects belong to. A @@ -170,12 +175,11 @@ type uploadCmd struct { } // AddSegment feeds one bare canonical MUXL segment (uuid+moof+mdat per track) -// for upload. The bytes are copied, so the caller may reuse its buffer. It is -// the caller's responsibility not to call AddSegment concurrently with Close -// (the StreamSession guarantees this by draining its goroutines before closing, -// see director.Start); the closed guard here is a best-effort backstop that -// turns a late call into an error rather than a send-on-closed-channel panic. +// for upload. The bytes are copied, so the caller may reuse its buffer. A +// call that meets Close returns an error. func (u *S3Uploader) AddSegment(ctx context.Context, data []byte) error { + u.closing.RLock() + defer u.closing.RUnlock() if u.closed.Load() { return fmt.Errorf("s3 uploader closed") } @@ -196,6 +200,8 @@ func (u *S3Uploader) AddSegment(ctx context.Context, data []byte) error { // can find the completed objects right away. A no-op if there's no current // object, and a no-op after Close. func (u *S3Uploader) Cutover(ctx context.Context) error { + u.closing.RLock() + defer u.closing.RUnlock() if u.closed.Load() { return nil } @@ -218,8 +224,10 @@ func (u *S3Uploader) Cutover(ctx context.Context) error { // symmetry. func (u *S3Uploader) Close(ctx context.Context) error { u.closeOnce.Do(func() { + u.closing.Lock() u.closed.Store(true) close(u.segCh) + u.closing.Unlock() if uploadErr := <-u.done; uploadErr != nil { u.closeErr = fmt.Errorf("error uploading: %w", uploadErr) } diff --git a/pkg/spxrpc/place_stream_broadcast.go b/pkg/spxrpc/place_stream_broadcast.go index f1761c799..8981e5369 100644 --- a/pkg/spxrpc/place_stream_broadcast.go +++ b/pkg/spxrpc/place_stream_broadcast.go @@ -10,9 +10,17 @@ import ( func (s *Server) handlePlaceStreamBroadcastGetBroadcaster(ctx context.Context) (*placestreamtypes.BroadcastGetBroadcaster_Output, error) { broadcaster := fmt.Sprintf("did:web:%s", s.cli.BroadcasterHost) server := fmt.Sprintf("did:web:%s", s.cli.ServerHost) - return &placestreamtypes.BroadcastGetBroadcaster_Output{ + out := &placestreamtypes.BroadcastGetBroadcaster_Output{ Broadcaster: broadcaster, Server: &server, Admins: s.cli.AdminDIDs, - }, nil + } + // Asked for on every dial by a peer pushing streams here (--peers): + // the URL carries a self-signed certificate's hash, which rotates. + if s.MoqURL != nil { + if u := s.MoqURL(); u != "" { + out.MoqURL = &u + } + } + return out, nil } diff --git a/pkg/spxrpc/spxrpc.go b/pkg/spxrpc/spxrpc.go index 456302e0b..0953a1168 100644 --- a/pkg/spxrpc/spxrpc.go +++ b/pkg/spxrpc/spxrpc.go @@ -65,6 +65,9 @@ type Server struct { aliases map[string]string identityDirectory func() identity.Directory identityRefreshes *cache.Cache + // MoqURL reports the node's Media over QUIC URL for getBroadcaster; + // nil when the node has no listener. + MoqURL func() string } // vodCDN returns the playlist-generation CDN settings. diff --git a/pkg/statedb/lease.go b/pkg/statedb/lease.go new file mode 100644 index 000000000..34cdc1738 --- /dev/null +++ b/pkg/statedb/lease.go @@ -0,0 +1,44 @@ +package statedb + +import ( + "time" +) + +// Lease is a time-bounded exclusive claim on a job the nodes sharing this +// database must not do twice at once: recording a stream, driving its +// multistream targets. The holder renews it well inside its TTL; a holder +// that dies simply stops, and the next node to ask after the TTL takes the +// job over. Clocks are the nodes' own, so the TTL must dwarf their skew. +type Lease struct { + Name string `gorm:"column:name;primarykey"` + Holder string `gorm:"column:holder"` + // ExpiresAt is Unix milliseconds: an integer compares the same way in + // every database. + ExpiresAt int64 `gorm:"column:expires_at"` +} + +func (Lease) TableName() string { + return "leases" +} + +// AcquireLease takes the lease for holder, or renews it, when it is free, +// expired, or already theirs; it reports whether holder now has it. One +// upsert, so two nodes asking at once cannot both win. +func (state *StatefulDB) AcquireLease(name, holder string, ttl time.Duration) (bool, error) { + now := time.Now() + res := state.DB.Exec(` + INSERT INTO leases (name, holder, expires_at) VALUES (?, ?, ?) + ON CONFLICT (name) DO UPDATE SET holder = excluded.holder, expires_at = excluded.expires_at + WHERE leases.holder = excluded.holder OR leases.expires_at < ?`, + name, holder, now.Add(ttl).UnixMilli(), now.UnixMilli()) + if res.Error != nil { + return false, res.Error + } + return res.RowsAffected == 1, nil +} + +// ReleaseLease gives the lease up, if holder has it, so the next node to +// ask gets it now rather than after the TTL. +func (state *StatefulDB) ReleaseLease(name, holder string) error { + return state.DB.Exec(`DELETE FROM leases WHERE name = ? AND holder = ?`, name, holder).Error +} diff --git a/pkg/statedb/lease_test.go b/pkg/statedb/lease_test.go new file mode 100644 index 000000000..364917f04 --- /dev/null +++ b/pkg/statedb/lease_test.go @@ -0,0 +1,56 @@ +package statedb + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// A lease is held by one node at a time: a second asker is refused while +// it is live, gets it once it has expired or been released, and the first +// holder's next renewal then fails instead of silently sharing the job. +func TestLease(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + const name = "stream:record:did:plc:a" + held, err := state.AcquireLease(name, "did:web:a", time.Minute) + require.NoError(t, err) + require.True(t, held) + + held, err = state.AcquireLease(name, "did:web:b", time.Minute) + require.NoError(t, err) + require.False(t, held, "a live lease is exclusive") + + held, err = state.AcquireLease(name, "did:web:a", time.Minute) + require.NoError(t, err) + require.True(t, held, "the holder renews") + + require.NoError(t, state.ReleaseLease(name, "did:web:b")) + held, err = state.AcquireLease(name, "did:web:b", time.Minute) + require.NoError(t, err) + require.False(t, held, "only the holder can release") + + require.NoError(t, state.ReleaseLease(name, "did:web:a")) + held, err = state.AcquireLease(name, "did:web:b", time.Minute) + require.NoError(t, err) + require.True(t, held, "released, so the next asker takes it") + + held, err = state.AcquireLease(name, "did:web:a", time.Minute) + require.NoError(t, err) + require.False(t, held, "the old holder has lost it") + + // Expiry: a holder that stops renewing loses the lease to the + // next asker once its TTL has passed. + held, err = state.AcquireLease(name, "did:web:b", -time.Millisecond) + require.NoError(t, err) + require.True(t, held) + held, err = state.AcquireLease(name, "did:web:a", time.Minute) + require.NoError(t, err) + require.True(t, held, "an expired lease is up for grabs") + + // Leases are independent of each other. + held, err = state.AcquireLease("stream:transcode:did:plc:a", "did:web:b", time.Minute) + require.NoError(t, err) + require.True(t, held) + }) +} diff --git a/pkg/statedb/statedb.go b/pkg/statedb/statedb.go index 2785bd175..75d21dde7 100644 --- a/pkg/statedb/statedb.go +++ b/pkg/statedb/statedb.go @@ -86,6 +86,7 @@ var StatefulDBModels = []any{ Upload{}, DraftVideo{}, CDNLogPart{}, + Lease{}, } var NoPostgresDatabaseCode = "3D000"