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"