From b2ef00862fe2c107ffbee8ae0b62146e00d08467 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Thu, 15 Oct 2026 17:54:02 -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 instead: for each --peers entry it keeps a MoQ session open, publishing the streams it is ingesting, each announced from its first segment until its session times out and served from the bus like the listener serves it. An https:// peer is asked for its current MoQ URL (getBroadcaster's new moqURL) on every dial. The listener gives each incoming session its own origin, so a pushed announcement never becomes a route it serves, and pulls each announced streamer DID through the same validation as a pull, at most 32 per session. While a pushed stream validates, the receiving node advertises itself in broadcast_origins for it, so its station pulls from it rather than from a pusher it cannot reach. Recording, multistreaming and transcoding now each run on one node per stream, through leases in the shared statedb (pkg/director/roles.go). A replicated segment more than ten minutes from now is refused, before the dedup skip, so old signed segments cannot be replayed as live. --peers with --disable-syndication is refused. --- docs/moq.md | 67 ++- 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 | 5 + pkg/cmd/e2e.go | 55 ++- pkg/cmd/streamplace.go | 11 +- pkg/config/config.go | 7 +- pkg/config/config_check_test.go | 4 + pkg/director/roles.go | 152 +++++++ pkg/director/roles_test.go | 95 +++++ pkg/director/s3_upload.go | 67 ++- pkg/director/stream_session.go | 100 +++-- pkg/localdb/segment.go | 5 +- pkg/localdb/segment_get_test.go | 27 ++ pkg/media/validate.go | 20 + pkg/placestream/broadcastgetbroadcaster.go | 2 + pkg/replication/iroh_replicator/kv.go | 5 + pkg/replication/replicator.go | 4 + pkg/replication/websocketrep/moq.go | 389 +++++++++++++++++- pkg/replication/websocketrep/moq_test.go | 142 ++++++- .../websocketrep/websocket_replicator.go | 92 +++-- pkg/s3/s3.go | 17 +- pkg/spxrpc/place_stream_broadcast.go | 12 +- pkg/spxrpc/spxrpc.go | 3 + pkg/statedb/lease.go | 45 ++ pkg/statedb/lease_test.go | 57 +++ pkg/statedb/statedb.go | 1 + 31 files changed, 1318 insertions(+), 157 deletions(-) create mode 100644 js/e2e-web/flows/push.spec.ts create mode 100644 pkg/director/roles.go create mode 100644 pkg/director/roles_test.go create mode 100644 pkg/localdb/segment_get_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 a4b708b37..0320c38b0 100644 --- a/docs/moq.md +++ b/docs/moq.md @@ -4,9 +4,11 @@ Streamplace nodes hand live streams to each other over Media over QUIC, using the official implementation: `moq.dev/moq`, the Go wrapper over the Rust `moq-ffi` that Meson builds from source into the node (see `rust/streamplace-native`). `pkg/moq` holds only what Streamplace adds, the -listener's certificate handling. The listener and the pull side live in the -`websocketrep` replicator (`pkg/replication/websocketrep/moq.go`), which -prefers an origin's MoQ URL over its websocket when one is advertised. +listener's certificate handling. The listener 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 @@ -69,10 +71,11 @@ version 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, reading its certificate from files (the data directory's `moq-tls`) that @@ -94,3 +97,53 @@ 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 for as long as it runs, publishing the streams it +is ingesting: each is announced, one broadcast per streamer DID, once its +first published segment's tracks are in the catalog, until the stream +session times out, its tracks served from the bus exactly as the listener +serves them. The peer gives every incoming session its own origin, so a +pushed announcement never becomes a route it serves; it pulls each +announced stream's source tracks from that session and validates them +exactly as it would a pull, and ignores +announcements past 32 streams or that are not DIDs. It takes no renditions +from a pusher, whose signer could be anyone: as the stream's origin it mints +its own. 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 frame that fails validation ends the session. +Validation refuses a replicated segment whose start time is more than ten +minutes from the node's clock, so a pusher cannot replay a public +streamer's old signed segments to make them appear live (the same window +applies to every replicated segment, `/api/segment` included). + +## 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 for recording, 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. + +Transcoding is contended for only by a node that is an origin for the +stream, one that ingests it or that a peer pushes it to, and that has a +Livepeer gateway: peers take a stream's renditions from its origin, so +renditions minted on a pull-only replica would reach nobody but its own +viewers. diff --git a/hack/lib/e2e-harness.sh b/hack/lib/e2e-harness.sh index 1d843eb6f..e354d8ddd 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), PDS_HTTP_URL (the PDS's # plain-HTTP listener, for checking stored records from the host), and in @@ -81,7 +82,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 PDS_HTTP_URL # Every node's log, for specs that check how something happened. export E2E_LOGFILE 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 73f18aef1..c8c512894 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -4254,6 +4254,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 5b88dda0e..c8ab37cda 100644 --- a/js/e2e-web/README.md +++ b/js/e2e-web/README.md @@ -105,9 +105,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 1030f6128..190b25d13 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -79,6 +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 + // 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 @@ -178,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/cmd/e2e.go b/pkg/cmd/e2e.go index 90cb14899..03d1a75fe 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 { @@ -532,8 +583,8 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri // the local file the ingest loop is streaming, which tests upload through // the app to get themselves a playable VOD. PDS_HTTP_URL is the PDS's own // plain-HTTP listener, for host-side checks of the records it stores. - 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\nPDS_HTTP_URL=%s\n", - httpAddr, peerAddr, out.Handle, out.Did, password, videoURI, fixture, env.PDSURL) + 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\nPDS_HTTP_URL=%s\n", + httpAddr, peerAddr, pushAddr, out.Handle, out.Did, password, videoURI, fixture, env.PDSURL) 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 ef5883a1e..6a2bbb6d7 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -429,7 +429,7 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu } var wsrep *websocketrep.WebsocketReplicator if slices.Contains(cli.Replicators, config.ReplicatorWebsocket) { - wsrep = websocketrep.NewWebsocketReplicator(b, mod, mm, state, moqSelfSigned) + wsrep = websocketrep.NewWebsocketReplicator(cli, b, mod, mm, state, moqSelfSigned) replicator = wsrep } @@ -579,6 +579,11 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu if err != nil { return err } + // Before any server builds its handler: getBroadcaster reports this + // URL, which is how a peer pushing streams here finds the listener. + if cli.MoqAddr != "" && wsrep != nil { + a.MoqURL = func() string { return websocketrep.AdvertisedMoqURL(cli, moqSelfSigned) } + } ctx = log.WithLogValues(ctx, "version", build.Version) @@ -635,8 +640,8 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu }) } - // The MoQ listener is the websocket replicator's (see docs/moq.md). It - // reads its certificate from files, which Go keeps current. + // 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: diff --git a/pkg/config/config.go b/pkg/config/config.go index 7c13392b3..059e59497 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 @@ -1520,6 +1520,11 @@ func (cli *CLI) CheckConfig() error { errs = append(errs, fmt.Errorf("--signing-key: %w", err)) } } + if cli.DisableSyndication && len(cli.Peers) > 0 { + // A node pushing to peers announces its streams, then refuses every + // subscription to them with syndication off: the peers retry forever. + errs = append(errs, fmt.Errorf("--peers pushes this node's streams, which --disable-syndication refuses to serve; drop one of them")) + } if err := cli.validateVODCDN(); err != nil { errs = append(errs, err) } diff --git a/pkg/config/config_check_test.go b/pkg/config/config_check_test.go index 0534ffc52..bcfa1f1c3 100644 --- a/pkg/config/config_check_test.go +++ b/pkg/config/config_check_test.go @@ -59,6 +59,10 @@ func TestCheckConfig(t *testing.T) { bad(CLI{TranscodeRenditions: "not-a-rendition", FirebaseServiceAccountFile: filepath.Join(t.TempDir(), "nope.json")}, "--transcode-renditions", "--firebase-service-account-file") + // Pushing to peers needs syndication: a pushing node refuses the peers' + // subscriptions without it. + bad(CLI{DataDir: t.TempDir(), DisableSyndication: true, Peers: []string{"https://peer.example"}}, + "--peers", "--disable-syndication") // Multiple problems are reported together, not one at a time. multi := CLI{LivepeerGateway: true, LivepeerGatewayURL: "http://127.0.0.1:8935", diff --git a/pkg/director/roles.go b/pkg/director/roles.go new file mode 100644 index 000000000..c5248d8f6 --- /dev/null +++ b/pkg/director/roles.go @@ -0,0 +1,152 @@ +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. The loops run on the +// session's context, so ending the session puts every job down. +func (ss *StreamSession) startRoles(repoDID string) { + ss.roles.Do(func() { + ctx := ss.ctx + 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() {}, + }) + }) +} + +// startTranscodeRole enters the session into contention for transcoding, +// once, and only on a node that is an origin for the stream: one that +// ingests it, or that a peer pushes it to. Peers take a stream's +// renditions from its origin, so renditions minted anywhere else would +// reach nobody but that node's own viewers. +func (ss *StreamSession) startTranscodeRole(repoDID string) { + if ss.cli.LivepeerGatewayURL == "" { + return + } + ss.transcodeRole.Do(func() { + ss.contend(ss.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. The +// first check happens before contend returns, so a job this node wins is +// up before the segment that started the contention is handled; the rest +// run in the background. +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) + var held bool + var renewed time.Time + var cancel context.CancelFunc + check := func() { + // The job is kept only while its last confirmed lease lasts, so a + // renewal the database never answers gives up when that runs out. + asked := time.Now() + deadline := asked.Add(roleLeaseRenew) + if held { + deadline = renewed.Add(roleLeaseTTL) + } + dctx, done := context.WithDeadline(ctx, deadline) + ok, err := ss.statefulDB.AcquireLease(dctx, name, holder, roleLeaseTTL) + done() + 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 = asked + } + 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") + held = false + cancel() + r.stop() + } + } + check() + ss.Go(ctx, func() error { + defer func() { + if held { + cancel() + r.stop() + rctx, done := context.WithTimeout(context.WithoutCancel(ctx), roleLeaseRenew) + defer done() + if err := ss.statefulDB.ReleaseLease(rctx, name, holder); err != nil { + log.Warn(ctx, "could not release stream role lease", "error", err) + } + } + }() + for { + select { + case <-ctx.Done(): + return nil + case <-time.After(roleLeaseRenew): + } + check() + } + }) +} diff --git a/pkg/director/roles_test.go b/pkg/director/roles_test.go new file mode 100644 index 000000000..587663f58 --- /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(t.Context(), "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 51ec20eee..a64fe9583 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) // The object being uploaded can't be read back until it completes, so // hold the stream's segments in memory that long for clips to draw on. ss.mm.RetainSegments(repoDID, s3.MaxOpenObjectAge) @@ -97,19 +91,44 @@ 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) } -func (ss *StreamSession) s3Upload(ctx context.Context, notif *media.NewSegmentNotification) { - if ss.s3Uploader == nil { +// 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 +} + +// withUploader runs f on the live recording, if there is one, as a session +// goroutine s3Close lets finish before it closes that recording. +func (ss *StreamSession) withUploader(ctx context.Context, f func(u *s3.S3Uploader) error) { + ss.s3Mu.Lock() + u := ss.s3Uploader + if u != nil { + ss.s3Pending.Add(1) + } + ss.s3Mu.Unlock() + if u == nil { return } ss.Go(ctx, func() error { + defer ss.s3Pending.Done() + return f(u) + }) +} + +func (ss *StreamSession) s3Upload(ctx context.Context, notif *media.NewSegmentNotification) { + ss.withUploader(ctx, func(u *s3.S3Uploader) 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, notif.Segment.StartTime) + return u.AddSegment(ctx, notif.Muxl, notif.Segment.StartTime) }) } @@ -119,20 +138,26 @@ 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 { - return - } - ss.Go(ctx, func() error { - return ss.s3Uploader.Cutover(ctx) + ss.withUploader(ctx, func(u *s3.S3Uploader) error { + 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. Segments already handed to the recording go in +// first. 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()) + ss.s3Pending.Wait() + // 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..a63dbcbba 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,16 @@ 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 + // s3Pending counts the s3Uploader's queued segment workers. + s3Pending sync.WaitGroup + // roles enters the session into the stream's jobs once; see startRoles. + // transcodeRole is separate: it starts on the first segment that makes + // this node an origin for the stream. + roles, transcodeRole 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,13 +144,12 @@ 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. + // 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(notif.Segment.RepoDID) if notif.Local { - ss.startLocalRole(ctx, notif.Segment.RepoDID) + ss.startTranscodeRole(notif.Segment.RepoDID) } // Start background workers for status, origin, and livestream updates @@ -167,13 +171,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 +192,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,10 +255,13 @@ 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. Transcoding is contended for only once this node is an + // origin for the stream (see startTranscodeRole). + ss.startRoles(spseg.Creator) + if notif.Local || ss.replicator.ReceivesPush(spseg.Creator) { + ss.startTranscodeRole(spseg.Creator) } // stream.received is enqueued once per StreamSession when the first local media segment is accepted. @@ -298,9 +286,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 +339,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 from 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 +372,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 { @@ -1133,10 +1119,25 @@ func (ss *StreamSession) HandleMultistreamTargets(ctx context.Context) error { // {target.Uri}:{rec.Url} -> runningMultistream // no concurrency issues, it's only used from this one loop running := map[string]*runningMultistream{} + // wait reports whether to go round again, false once ctx ends. + wait := func() bool { + select { + case <-ctx.Done(): + return false + case <-time.After(time.Second * 5): + return true + } + } for { targets, err := ss.statefulDB.ListMultistreamTargets(ss.repoDID, 100, 0, &isTrue) if err != nil { - return fmt.Errorf("failed to list multistream targets: %w", err) + // Retried rather than returned: this node holds the multistream + // lease, so nobody else would drive the targets. + log.Error(ctx, "failed to list multistream targets", "error", err) + if !wait() { + return nil + } + continue } currentRunning := map[string]bool{} for _, targetView := range targets { @@ -1177,11 +1178,8 @@ func (ss *StreamSession) HandleMultistreamTargets(ctx context.Context) error { delete(running, key) } } - select { - case <-ctx.Done(): + if !wait() { return nil - case <-time.After(time.Second * 5): - continue } } } diff --git a/pkg/localdb/segment.go b/pkg/localdb/segment.go index 30c0aad9b..7859a1640 100644 --- a/pkg/localdb/segment.go +++ b/pkg/localdb/segment.go @@ -322,11 +322,14 @@ func (m *LocalDatabase) LatestSegmentsForUser(user string, limit int, includeUnp return segs, nil } +// GetSegment returns the segment with the given id, or nil if it is not +// known. Receiving a segment twice is normal (a MoQ subscription is primed +// with recent segments on every reconnect), so this is the dedup check and +// must not fail on a known segment. func (m *LocalDatabase) GetSegment(id string) (*Segment, error) { var seg Segment err := m.DB.Model(&Segment{}). - Preload("Repo"). Where("id = ?", id). First(&seg).Error diff --git a/pkg/localdb/segment_get_test.go b/pkg/localdb/segment_get_test.go new file mode 100644 index 000000000..e7f4274d3 --- /dev/null +++ b/pkg/localdb/segment_get_test.go @@ -0,0 +1,27 @@ +package localdb + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// GetSegment is the receive-side dedup check: a known id must come back as +// the segment, not as an error, or a replayed segment (every MoQ reconnect +// is primed with recent ones) drops the whole session. +func TestGetSegment(t *testing.T) { + db, err := MakeDB(":memory:") + require.NoError(t, err) + ldb := db.(*LocalDatabase) + + seg, err := ldb.GetSegment("urn:c2pa:unknown") + require.NoError(t, err) + require.Nil(t, seg) + + require.NoError(t, ldb.CreateSegment(&Segment{ID: "urn:c2pa:known", RepoDID: "did:plc:a", StartTime: time.Now().UTC()})) + seg, err = ldb.GetSegment("urn:c2pa:known") + require.NoError(t, err) + require.NotNil(t, seg) + require.Equal(t, "did:plc:a", seg.RepoDID) +} diff --git a/pkg/media/validate.go b/pkg/media/validate.go index 75df0fbce..e9e9018b1 100644 --- a/pkg/media/validate.go +++ b/pkg/media/validate.go @@ -37,6 +37,12 @@ type segmentValidation struct { ValidationState string `json:"validation_state"` } +// replicatedMaxAge is how far a replicated (non-local) segment's start time +// may be from this node's clock before it is refused as a replay; see +// validateSource. A live segment is seconds old; the slack is for clock +// skew between nodes and a pull that resumes after an outage. +const replicatedMaxAge = 10 * time.Minute + func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader, local bool) error { return mm.validateMP4(ctx, input, local, "") } @@ -122,6 +128,20 @@ func (mm *MediaManager) validateSource(ctx context.Context, buf []byte, local bo return nil, fmt.Errorf("segment manifest has no label") } label := *valid.Manifest.Label + // A replicated segment is live by definition: it reached us over a + // peer's subscription, a push, or /api/segment, all of which carry what + // a streamer is producing now. One from the past is a replay: anyone can + // fetch a public streamer's old signed segments, and without this a + // node would take them as that streamer going live again. The window + // is generous because the clocks are the nodes' own. Checked before the + // dedup skip below: a replay of a segment this node already holds must + // not pass as a successful ingest either. + if !local { + if age := time.Since(meta.StartTime.Time()); age > replicatedMaxAge || age < -replicatedMaxAge { + return nil, fmt.Errorf("replicated segment of %s starts at %s, %s from now; refusing a replay", meta.Creator, meta.StartTime.Time().Format(time.RFC3339), age.Round(time.Second)) + } + } + if mm.model != nil { _, dbSpan := tracer.Start(ctx, "ValidateMP4.LocalDB.GetSegment") oldSeg, err := mm.localDB.GetSegment(label) 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/iroh_replicator/kv.go b/pkg/replication/iroh_replicator/kv.go index 92136ada3..e80c717f5 100644 --- a/pkg/replication/iroh_replicator/kv.go +++ b/pkg/replication/iroh_replicator/kv.go @@ -126,6 +126,11 @@ func (swarm *IrohSwarm) BuildOriginRecord(origin *placestream.BroadcastOrigin) e return nil } +// ReceivesPush: the swarm has no push; a stream is local or gossiped. +func (swarm *IrohSwarm) ReceivesPush(string) bool { + return false +} + func (swarm *IrohSwarm) Start(ctx context.Context, cli *config.CLI) error { if len(cli.Tickets) > 0 { err := swarm.Node.JoinPeers(cli.Tickets) diff --git a/pkg/replication/replicator.go b/pkg/replication/replicator.go index fef1ba317..c8573570a 100644 --- a/pkg/replication/replicator.go +++ b/pkg/replication/replicator.go @@ -15,4 +15,8 @@ type Replicator interface { SendSegment(context.Context, *media.NewSegmentNotification) error // populate this origin record with whatever fields are pertinent to your replicator BuildOriginRecord(*placestream.BroadcastOrigin) error + // ReceivesPush reports whether a peer is pushing streamer's stream to + // this node right now, which makes this node an origin for it: peers + // take the stream, renditions included, from here. + ReceivesPush(streamer string) bool } diff --git a/pkg/replication/websocketrep/moq.go b/pkg/replication/websocketrep/moq.go index 23f282a47..07a88582a 100644 --- a/pkg/replication/websocketrep/moq.go +++ b/pkg/replication/websocketrep/moq.go @@ -5,8 +5,11 @@ import ( "context" "encoding/binary" "encoding/hex" + "encoding/json" "errors" "fmt" + "net" + "net/http" "net/url" "slices" "strconv" @@ -14,12 +17,16 @@ import ( "sync" "time" + "github.com/bluesky-social/indigo/atproto/syntax" "golang.org/x/sync/errgroup" official "moq.dev/moq" + "stream.place/streamplace/pkg/aqhttp" + "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/media" "stream.place/streamplace/pkg/moq" "stream.place/streamplace/pkg/muxl" + "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/spmetrics" ) @@ -49,7 +56,8 @@ var moqLivenessInterval = time.Second var moqTLSRetry = 5 * time.Second // serveMoQ runs the node's Media over QUIC listener on --moq-addr: every -// stream live on this node's bus, for the peers that pull it. +// stream live on this node's bus, for the peers that pull it, and the +// other way, peers pushing their streams here. func (r *WebsocketReplicator) serveMoQ(ctx context.Context) error { if r.MoqTLS == nil { return fmt.Errorf("moq listener has no TLS files") @@ -85,7 +93,29 @@ func (r *WebsocketReplicator) serveMoQ(ctx context.Context) error { r.MoqTLS.Refresh(gctx, func(err error) { log.Error(gctx, "refreshing moq TLS", "error", err) }) return nil }) - g.Go(func() error { return srv.Serve(gctx) }) + g.Go(func() error { + var wg sync.WaitGroup + defer wg.Wait() + for req, err := range srv.Requests(gctx) { + if err != nil { + return err + } + // Each session gets a private origin for what the peer + // pushes: an announcement proves nothing, so it must never + // become a route this node serves. + pushed := official.NewOriginProducer() + if err := req.SetConsume(pushed); err != nil { + _ = req.Reject(gctx, 500) + continue + } + wg.Add(1) + go func() { + defer wg.Done() + r.acceptSession(gctx, req, pushed.Consume()) + }() + } + return nil + }) g.Go(func() error { var wg sync.WaitGroup defer wg.Wait() @@ -286,14 +316,18 @@ func cmafFragments(track []byte) (fragments [][]byte, catalog []byte, err error) } // dialMoq opens a session to a moqt:// URL, pinning the certificate its -// certhash parameter names when it has one. -func (r *WebsocketReplicator) dialMoq(ctx context.Context, rawURL string) (*official.Client, error) { +// certhash parameter names when it has one; publish, if set, is what the +// session offers the peer (a push), else nothing (a pull). +func (r *WebsocketReplicator) dialMoq(ctx context.Context, rawURL string, publish *official.OriginProducer) (*official.Client, error) { u, err := url.Parse(rawURL) if err != nil { return nil, fmt.Errorf("could not parse moq URL: %w", err) } + if publish == nil { + publish = official.NewOriginProducer() + } opts := []official.ClientOption{ - official.WithPublishOrigin(official.NewOriginProducer()), + official.WithPublishOrigin(publish), official.WithSubscribeOrigin(official.NewOriginProducer()), // The pull and push loops own reconnects, so each attempt // rediscovers the URL and its certificate hash. @@ -340,21 +374,30 @@ func requestMoqBroadcast(ctx context.Context, client *official.Client, streamer func (r *WebsocketReplicator) consume(ctx context.Context, broadcast *official.BroadcastConsumer, streamer string) error { spmetrics.ReplicationOutboundOpen.WithLabelValues(streamer).Inc() defer spmetrics.ReplicationOutboundOpen.WithLabelValues(streamer).Dec() - return consumeBroadcast(ctx, broadcast, - func(segment []byte) error { - if err := r.mm.ValidateReplicatedMP4(context.Background(), bytes.NewReader(segment), streamer); err != nil { - return fmt.Errorf("could not validate segment: %w", err) - } - return nil - }, + return consumeBroadcast(ctx, broadcast, r.validateSource(streamer, nil), func(ctx context.Context, addendum []byte) error { return r.feedRenditions(ctx, streamer, addendum) }) } +// validateSource feeds streamer's source segments through validation; +// onSource, if set, runs after each one validates. +func (r *WebsocketReplicator) validateSource(streamer string, onSource func()) func([]byte) error { + return func(segment []byte) error { + if err := r.mm.ValidateReplicatedMP4(context.Background(), bytes.NewReader(segment), streamer); err != nil { + return fmt.Errorf("could not validate segment: %w", err) + } + if onSource != nil { + onSource() + } + return nil + } +} + // consumeBroadcast hands each source segment the broadcast's catalog // leads to, reassembled, to source, and each rendition track's segments to -// rendition, until something fails or the broadcast or ctx ends. +// rendition (none, when nil), until something fails or the broadcast or +// ctx ends; a broadcast its publisher ends is errMoqTrackEnded. func consumeBroadcast(ctx context.Context, broadcast *official.BroadcastConsumer, source func([]byte) error, rendition func(context.Context, []byte) error) error { catalogs, err := broadcast.SubscribeCatalog(ctx) if err != nil { @@ -381,7 +424,7 @@ func consumeBroadcast(ctx context.Context, broadcast *official.BroadcastConsumer return fmt.Errorf("source tracks changed from %v to %v", sourceTracks, src) } for _, name := range ren { - if renditions[name] { + if rendition == nil || renditions[name] { continue } renditions[name] = true @@ -392,7 +435,7 @@ func consumeBroadcast(ctx context.Context, broadcast *official.BroadcastConsumer }) } } - return errors.New("catalog ended") + return fmt.Errorf("catalog: %w", errMoqTrackEnded) }) return g.Wait() } @@ -522,7 +565,7 @@ func nextMoqGroup(ctx context.Context, track *official.TrackConsumer, name strin return nil, fmt.Errorf("%s track: %w", name, err) } if group == nil { - return nil, fmt.Errorf("%s track ended", name) + return nil, fmt.Errorf("%s: %w", name, errMoqTrackEnded) } return group, nil } @@ -562,5 +605,319 @@ func isMuxl(b []byte) bool { return len(b) >= 8 && string(b[4:8]) == "uuid" } +// errMoqTrackEnded is a track its publisher finished: the stream is over +// there, not broken. +var errMoqTrackEnded = errors.New("track ended") + // errMoqUnreachable is a peer whose MoQ listener could not be dialled. var errMoqUnreachable = errors.New("could not dial moq") + +// 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() +} + +// liveStream is a stream this node is ingesting, announced to --peers. +type liveStream struct { + last time.Time + stop context.CancelFunc +} + +// touchLive notes a local segment of streamer's. On the first, the stream +// is announced on pushOrigin, its tracks served from the bus like the +// listener's; it is retracted once expireLive finds it quiet. +func (r *WebsocketReplicator) touchLive(ctx context.Context, streamer string) { + r.liveMu.Lock() + defer r.liveMu.Unlock() + if s := r.live[streamer]; s != nil { + s.last = time.Now() + return + } + broadcast, err := r.pushOrigin.CreateBroadcast(streamer) + if err != nil { + log.Warn(ctx, "moq push: could not create broadcast", "streamer", streamer, "error", err) + return + } + ctx, stop := context.WithCancel(ctx) + r.live[streamer] = &liveStream{last: time.Now(), stop: stop} + r.group.Go(func() error { + defer broadcast.Close() + r.serveTracks(ctx, streamer, broadcast, func() error { return broadcast.Announce(official.Route{}) }) + return nil + }) +} + +// expireLive retracts the streams with no local segment for the stream +// session timeout: the moment the director ends the session and the bus +// forgets the stream. +func (r *WebsocketReplicator) expireLive() { + r.liveMu.Lock() + defer r.liveMu.Unlock() + for streamer, s := range r.live { + if time.Since(s.last) > r.cli.StreamSessionTimeout { + s.stop() + delete(r.live, streamer) + } + } +} + +// 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, publishing this node's live streams, 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 + } + client, err := r.dialMoq(ctx, moqURL, r.pushOrigin) + if err != nil { + return err + } + defer client.Close() + log.Log(ctx, "moq push: connected to peer", "url", moqURL) + if err := client.Session().Closed(ctx); err != nil { + return err + } + return fmt.Errorf("session closed") +} + +// 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 +} + +// acceptSession holds one incoming session while acceptPush takes what +// the peer pushes, and ends it when that stops. +func (r *WebsocketReplicator) acceptSession(ctx context.Context, req *official.Request, pushed *official.OriginConsumer) { + acceptCtx, cancel := context.WithTimeout(ctx, 15*time.Second) + sess, err := req.Accept(acceptCtx) + cancel() + if err != nil { + log.Debug(ctx, "moq: accepting a session failed", "error", err) + return + } + defer sess.Shutdown() + ctx, cancel = context.WithCancel(ctx) + defer cancel() + go func() { + _ = sess.Closed(ctx) + cancel() + }() + r.acceptPush(ctx, pushed) +} + +// moqPushMaxStreams bounds the streams one session may push at once. A +// node ingests a handful; a peer announcing more is not describing +// streams, and is ignored past this rather than allowed to grow our state. +const moqPushMaxStreams = 32 + +// acceptPush is the receiving end of push: the streams a peer announces +// are the ones it is ingesting. Each one's source is pulled 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's segments validate 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. The row follows validated segments, +// never the announcement alone: naming a DID proves nothing, a signed +// segment does. +func (r *WebsocketReplicator) acceptPush(ctx context.Context, pushed *official.OriginConsumer) { + ann, err := pushed.Announced(official.AnnounceOptions{}) + if err != nil { + log.Warn(ctx, "moq push: could not watch announcements", "error", err) + return + } + defer ann.Cancel() + g, gctx := errgroup.WithContext(ctx) + running := map[string]context.CancelFunc{} + for { + a, err := ann.Next(gctx) + if err != nil || a == nil { + break + } + streamer := a.Prefix() + stop, ok := running[streamer] + if !a.Active() { + if ok { + stop() + delete(running, streamer) + } + continue + } + if ok { + continue + } + if _, err := syntax.ParseDID(streamer); err != nil { + log.Warn(ctx, "moq push: ignoring an announcement that is not a streamer", "path", streamer) + continue + } + if len(running) >= moqPushMaxStreams { + log.Warn(ctx, "moq push: peer pushes too many streams, ignoring one", "streamer", streamer) + continue + } + if err := r.cli.StreamIsAllowed(streamer); err != nil { + log.Warn(ctx, "moq push: refusing pushed stream", "streamer", streamer, "error", err) + continue + } + sctx, stop := context.WithCancel(gctx) + running[streamer] = stop + g.Go(func() error { + log.Log(sctx, "moq push: receiving pushed stream", "streamer", streamer) + r.pushedMu.Lock() + r.pushed[streamer]++ + r.pushedMu.Unlock() + defer func() { + r.pushedMu.Lock() + r.pushed[streamer]-- + if r.pushed[streamer] == 0 { + delete(r.pushed, streamer) + } + r.pushedMu.Unlock() + }() + // Only the source: renditions a pusher sends are signed by + // whoever it is, and this node, the stream's origin now, mints + // its own (see startTranscodeRole). + broadcast, err := pushed.RequestBroadcast(sctx, streamer) + if err == nil { + err = consumeBroadcast(sctx, broadcast, r.validateSource(streamer, r.originAdvertiser(sctx, streamer)), nil) + } + if sctx.Err() != nil || errors.Is(err, errMoqTrackEnded) { + // Retracted or finished by the peer, or the session is over. + return nil + } + return fmt.Errorf("pushed stream %s: %w", streamer, err) + }) + } + for _, stop := range running { + stop() + } + if err := g.Wait(); err != nil { + log.Warn(ctx, "moq push: dropping the session", "error", err) + } +} + +// ReceivesPush reports whether a peer is pushing streamer's stream here. +func (r *WebsocketReplicator) ReceivesPush(streamer string) bool { + r.pushedMu.Lock() + defer r.pushedMu.Unlock() + return r.pushed[streamer] > 0 +} + +// originRefresh is how often a pushed stream's broadcast_origins row is +// touched while its segments validate; peers read the table every +// originPollInterval. +var originRefresh = 5 * time.Second + +// originAdvertiser returns a per-validated-segment hook that keeps this +// node's broadcast_origins row for streamer fresh; nil with no statedb. +func (r *WebsocketReplicator) originAdvertiser(ctx context.Context, streamer string) func() { + if r.state == nil { + return nil + } + var last time.Time + return func() { + if time.Since(last) < originRefresh { + return + } + last = time.Now() + if err := r.state.UpsertBroadcastOrigin(streamer, r.cli.ServerDID(), r.getMyMoqURL(), last); err != nil { + log.Warn(ctx, "moq push: could not record this node as origin", "streamer", streamer, "error", err) + } + } +} diff --git a/pkg/replication/websocketrep/moq_test.go b/pkg/replication/websocketrep/moq_test.go index 2bf7a5970..7d7d14460 100644 --- a/pkg/replication/websocketrep/moq_test.go +++ b/pkg/replication/websocketrep/moq_test.go @@ -3,9 +3,12 @@ package websocketrep import ( "bytes" "context" + "encoding/json" "errors" "fmt" "net" + "net/http" + "net/http/httptest" "os" "slices" "strconv" @@ -13,12 +16,14 @@ import ( "time" "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" official "moq.dev/moq" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/media" "stream.place/streamplace/pkg/moq" "stream.place/streamplace/pkg/muxl" + "stream.place/streamplace/pkg/placestream" ) // signedSegments signs the fragmented fixture the way live ingest does and @@ -110,7 +115,7 @@ func TestMoqServesTheBus(t *testing.T) { moqURL := fmt.Sprintf("moqt://%s?%s=%s", addr, moq.CertHashParam, cert.Hash()) var client *official.Client require.Eventually(t, func() bool { - client, err = r.dialMoq(ctx, moqURL) + client, err = r.dialMoq(ctx, moqURL, nil) return err == nil }, 10*time.Second, 100*time.Millisecond, "the listener comes up") defer client.Close() @@ -207,3 +212,138 @@ func TestMoqGroupIsBounded(t *testing.T) { require.NoError(t, read(frame, frame), "at the cap") require.ErrorContains(t, read(frame, frame, frame), "over 16 bytes") } + +// A stream this node ingests is announced to the peers it pushes to from +// its first segment, once, and retracted when its segments stop. +func TestLiveStreamsAreAnnounced(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + b := bus.NewBus() + r := NewWebsocketReplicator(&config.CLI{StreamSessionTimeout: 50 * time.Millisecond}, b, nil, nil, nil, nil) + r.group, ctx = errgroup.WithContext(ctx) + announced, err := r.pushOrigin.Consume().Announced(official.AnnounceOptions{}) + require.NoError(t, err) + defer announced.Cancel() + + b.PublishSegment(ctx, "did:plc:a", moqSourceChannel, &bus.Seg{Muxl: signedSegments(t, ctx)[0], Published: true}) + r.touchLive(ctx, "did:plc:a") + r.touchLive(ctx, "did:plc:a") + a, err := announced.Next(ctx) + require.NoError(t, err) + require.Equal(t, "did:plc:a", a.Prefix()) + require.True(t, a.Active()) + + r.expireLive() + require.Contains(t, r.live, "did:plc:a", "still live inside the session timeout") + time.Sleep(80 * time.Millisecond) + r.expireLive() + a, err = announced.Next(ctx) + require.NoError(t, err) + require.Equal(t, "did:plc:a", a.Prefix()) + require.False(t, a.Active(), "retracted, after one announcement") +} + +// A peer this node pushes to gets each live stream announced on the session +// and its segments served for as long as the stream is live, not just the +// first one. +func TestMoqPushesLiveStreams(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + pc, err := net.ListenPacket("udp", "127.0.0.1:0") + require.NoError(t, err) + addr := pc.LocalAddr().String() + require.NoError(t, pc.Close()) + cert, err := moq.NewSelfSignedCert() + require.NoError(t, err) + tlsFiles, err := moq.ManagedTLSFiles(t.TempDir(), cert.Certificate) + require.NoError(t, err) + require.NoError(t, tlsFiles.Sync()) + // The peer: what acceptSession and acceptPush do, short of validation. + srv, err := official.Listen(ctx, addr, official.WithTLSCert(tlsFiles.CertPath), official.WithTLSKey(tlsFiles.KeyPath)) + require.NoError(t, err) + defer srv.Close() + got := make(chan []byte, 8) + go func() { + for req, err := range srv.Requests(ctx) { + if err != nil { + return + } + pushed := official.NewOriginProducer() + if req.SetConsume(pushed) != nil { + return + } + sess, err := req.Accept(ctx) + if err != nil { + return + } + defer sess.Shutdown() + announced, err := pushed.Consume().AnnouncedBroadcast("did:plc:a") + if err != nil { + return + } + broadcast, err := announced.Available(ctx) + if err != nil { + return + } + _ = consumeBroadcast(ctx, broadcast, func(segment []byte) error { + got <- segment + return nil + }, nil) + return + } + }() + + b := bus.NewBus() + r := NewWebsocketReplicator(&config.CLI{StreamSessionTimeout: time.Minute}, b, nil, nil, nil, nil) + r.group, ctx = errgroup.WithContext(ctx) + segments := signedSegments(t, ctx) + b.PublishSegment(ctx, "did:plc:a", moqSourceChannel, &bus.Seg{Muxl: segments[0], Published: true}) + r.touchLive(ctx, "did:plc:a") + client, err := r.dialMoq(ctx, fmt.Sprintf("moqt://%s?%s=%s", addr, moq.CertHashParam, cert.Hash()), r.pushOrigin) + require.NoError(t, err) + defer client.Close() + next := func() []byte { + select { + case payload := <-got: + return payload + case <-ctx.Done(): + t.Fatal("no segment arrived") + } + return nil + } + require.Equal(t, segments[0], next()) + b.PublishSegment(ctx, "did:plc:a", moqSourceChannel, &bus.Seg{Muxl: segments[1], Published: true}) + require.Equal(t, segments[1], next(), "segments keep coming after the first") +} + +// 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 b0bd86d49..f10c6831a 100644 --- a/pkg/replication/websocketrep/websocket_replicator.go +++ b/pkg/replication/websocketrep/websocket_replicator.go @@ -5,7 +5,6 @@ import ( "context" "errors" "fmt" - "net" "net/url" "strings" "sync" @@ -14,6 +13,7 @@ import ( "github.com/gorilla/websocket" glex "github.com/streamplace/glex/runtime" "golang.org/x/sync/errgroup" + official "moq.dev/moq" "stream.place/streamplace/pkg/appbsky" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/config" @@ -49,11 +49,20 @@ type WebsocketReplicator struct { // MoqTLS is the certificate the MoQ listener serves, as files: set // before Start when --moq-addr is on (see serveMoQ). MoqTLS *moq.TLSFiles + // pushOrigin is what this node's sessions to --peers publish: the + // streams it is ingesting, each announced while in live. + pushOrigin *official.OriginProducer + liveMu sync.Mutex + live map[string]*liveStream + // pushed counts the sessions currently pushing each streamer here. + pushedMu sync.Mutex + pushed map[string]int } -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 { return &WebsocketReplicator{ bus: bus, + cli: cli, mod: mod, state: state, conns: make(map[string]bool), @@ -61,6 +70,9 @@ func NewWebsocketReplicator(bus *bus.Bus, mod model.Model, mm *media.MediaManage latest: make(map[string]latestOrigin), mm: mm, moqCert: moqCert, + pushOrigin: official.NewOriginProducer(), + live: map[string]*liveStream{}, + pushed: map[string]int{}, } } @@ -106,7 +118,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) @@ -121,7 +132,35 @@ func (r *WebsocketReplicator) Start(ctx context.Context, cli *config.CLI) error return r.serveMoQ(ctx) }) } - return r.startBusSubscribe(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.expireLive() + } + } + }) + 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 @@ -334,7 +373,7 @@ func (r *WebsocketReplicator) open(ctx context.Context, view *placestream.Broadc // openMoq requests the streamer's broadcast from the origin over Media over // QUIC and feeds it through the same validation as openWebsocket. func (r *WebsocketReplicator) openMoq(ctx context.Context, origin *placestream.BroadcastOrigin) error { - client, err := r.dialMoq(ctx, *origin.MoqURL) + client, err := r.dialMoq(ctx, *origin.MoqURL, nil) if err != nil { return err } @@ -438,8 +477,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. The stream is announced on +// its first segment: the tracks a peer then asks for are fed from the bus, +// which hands a subscriber that arrives before the director publishes the +// segment everything from then on. func (r *WebsocketReplicator) SendSegment(ctx context.Context, seg *media.NewSegmentNotification) error { + if seg.Local { + r.touchLive(ctx, seg.Segment.RepoDID) + } return nil } @@ -458,38 +504,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 c135a75e6..ca9bf81c7 100644 --- a/pkg/s3/s3.go +++ b/pkg/s3/s3.go @@ -82,6 +82,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 @@ -192,12 +197,10 @@ type uploadCmd struct { // AddSegment feeds one bare canonical MUXL segment (uuid+moof+mdat per track) // that starts at start 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. +// reuse its buffer. A call that meets Close returns an error. func (u *S3Uploader) AddSegment(ctx context.Context, data []byte, start time.Time) error { + u.closing.RLock() + defer u.closing.RUnlock() if u.closed.Load() { return fmt.Errorf("s3 uploader closed") } @@ -218,6 +221,8 @@ func (u *S3Uploader) AddSegment(ctx context.Context, data []byte, start time.Tim // 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 } @@ -240,8 +245,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 256511812..2c747ce72 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..ba0387a7d --- /dev/null +++ b/pkg/statedb/lease.go @@ -0,0 +1,45 @@ +package statedb + +import ( + "context" + "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(ctx context.Context, name, holder string, ttl time.Duration) (bool, error) { + now := time.Now() + res := state.DB.WithContext(ctx).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(ctx context.Context, name, holder string) error { + return state.DB.WithContext(ctx).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..57393754d --- /dev/null +++ b/pkg/statedb/lease_test.go @@ -0,0 +1,57 @@ +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" + ctx := t.Context() + held, err := state.AcquireLease(ctx, name, "did:web:a", time.Minute) + require.NoError(t, err) + require.True(t, held) + + held, err = state.AcquireLease(ctx, name, "did:web:b", time.Minute) + require.NoError(t, err) + require.False(t, held, "a live lease is exclusive") + + held, err = state.AcquireLease(ctx, name, "did:web:a", time.Minute) + require.NoError(t, err) + require.True(t, held, "the holder renews") + + require.NoError(t, state.ReleaseLease(ctx, name, "did:web:b")) + held, err = state.AcquireLease(ctx, name, "did:web:b", time.Minute) + require.NoError(t, err) + require.False(t, held, "only the holder can release") + + require.NoError(t, state.ReleaseLease(ctx, name, "did:web:a")) + held, err = state.AcquireLease(ctx, 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(ctx, 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(ctx, name, "did:web:b", -time.Millisecond) + require.NoError(t, err) + require.True(t, held) + held, err = state.AcquireLease(ctx, 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(ctx, "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 93391780e..ce9df7a2a 100644 --- a/pkg/statedb/statedb.go +++ b/pkg/statedb/statedb.go @@ -89,6 +89,7 @@ var StatefulDBModels = []any{ Upload{}, DraftVideo{}, CDNLogPart{}, + Lease{}, } var NoPostgresDatabaseCode = "3D000" -- 2.54.0