Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
21 kB · 582 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583package cmd
import ( "bufio" "context" "encoding/json" "errors" "fmt" "net" "net/http" "net/url" "os" "os/exec" "os/signal" "path/filepath" "strings" "syscall" "time"
comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/util" "github.com/bluesky-social/indigo/xrpc" "github.com/google/uuid" glex "github.com/streamplace/glex/runtime" urfavecli "github.com/urfave/cli/v3" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/aqhttp" "stream.place/streamplace/pkg/atproto" spcomatproto "stream.place/streamplace/pkg/comatproto" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/crypto/spkey" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/test/remote")
// createRecord writes a glex record into the given repo via the public// com.atproto.repo.createRecord XRPC. The repo's own generated comatproto// types are used (rather than indigo's) so the record's $type survives// marshalling: indigo's LexiconTypeDecoder only accepts structs with a const// $type field, which glex-generated records do not have.func createRecord(ctx context.Context, client *xrpc.Client, collection, repo string, record glex.Record) (string, error) { inp := spcomatproto.RepoCreateRecord_Input{ Collection: collection, Repo: repo, Record: &glex.LexiconTypeDecoder{Val: record}, } out := spcomatproto.RepoCreateRecord_Output{} if err := client.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.createRecord", map[string]any{}, inp, &out); err != nil { return "", err } return out.Uri, nil}
// putRecord writes (creates or replaces) a repo record at an explicit rkey. The// node keys a user's settings for it by rkey — not by any field in the record —// so place.stream.server.settings has to live at the node's host, exactly as the// app writes it.func putRecord(ctx context.Context, client *xrpc.Client, collection, repo, rkey string, record glex.Record) error { inp := spcomatproto.RepoPutRecord_Input{ Collection: collection, Repo: repo, Rkey: rkey, Record: &glex.LexiconTypeDecoder{Val: record}, } out := spcomatproto.RepoPutRecord_Output{} return client.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.putRecord", map[string]any{}, inp, &out)}
func makeE2eCommand(build *config.BuildFlags) *urfavecli.Command { return &urfavecli.Command{ Name: "e2e", Usage: "start a self-contained e2e test environment with a test account and live stream", Flags: []urfavecli.Flag{ &urfavecli.StringFlag{ Name: "dev-env", Usage: "path to js/dev-env/run.mjs", Value: "js/dev-env/run.mjs", Sources: urfavecli.EnvVars("SP_DEV_ENV_MJS"), }, // Together these turn on locally trusted HTTPS, so atproto OAuth // works end to end; see e2e_https.go. &urfavecli.StringFlag{ Name: "https-pds-hostname", Usage: "serve the PDS at https://<this>, with account handles under it; it and *.<it> must resolve to 127.0.0.1 (binds 127.0.0.1:443)", Sources: urfavecli.EnvVars("SP_E2E_HTTPS_PDS_HOSTNAME"), }, &urfavecli.StringFlag{ Name: "https-station-hostname", Usage: "the node's broadcaster host, served at https://<this>, which must resolve to 127.0.0.1", Sources: urfavecli.EnvVars("SP_E2E_HTTPS_STATION_HOSTNAME"), }, &urfavecli.IntFlag{ Name: "https-port", Usage: "port the HTTPS front end listens on; the URLs stay portless, so anything but 443 only works where loopback 443 is redirected to it", Value: 443, Sources: urfavecli.EnvVars("SP_E2E_HTTPS_PORT"), }, &urfavecli.StringFlag{ Name: "app-bundle-id", Usage: "bundle id of the mobile app under test, which the node hands OAuth logins back to (its --app-bundle-id)", Sources: urfavecli.EnvVars("SP_E2E_APP_BUNDLE_ID"), }, }, Action: func(ctx context.Context, cmd *urfavecli.Command) error { // Canonical form, so they compare equal to the SNI names the // TLS front end routes on. canon := func(h string) string { return strings.ToLower(strings.TrimSuffix(h, ".")) } pdsHost, stationHost := canon(cmd.String("https-pds-hostname")), canon(cmd.String("https-station-hostname")) if (pdsHost == "") != (stationHost == "") { return errors.New("--https-pds-hostname and --https-station-hostname go together") } return runE2E(ctx, cmd.String("dev-env"), pdsHost, stationHost, int(cmd.Int("https-port")), cmd.String("app-bundle-id")) }, }}
// nodeReadyTimeout bounds how long runE2E waits for the forked node to answer// /api/healthz. A cold dev node has to run a GStreamer self-test and open its// databases first, so this is generous — but it is a bound, not forever.const nodeReadyTimeout = 90 * time.Second
type e2eDevEnv struct { PDSURL string `json:"pds-url"` PLCURL string `json:"plc-url"` // Only in HTTPS mode: the account the PDS resolves lexicons from. LexiconDID string `json:"lexicon-did"` LexiconPassword string `json:"lexicon-password"`}
func freePort() (int, error) { l, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { return 0, err } defer l.Close() return l.Addr().(*net.TCPAddr).Port, nil}
func freeUDPPort() (int, error) { c, err := net.ListenPacket("udp", "127.0.0.1:0") if err != nil { return 0, err } defer c.Close() return c.LocalAddr().(*net.UDPAddr).Port, nil}
// startNode forks this binary as a Streamplace node on the dev-env's PDS and// PLC, with extra on top of the configuration every harness node shares.func startNode(ctx context.Context, self string, env e2eDevEnv, dataDir string, extra []string) (*exec.Cmd, error) { nodeCmd := exec.CommandContext(ctx, self) // Inherit the parent environment (dev builds need LD_LIBRARY_PATH etc.) // but strip any SP_ vars so the node only gets our config. for _, kv := range os.Environ() { if !strings.HasPrefix(kv, "SP_") { nodeCmd.Env = append(nodeCmd.Env, kv) } } nodeCmd.Env = append(nodeCmd.Env, fmt.Sprintf("SP_RELAY_HOST=%s", strings.ReplaceAll(env.PDSURL, "http://", "ws://")), fmt.Sprintf("SP_PLC_URL=%s", env.PLCURL), fmt.Sprintf("SP_DATA_DIR=%s", dataDir), "SP_STREAM_SESSION_TIMEOUT=30s", "SP_TRUST_PRIVATE_NETWORK=true", ) nodeCmd.Env = append(nodeCmd.Env, extra...) nodeCmd.Stdout = os.Stderr nodeCmd.Stderr = os.Stderr // Own process group, so cleanup can take the whole tree down at once. setNodeProcessGroup(nodeCmd) if err := nodeCmd.Start(); err != nil { return nil, fmt.Errorf("start node: %w", err) } return nodeCmd, nil}
// waitForNode waits for the node at httpAddr to answer /api/healthz. Bounded,// so a node that never comes up reports why instead of hanging the harness// (and `make provision`) with no output forever.func waitForNode(ctx context.Context, httpAddr string) error { healthURL := fmt.Sprintf("http://%s/api/healthz", httpAddr) httpClient := &http.Client{Timeout: 2 * time.Second} deadline := time.Now().Add(nodeReadyTimeout) var lastErr error for { if err := ctx.Err(); err != nil { return err } resp, err := httpClient.Get(healthURL) switch { case err != nil: lastErr = err case resp.StatusCode == http.StatusOK: // Close every response, not just the happy one: a body left open // holds its connection out of the pool, so a node answering // non-200 exhausts them. resp.Body.Close() return nil default: lastErr = fmt.Errorf("GET %s: %s", healthURL, resp.Status) resp.Body.Close() } if time.Now().After(deadline) { if lastErr == nil { lastErr = errors.New("no response") } return fmt.Errorf("node was not ready after %s: %w", nodeReadyTimeout, lastErr) } time.Sleep(200 * time.Millisecond) }}
func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost string, httpsPort int, appBundleID string) error { // Ctrl-C / SIGTERM must unwind through the normal path, or none of the // teardown below runs: Go's default handling exits immediately, which left // the dev-env node, the forked node and the temp data dir behind. ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) defer stop()
// Claim the HTTPS listeners first: failing to bind 443 should stop us // before anything else has started. var tlsEnv *e2eHTTPS handleDomain := "test" if httpsPDSHost != "" { var err error tlsEnv, err = newE2EHTTPS(httpsPDSHost, httpsStationHost, httpsPort) if err != nil { return err } defer tlsEnv.Close() handleDomain = httpsPDSHost }
// Start the Node.js PDS/PLC dev environment. devEnvCmd := exec.CommandContext(ctx, "node", devEnvPath) if tlsEnv != nil { devEnvCmd.Env = append(os.Environ(), tlsEnv.DevEnvEnv()...) } devEnvCmd.Stderr = os.Stderr devEnvStdout, err := devEnvCmd.StdoutPipe() if err != nil { return fmt.Errorf("dev-env stdout pipe: %w", err) } if err := devEnvCmd.Start(); err != nil { return fmt.Errorf("start dev-env: %w", err) } defer devEnvCmd.Process.Kill() //nolint:errcheck
// The PDS logs JSON to stdout too when LOG_ENABLED is set, so skip lines // until the one that carries the URLs. var env e2eDevEnv scanner := bufio.NewScanner(devEnvStdout) for env.PDSURL == "" { if !scanner.Scan() { return fmt.Errorf("dev-env exited without printing its URLs: %w", scanner.Err()) } log.Log(ctx, "dev-env: "+scanner.Text()) _ = json.Unmarshal(scanner.Bytes(), &env) //nolint:errcheck // not every line is ours } go func() { for scanner.Scan() { log.Log(ctx, "dev-env: "+scanner.Text()) } }()
// Create a test account on the local PDS. xrpcc := &xrpc.Client{Host: env.PDSURL, Client: &aqhttp.TrustedClient} uu, err := uuid.NewRandom() if err != nil { return err } handle := fmt.Sprintf("sp-%s.%s", uu.String()[:8], handleDomain) email := fmt.Sprintf("%s@example.com", handle) password := "test" out, err := comatproto.ServerCreateAccount(ctx, xrpcc, &comatproto.ServerCreateAccount_Input{ Handle: handle, Email: &email, Password: &password, }) if err != nil { return fmt.Errorf("create account: %w", err) } session, err := comatproto.ServerCreateSession(ctx, xrpcc, &comatproto.ServerCreateSession_Input{ Identifier: out.Handle, Password: password, }) if err != nil { return fmt.Errorf("create session: %w", err) } xrpcc = &xrpc.Client{ Host: env.PDSURL, Client: &aqhttp.TrustedClient, Auth: &xrpc.AuthInfo{ Did: out.Did, AccessJwt: session.AccessJwt, RefreshJwt: session.RefreshJwt, Handle: out.Handle, }, }
if env.LexiconDID != "" { if err := publishLexicons(ctx, env); err != nil { return fmt.Errorf("publish lexicons: %w", err) } }
// Pick free ports for the node. httpPort, err := freePort() if err != nil { return err } internalPort, err := freePort() if err != nil { return err } rtmpPort, err := freePort() if err != nil { return err } moqPort, err := freeUDPPort() if err != nil { return err } httpAddr := fmt.Sprintf("127.0.0.1:%d", httpPort) broadcasterHost := httpAddr if tlsEnv != nil { broadcasterHost = httpsStationHost pdsURL, err := url.Parse(env.PDSURL) if err != nil { return fmt.Errorf("parse dev-env pds url: %w", err) } plcURL, err := url.Parse(env.PLCURL) if err != nil { return fmt.Errorf("parse dev-env plc url: %w", err) } tlsEnv.Serve(ctx, pdsURL.Host, plcURL.Host, httpAddr) }
// Fork ourselves as a Streamplace server node. self, err := os.Executable() if err != nil { return fmt.Errorf("find executable: %w", err) } dataDir, err := os.MkdirTemp("", "streamplace-e2e-*") if err != nil { return err } defer os.RemoveAll(dataDir) //nolint:errcheck
// Break the debug-recording sink on purpose: a regular file where the // recording tree (debug-recordings/<did>/) belongs makes every // DebugRecordingCreate fail — the local-disk shape of a misconfigured or // unpaid S3 bucket. The test account has debugRecording enabled below, so // every flow runs against a recorder that fails on every attempt: recordings // are best-effort and must never stall ingest or block playback. Run the // whole suite this way on purpose; that is the regression it covers. if err := os.WriteFile(filepath.Join(dataDir, "debug-recordings"), []byte("e2e: recording sink intentionally broken\n"), 0o644); err != nil { return fmt.Errorf("break debug-recording sink: %w", err) }
nodeEnv := []string{ fmt.Sprintf("SP_HTTP_ADDR=%s", httpAddr), fmt.Sprintf("SP_HTTP_INTERNAL_ADDR=127.0.0.1:%d", internalPort), fmt.Sprintf("SP_RTMP_ADDR=127.0.0.1:%d", rtmpPort), fmt.Sprintf("SP_MOQ_ADDR=127.0.0.1:%d", moqPort), fmt.Sprintf("SP_DEV_ACCOUNT_CREDS=%s=%s", out.Did, password), // The account administers the node's branding, which web flows edit // through Settings -> Branding. fmt.Sprintf("SP_ADMIN_DIDS=%s", out.Did), 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), } if tlsEnv != nil { nodeEnv = append(nodeEnv, tlsEnv.NodeEnv()...) } if appBundleID != "" { nodeEnv = append(nodeEnv, "SP_APP_BUNDLE_ID="+appBundleID) } nodeCmd, err := startNode(ctx, self, env, dataDir, nodeEnv) if err != nil { return err } // Kill the node *and* the ingest workers it detaches; the plain // Process.Kill that used to be here left those behind. defer killNode(ctx, nodeCmd, out.Did) if err := waitForNode(ctx, httpAddr); err != nil { return err }
// A second node, a peer that syndicates whatever the first ingests: it // hears the origin record on the shared PDS firehose and pulls the // stream over MoQ (see docs/moq.md), and playing the stream from it is // what proves that transfer end to end. Started now, before the records // below, so it indexes the stream key as it is created like the first. peerHTTPPort, err := freePort() if err != nil { return err } peerInternalPort, err := freePort() if err != nil { return err } peerRTMPPort, err := freePort() if err != nil { return err } peerMoqPort, err := freeUDPPort() if err != nil { return err } peerAddr := fmt.Sprintf("127.0.0.1:%d", peerHTTPPort) peerDataDir, err := os.MkdirTemp("", "streamplace-e2e-peer-*") if err != nil { return err } defer os.RemoveAll(peerDataDir) //nolint:errcheck peerEnv := []string{ fmt.Sprintf("SP_HTTP_ADDR=%s", peerAddr), fmt.Sprintf("SP_HTTP_INTERNAL_ADDR=127.0.0.1:%d", peerInternalPort), fmt.Sprintf("SP_RTMP_ADDR=127.0.0.1:%d", peerRTMPPort), fmt.Sprintf("SP_MOQ_ADDR=127.0.0.1:%d", peerMoqPort), fmt.Sprintf("SP_BROADCASTER_HOST=%s", peerAddr), fmt.Sprintf("SP_WEBSOCKET_URL=ws://%s", peerAddr), "SP_SYNDICATE=*", } if tlsEnv != nil { // The peer syncs the streamer's repo from the same HTTPS PDS. peerEnv = append(peerEnv, tlsEnv.TrustEnv()...) } peerCmd, err := startNode(ctx, self, env, peerDataDir, peerEnv) if err != nil { return fmt.Errorf("peer: %w", err) } defer killNode(ctx, peerCmd, out.Did) if err := waitForNode(ctx, peerAddr); err != nil { return fmt.Errorf("peer: %w", err) }
// Register a stream key for the test account. priv, pub, err := spkey.GenerateStreamKeyForDID(out.Did) if err != nil { return fmt.Errorf("generate stream key: %w", err) } createdBy := "e2e" streamKey := placestream.Key{ SigningKey: pub.DIDKey(), CreatedAt: time.Now().Format(util.ISO8601), CreatedBy: &createdBy, } if _, err := createRecord(ctx, xrpcc, "place.stream.key", out.Did, &streamKey); err != nil { return fmt.Errorf("register stream key: %w", err) } // Turn debug recording on for the test account, whose sink is broken above: // every stream session therefore attempts a recording that fails, which is // what makes the whole suite cover "a failed recording must not take the // stream down". debugRecording := true if err := putRecord(ctx, xrpcc, "place.stream.server.settings", out.Did, broadcasterHost, &placestream.ServerSettings{ LexiconTypeID: "place.stream.server.settings", DebugRecording: &debugRecording, }); err != nil { return fmt.Errorf("enable debug recording: %w", err) } // Create a livestream record so the stream shows up in feeds; normally the // app does this via place.stream.live.startLivestream when a user goes // live. The node keeps lastSeenAt fresh on our behalf via // SP_DEV_ACCOUNT_CREDS while segments are ingesting. now := time.Now().UTC().Format(util.ISO8601) livestream := placestream.Livestream{ LexiconTypeID: "place.stream.livestream", CreatedAt: now, LastSeenAt: &now, Title: "e2e test stream", } if _, err := createRecord(ctx, xrpcc, "place.stream.livestream", out.Did, &livestream); err != nil { return fmt.Errorf("create livestream record: %w", err) } // And a VOD, so flows have a video page to open. It has no source tracks: // its metadata (title, author) loads, but there is nothing to play. video := placestream.Video{ LexiconTypeID: "place.stream.video", CreatedAt: now, Title: "e2e test video", DurationMs: 10_000, Source: placestream.Video_Source{ MediaDefs_SourceTracks: &placestream.MediaDefs_SourceTracks{ LexiconTypeID: "place.stream.media.defs#sourceTracks", Tracks: []spcomatproto.RepoStrongRef{}, }, }, } videoURI, err := createRecord(ctx, xrpcc, "place.stream.video", out.Did, &video) if err != nil { return fmt.Errorf("create video record: %w", err) }
// Give the node a moment to index the key before we start streaming. time.Sleep(1 * time.Second)
// Stream a test fixture in a loop so it outlasts any Maestro test run. fixture := remote.RemoteFixture("3188c071b354f2e548d7f2d332699758e8e3ab1600280e5b07cb67eedc64f274/BigBuckBunny_1sGOP_240p30_NoBframes.mp4") g, streamCtx := errgroup.WithContext(ctx) g.Go(func() error { for { select { case <-streamCtx.Done(): return nil default: } whip := &WHIPClient{ StreamKey: priv, File: fixture, Endpoint: fmt.Sprintf("http://%s", httpAddr), Count: 1, } if err := whip.WHIP(streamCtx); err != nil && streamCtx.Err() == nil { log.Log(streamCtx, "whip stream ended, restarting", "err", err) } // Brief pause between restarts so we don't spin on errors. select { case <-streamCtx.Done(): return nil case <-time.After(500 * time.Millisecond): } } })
// Print the env vars for the workflow to consume, in one write: callers // poll for SERVER_URL and then read the whole file. E2E_FIXTURE_MP4 is // the local file the ingest loop is streaming, which tests upload through // the app to get themselves a playable VOD. 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) 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 // leaf by SPKI, a device installs the CA. vars += fmt.Sprintf("SERVER_HTTPS_URL=%s\nPDS_HTTPS_URL=%s\nE2E_PROXY_URL=%s\nE2E_TLS_SPKI=%s\nE2E_TLS_CA=%s\n", tlsEnv.StationURL(), tlsEnv.PDSURL(), tlsEnv.ProxyURL(), tlsEnv.spki, tlsEnv.caPath) } fmt.Print(vars)
<-ctx.Done() return g.Wait()}
// publishLexicons writes this build's lexicons, OAuth permission sets// included, into the dev-env lexicon authority account: the local version of// what the account behind _lexicon.stream.place holds in production. The PDS// resolves `include:place.stream.authFull` from it when the app logs in.func publishLexicons(ctx context.Context, env e2eDevEnv) error { xrpcc := &xrpc.Client{Host: env.PDSURL, Client: &aqhttp.TrustedClient} session, err := comatproto.ServerCreateSession(ctx, xrpcc, &comatproto.ServerCreateSession_Input{ Identifier: env.LexiconDID, Password: env.LexiconPassword, }) if err != nil { return fmt.Errorf("create session: %w", err) } xrpcc.Auth = &xrpc.AuthInfo{Did: session.Did, AccessJwt: session.AccessJwt, RefreshJwt: session.RefreshJwt} lexs, err := atproto.PublishedLexicons(ctx) if err != nil { return err } for _, lex := range lexs { rkey := lex.ID inp := spcomatproto.RepoCreateRecord_Input{ Collection: "com.atproto.lexicon.schema", Repo: session.Did, Rkey: &rkey, Record: &glex.LexiconTypeDecoder{Val: &atproto.SchemaFileWrapper{SchemaFile: *lex}}, } out := spcomatproto.RepoCreateRecord_Output{} if err := xrpcc.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.createRecord", map[string]any{}, inp, &out); err != nil { return fmt.Errorf("%s: %w", lex.ID, err) } } return nil}