From d7eb7f8487ddc37e48eb8d7b5fec3ea68239ca9d Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Sun, 11 Oct 2026 15:38:55 -0700 Subject: [PATCH] Push streams to --peers over MoQ; lease the stream's jobs A node that cannot be reached (beside a streamer's encoder, behind NAT) pushes its live streams to the nodes in --peers over moq-lite: it dials with the publisher Role, announces the streams it is ingesting, and the peer subscribes to them and validates them like a pull. A peer is named by its https:// origin and asked for its current MoQ URL through place.stream.broadcast.getBroadcaster (new moqURL field), since a self-signed MoQ certificate rotates. The MoQ listener, bus publisher and both directions of transfer now live in the websocketrep replicator. Recording, multistream targets and transcoding become leases in the station's statedb: every node carrying a stream contends for the jobs it is equipped to do, renews every 2s, and a dead holder's job moves to another node once the 6s TTL lapses (at once on a clean exit). The web e2e harness starts a third node the first pushes to, and push.spec.ts plays the stream from it. --- docs/moq.md | 54 +- hack/lib/e2e-harness.sh | 5 +- .../place-stream-broadcast-getbroadcaster.md | 16 +- .../content/docs/lex-reference/openapi.json | 5 + js/e2e-web/README.md | 7 +- js/e2e-web/flows/push.spec.ts | 53 ++ .../stream/broadcast/getBroadcaster.json | 5 + pkg/api/api.go | 9 +- pkg/api/moq.go | 114 ---- pkg/cmd/e2e.go | 55 +- pkg/cmd/streamplace.go | 18 +- pkg/config/config.go | 2 +- pkg/director/roles.go | 123 +++++ pkg/director/roles_test.go | 95 ++++ pkg/director/s3_upload.go | 47 +- pkg/director/stream_session.go | 73 +-- pkg/moq/announce.go | 209 ++++++++ pkg/moq/dial.go | 37 +- pkg/moq/moq_test.go | 13 +- pkg/moq/publish.go | 67 ++- pkg/moq/push_test.go | 171 ++++++ pkg/moq/server.go | 51 +- pkg/moq/session.go | 95 +++- pkg/moq/wire.go | 14 +- pkg/placestream/broadcastgetbroadcaster.go | 2 + pkg/replication/websocketrep/moq.go | 498 ++++++++++++++++++ pkg/replication/websocketrep/moq_test.go | 74 +++ .../websocketrep/websocket_replicator.go | 164 +++--- pkg/s3/s3.go | 18 +- pkg/spxrpc/place_stream_broadcast.go | 12 +- pkg/spxrpc/spxrpc.go | 3 + pkg/statedb/lease.go | 44 ++ pkg/statedb/lease_test.go | 56 ++ pkg/statedb/statedb.go | 1 + 34 files changed, 1851 insertions(+), 359 deletions(-) create mode 100644 js/e2e-web/flows/push.spec.ts delete mode 100644 pkg/api/moq.go create mode 100644 pkg/director/roles.go create mode 100644 pkg/director/roles_test.go create mode 100644 pkg/moq/announce.go create mode 100644 pkg/moq/push_test.go create mode 100644 pkg/replication/websocketrep/moq.go create mode 100644 pkg/replication/websocketrep/moq_test.go create mode 100644 pkg/statedb/lease.go create mode 100644 pkg/statedb/lease_test.go 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" -- 2.54.0