diff --git a/internal/api/redact_drift_test.go b/internal/api/redact_drift_test.go index ce02a618..3602f41b 100644 --- a/internal/api/redact_drift_test.go +++ b/internal/api/redact_drift_test.go @@ -92,6 +92,7 @@ var leafSensitivity = map[string]sensitivity{ // are exactly where somebody debugging one would look. "destination.keyUnreadable": sPublic, "destination.kind": sPublic, + "destination.multitrack": sPublic, "destination.name": sPublic, "destination.platform": sPublic, "destination.position": sPublic, @@ -128,6 +129,27 @@ var leafSensitivity = map[string]sensitivity{ "destination.transport.rwTimeoutSeconds": sPublic, "destination.updatedAt": sPublic, "destination.url": sMasked, + "destination.vodProfile.delayMs": sPublic, + "destination.vodProfile.ducking.attackMs": sPublic, + "destination.vodProfile.ducking.ratio": sPublic, + "destination.vodProfile.ducking.releaseMs": sPublic, + "destination.vodProfile.ducking.target": sPublic, + "destination.vodProfile.ducking.thresholdDb": sPublic, + "destination.vodProfile.ducking.trigger": sPublic, + "destination.vodProfile.excludeRoles": sPublic, + "destination.vodProfile.loudness.rangeLu": sPublic, + "destination.vodProfile.loudness.targetLufs": sPublic, + "destination.vodProfile.loudness.truePeakDb": sPublic, + "destination.vodProfile.matrix.channel": sPublic, + "destination.vodProfile.matrix.gain": sPublic, + "destination.vodProfile.matrix.out": sPublic, + "destination.vodProfile.matrix.track": sPublic, + "destination.vodProfile.mode": sPublic, + "destination.vodProfile.normalize": sPublic, + "destination.vodProfile.sampleRate": sPublic, + "destination.vodProfile.tracks.enabled": sPublic, + "destination.vodProfile.tracks.gain": sPublic, + "destination.vodProfile.tracks.track": sPublic, "settings.alerts.retryAttempts": sPublic, "settings.automod.enabled": sPublic, "settings.automod.history.action": sPublic, diff --git a/internal/db/destinations.go b/internal/db/destinations.go index 54ce4885..0ee4cc87 100644 --- a/internal/db/destinations.go +++ b/internal/db/destinations.go @@ -133,6 +133,40 @@ type Destination struct { Enabled bool `json:"enabled"` AudioBitrate int `json:"audioBitrate"` // kbps Profile routing.Profile `json:"profile"` + // Multitrack opts this destination into Twitch Enhanced Broadcasting, which + // Amazon's IVS calls Multitrack Video: a negotiation at go-live that answers + // with an ingest endpoint, a minted stream key, and the audio tracks Twitch + // will accept. See internal/multitrack. + // + // FALSE IS THE RIGHT DEFAULT AND WILL STAY THE COMMON CASE. Twitch refuses + // any client without a supported GPU, by name, and polyemesis is built to be + // installed on the operator's own server -- a rented VPS has no GPU. Turning + // this on where negotiation cannot succeed is not a fault and is not + // punished: the destination falls back to the ordinary ingest and says so + // once. It is opt-in only because a network round trip at go-live should be + // something the operator asked for. + Multitrack bool `json:"multitrack,omitempty"` + // VODProfile is the SECOND audio mix -- the VOD track, separate from the + // live one, which is the whole ask of #141. + // + // Nil for every destination that has not opted in, which is nearly all of + // them, and nil produces byte for byte the filter graph and the argv the + // destination produced before this field existed. See routing.CompilePair, + // which compiles the pair, and ffmpeg.DestSpec.SecondAudioOutLabel, which + // maps and encodes it. + // + // A POINTER, NOT A VALUE, because "no second mix" and "a second mix that + // happens to be the zero profile" are different things and the zero profile + // is not valid anyway (Validate refuses it: no track enabled, no normalize + // mode, no sample rate). A value here would make every existing row look + // like it had asked for a broken second track. + // + // ON TWITCH THIS NEEDS Multitrack. The ordinary Twitch RTMP ingest takes one + // audio track; Enhanced Broadcasting is the only published path that takes + // two and says what the second is for. Nothing here enforces that pairing -- + // the engine reports it, because a setting that silently undoes itself is + // worse than one that explains itself. + VODProfile *routing.Profile `json:"vodProfile,omitempty"` // RenditionID selects the shared video encode this destination subscribes // to. nil is passthrough: no encode, no process, straight off the ingest // relay. Whatever the rendition, the destination still does -c:v copy plus @@ -797,6 +831,25 @@ func (d *DB) openStreamKey(enc []byte, plain string) (string, error) { return out, nil } +// marshalVODProfile renders the second (VOD) audio mix for storage. +// +// NIL BECOMES THE EMPTY STRING, NOT "null" AND NOT "{}". The read side treats +// empty as "no second track", so this is the half of that contract that has to +// agree: json.Marshal of a nil pointer produces the four bytes `null`, which is +// not empty, would take the decode branch, and would decode to a nil profile by +// a route the reader cannot distinguish from a corrupt value. One spelling of +// absence, checked at both ends. See TestADestinationWithNoVODMixStoresNoVODMix. +func marshalVODProfile(p *routing.Profile) (string, error) { + if p == nil { + return "", nil + } + b, err := json.Marshal(p) + if err != nil { + return "", fmt.Errorf("encode second (VOD) audio profile: %w", err) + } + return string(b), nil +} + func (d *DB) scanDestination(s interface{ Scan(...any) error }) (*Destination, error) { var ( dst Destination @@ -812,8 +865,13 @@ func (d *DB) scanDestination(s interface{ Scan(...any) error }) (*Destination, e // Same reasoning as complianceJSON: a row written before this column // existed must decode to a zero FacebookSettings, not fail the scan. facebookJSON = "{}" - created int64 - updated int64 + // The second (VOD) mix. Empty is "no second track" -- see the vod_profile + // migration for why empty and not "{}" -- so a row written before the + // column existed decodes to a nil VODProfile rather than to a profile + // that fails Validate. + vodProfileRaw = "" + created int64 + updated int64 ) err := s.Scan(&dst.ID, &dst.Name, &dst.Kind, &dst.Platform, &acct, &dst.URL, &dst.StreamKey, &streamEnc, @@ -825,6 +883,7 @@ func (d *DB) scanDestination(s interface{ Scan(...any) error }) (*Destination, e &dst.Resilience.MinBackoffSeconds, &dst.Resilience.MaxBackoffSeconds, &dst.Resilience.GiveUpAfter, &dst.Audio.Codec, &dst.Audio.Mono, &dst.Audio.Copy, &complianceJSON, &facebookJSON, + &dst.Multitrack, &vodProfileRaw, &dst.Position, &created, &updated) if err != nil { return nil, err @@ -890,6 +949,16 @@ func (d *DB) scanDestination(s interface{ Scan(...any) error }) (*Destination, e if err := json.Unmarshal([]byte(profileRaw), &dst.Profile); err != nil { return nil, fmt.Errorf("destination %d: decode routing profile: %w", dst.ID, err) } + if vodProfileRaw != "" { + // Named as the operator's setting, not as a column, because that is what + // the sentence has to mean to whoever reads it: "the VOD track on this + // destination is unreadable", not "column vod_profile failed to decode". + var vod routing.Profile + if err := json.Unmarshal([]byte(vodProfileRaw), &vod); err != nil { + return nil, fmt.Errorf("destination %d: decode second (VOD) audio profile: %w", dst.ID, err) + } + dst.VODProfile = &vod + } dst.CreatedAt = time.Unix(created, 0) dst.UpdatedAt = time.Unix(updated, 0) return &dst, nil @@ -903,6 +972,7 @@ const destColumns = `id, name, kind, platform, account_id, url, tr_no_duration_filesize, tr_mux_queue_packets, tr_mux_queue_bytes, tr_rw_timeout_seconds, rs_min_backoff_seconds, rs_max_backoff_seconds, rs_give_up_after, au_codec, au_mono, au_copy, compliance, facebook, + multitrack, vod_profile, position, created_at, updated_at` // The reads below, as whole compile-time constants. @@ -936,6 +1006,7 @@ const ( tr_rw_timeout_seconds=?, rs_min_backoff_seconds=?, rs_max_backoff_seconds=?, rs_give_up_after=?, au_codec=?, au_mono=?, au_copy=?, compliance=?, facebook=?, + multitrack=?, vod_profile=?, updated_at=? WHERE id=?` destUpdateQuery = `UPDATE destinations SET ` + destUpdateKeyCols + destUpdateCols destUpdateKeepKeyQuery = `UPDATE destinations SET ` + destUpdateCols @@ -1085,6 +1156,10 @@ func (d *DB) CreateDestination(dst *Destination) (*Destination, error) { if err != nil { return nil, err } + vodProfile, err := marshalVODProfile(dst.VODProfile) + if err != nil { + return nil, err + } keyEnc, keyPlain, err := d.sealStreamKey(dst.StreamKey) if err != nil { return nil, fmt.Errorf("seal stream key: %w", err) @@ -1111,8 +1186,9 @@ func (d *DB) CreateDestination(dst *Destination) (*Destination, error) { tr_no_duration_filesize, tr_mux_queue_packets, tr_mux_queue_bytes, tr_rw_timeout_seconds, rs_min_backoff_seconds, rs_max_backoff_seconds, rs_give_up_after, au_codec, au_mono, au_copy, compliance, facebook, + multitrack, vod_profile, position, created_at, updated_at) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, dst.Name, dst.Kind, dst.Platform, dst.AccountID, dst.URL, keyPlain, keyEnc, dst.BackupURL, backupPlain, backupEnc, dst.BackupIngestWanted, @@ -1123,6 +1199,7 @@ func (d *DB) CreateDestination(dst *Destination) (*Destination, error) { dst.Resilience.MinBackoffSeconds, dst.Resilience.MaxBackoffSeconds, dst.Resilience.GiveUpAfter, dst.Audio.Codec, dst.Audio.Mono, dst.Audio.Copy, string(compliance), string(facebook), + dst.Multitrack, vodProfile, dst.Position, now, now) if err != nil { return nil, err @@ -1158,6 +1235,10 @@ func (d *DB) UpdateDestination(dst *Destination) (*Destination, error) { if err != nil { return nil, err } + vodProfile, err := marshalVODProfile(dst.VODProfile) + if err != nil { + return nil, err + } // The key columns first, so that the two statements below differ only by a // prefix and the arguments line up with whichever one is used. args := []any{ @@ -1170,6 +1251,7 @@ func (d *DB) UpdateDestination(dst *Destination) (*Destination, error) { dst.Resilience.MinBackoffSeconds, dst.Resilience.MaxBackoffSeconds, dst.Resilience.GiveUpAfter, dst.Audio.Codec, dst.Audio.Mono, dst.Audio.Copy, string(compliance), string(facebook), + dst.Multitrack, vodProfile, time.Now().Unix(), dst.ID, } query := destUpdateQuery @@ -1441,6 +1523,22 @@ func (d *DB) MigrateDestinationExpertArgs() error { // row ran on, so an upgraded install emits the same command it did // yesterday for every destination that has not opted in. {"au_copy", `ALTER TABLE destinations ADD COLUMN au_copy INTEGER NOT NULL DEFAULT 0`}, + // Twitch Enhanced Broadcasting. 0 is "do not negotiate", which is what + // every existing row did, and it stays the common case: Twitch refuses a + // host with no supported GPU and most polyemesis installs are exactly + // that. See Destination.Multitrack. + {"multitrack", `ALTER TABLE destinations ADD COLUMN multitrack INTEGER NOT NULL DEFAULT 0`}, + // The second (VOD) audio mix, as one JSON blob for the reason compliance + // is one: it is the same shape as the `profile` column beside it and is + // edited as a unit. + // + // '' rather than '{}' is the no-op, and the difference matters. '{}' + // would decode to the zero routing.Profile -- no track enabled, no + // normalize mode, no sample rate -- which is a profile that fails + // Validate, so every upgraded row would carry a second mix that is + // broken rather than absent. '' decodes to nil, which is "no second + // track", which is what every existing row means. + {"vod_profile", `ALTER TABLE destinations ADD COLUMN vod_profile TEXT NOT NULL DEFAULT ''`}, // Compliance rides as one JSON blob rather than four columns: it is a // map plus two scalars, edited as a unit, and '{}' is "touch nothing". {"compliance", `ALTER TABLE destinations ADD COLUMN compliance TEXT NOT NULL DEFAULT '{}'`}, diff --git a/internal/db/multitrack_test.go b/internal/db/multitrack_test.go new file mode 100644 index 00000000..9cfdc64a --- /dev/null +++ b/internal/db/multitrack_test.go @@ -0,0 +1,181 @@ +package db + +import ( + "testing" + + "github.com/rainmanjam/polyemesis/internal/routing" +) + +// vodProfile is a second mix that is meaningfully DIFFERENT from the live one, +// so a round trip that quietly substituted the live profile would be visible. +func vodProfile() routing.Profile { + p := routing.DefaultProfile() + p.Tracks = []routing.TrackSel{{Track: 1, Enabled: true, Gain: 0.5}} + p.ExcludeRoles = []routing.TrackRole{routing.RoleMusic} + p.DelayMS = 120 + return p +} + +// TestADestinationRoundTripsItsSecondVODMix is the storage half of the VOD +// track: the operator's second mix has to survive a write and a read exactly, +// including the fields most likely to be dropped by a partial implementation. +// +// The gain of 0.5, the excluded role and the delay are all deliberate. A +// round trip that stored only the track list would pass a test that checked +// only the track list, and ExcludeRoles in particular is the DMCA switch -- +// silently losing it on the VOD mix is precisely the archive-carries-the-music +// failure the role exists to prevent. +// +// MUTATION: `vod_profile` dropped from destUpdateCols (the UPDATE column list) +// and from its argument. Observed: FAIL, "second mix after update: track 0 = +// {Track:1 Enabled:true Gain:0.5}, want {Track:1 Enabled:true Gain:0.25}" and +// the same again on re-read -- the update silently kept the old value. +// Restored from /tmp backup; `git diff --stat` clean. +// MUTATION: marshalVODProfile returns `"{}"` instead of `""` for nil. Observed: +// FAIL in TestADestinationWithNoVODMixStoresNoVODMix (below), not here. +// Restored from /tmp backup; `git diff --stat` clean. +func TestADestinationRoundTripsItsSecondVODMix(t *testing.T) { + d := testDB(t) + + in := validDest() + in.Multitrack = true + want := vodProfile() + in.VODProfile = &want + + created, err := d.CreateDestination(in) + if err != nil { + t.Fatalf("CreateDestination: %v", err) + } + if !created.Multitrack { + t.Error("multitrack was not stored") + } + if created.VODProfile == nil { + t.Fatal("the second (VOD) mix was not stored at all") + } + assertProfileEqual(t, "second mix after create", *created.VODProfile, want) + + // It must survive an UPDATE too, which is a different column list and the + // one most likely to be missed. + updated := *created + changed := vodProfile() + changed.Tracks[0].Gain = 0.25 + updated.VODProfile = &changed + got, err := d.UpdateDestination(&updated) + if err != nil { + t.Fatalf("UpdateDestination: %v", err) + } + if got.VODProfile == nil { + t.Fatal("the second (VOD) mix was lost by the update") + } + assertProfileEqual(t, "second mix after update", *got.VODProfile, changed) + + // And a re-read from the database, not just the value the writer returned. + reread, err := d.GetDestination(got.ID) + if err != nil { + t.Fatalf("GetDestination: %v", err) + } + if reread.VODProfile == nil { + t.Fatal("the second (VOD) mix did not survive a re-read") + } + assertProfileEqual(t, "second mix on re-read", *reread.VODProfile, changed) + if !reread.Multitrack { + t.Error("multitrack did not survive a re-read") + } +} + +// TestADestinationWithNoVODMixStoresNoVODMix is the compatibility half, and it +// is the one that matters to every existing install: nearly every destination +// has no second mix, and "no second mix" has to read back as NIL rather than as +// a profile that happens to be empty. +// +// The distinction is not cosmetic. The zero routing.Profile fails Validate -- +// no track enabled, no normalize mode, no sample rate -- so if absence decoded +// to a zero profile instead of nil, every row written before this column +// existed would come back carrying a second audio track that cannot compile. +// routing.CompilePair would then warn about a VOD mix the operator never asked +// for, on every destination in the install. +// +// MUTATION: marshalVODProfile returns `"{}"` for nil instead of `""`. Observed: +// FAIL, "a destination with no second mix came back with one". Restored from +// /tmp backup; `git diff --stat` clean. +// MUTATION: scanDestination's `if vodProfileRaw != ""` changed to an +// unconditional decode. Observed: FAIL, "decode second (VOD) audio profile: +// unexpected end of JSON input". Restored from /tmp backup; clean. +func TestADestinationWithNoVODMixStoresNoVODMix(t *testing.T) { + d := testDB(t) + + created, err := d.CreateDestination(validDest()) + if err != nil { + t.Fatalf("CreateDestination: %v", err) + } + if created.VODProfile != nil { + t.Errorf("a destination with no second mix came back with one: %+v", *created.VODProfile) + } + if created.Multitrack { + t.Error("multitrack defaulted to on; it must be opt-in") + } + + reread, err := d.GetDestination(created.ID) + if err != nil { + t.Fatalf("GetDestination: %v", err) + } + if reread.VODProfile != nil { + t.Errorf("a re-read invented a second mix: %+v", *reread.VODProfile) + } + + // Turning it on and then off again must leave nothing behind. A clear that + // wrote "{}" would read back as a broken second track rather than as none. + on := *reread + p := vodProfile() + on.VODProfile = &p + on.Multitrack = true + if _, err := d.UpdateDestination(&on); err != nil { + t.Fatalf("UpdateDestination (on): %v", err) + } + off := *reread + off.VODProfile = nil + off.Multitrack = false + cleared, err := d.UpdateDestination(&off) + if err != nil { + t.Fatalf("UpdateDestination (off): %v", err) + } + if cleared.VODProfile != nil { + t.Errorf("clearing the second mix left one behind: %+v", *cleared.VODProfile) + } + if cleared.Multitrack { + t.Error("clearing multitrack left it on") + } +} + +// assertProfileEqual compares the fields a second mix can actually differ in. +// Written out rather than reflect.DeepEqual so that a failure names WHICH +// setting was lost, which is the thing a person reading the failure needs. +func assertProfileEqual(t *testing.T, what string, got, want routing.Profile) { + t.Helper() + if len(got.Tracks) != len(want.Tracks) { + t.Fatalf("%s: got %d tracks, want %d", what, len(got.Tracks), len(want.Tracks)) + } + for i := range want.Tracks { + if got.Tracks[i] != want.Tracks[i] { + t.Errorf("%s: track %d = %+v, want %+v", what, i, got.Tracks[i], want.Tracks[i]) + } + } + if got.DelayMS != want.DelayMS { + t.Errorf("%s: delayMs = %d, want %d", what, got.DelayMS, want.DelayMS) + } + if got.Normalize != want.Normalize { + t.Errorf("%s: normalize = %q, want %q", what, got.Normalize, want.Normalize) + } + if got.SampleRate != want.SampleRate { + t.Errorf("%s: sampleRate = %d, want %d", what, got.SampleRate, want.SampleRate) + } + if len(got.ExcludeRoles) != len(want.ExcludeRoles) { + t.Fatalf("%s: got %d excluded roles, want %d -- losing this is the "+ + "archive-carries-the-music failure", what, len(got.ExcludeRoles), len(want.ExcludeRoles)) + } + for i := range want.ExcludeRoles { + if got.ExcludeRoles[i] != want.ExcludeRoles[i] { + t.Errorf("%s: excluded role %d = %q, want %q", what, i, got.ExcludeRoles[i], want.ExcludeRoles[i]) + } + } +} diff --git a/internal/engine/destinations.go b/internal/engine/destinations.go index caf37d8e..c2e3b127 100644 --- a/internal/engine/destinations.go +++ b/internal/engine/destinations.go @@ -79,6 +79,26 @@ func (e *Engine) planDestinations(rows []*db.Destination, wantRends map[int64]st compile = routing.CompileProvisional } compiled, cerr := compile(row.Profile, src) + // The second (VOD) audio mix, when this destination asked for one. + // + // NOT on the provisional path. A provisional compile is already running + // on a guessed layout and saying so; adding a second guessed mix on top + // doubles what is approximate while the operator is being told the first + // one is unreliable. The VOD track comes back by itself on the next + // reconcile after a probe succeeds, which is the same moment the live + // mix stops being provisional. + if !provisional && cerr == nil && row.VODProfile != nil { + paired, perr := routing.CompilePair(row.Profile, row.VODProfile, src) + if perr != nil { + // CompilePair fails only where Compile just failed, so reaching + // here means the live mix is broken too and cerr would have been + // set. Belt and braces: report it rather than silently publish + // one track. + cerr = perr + } else { + compiled = paired.Result + } + } if cerr != nil { p.err = cerr.Error() } else { @@ -281,9 +301,16 @@ func destSpecFor(log *slog.Logger, row *db.Destination, compiled routing.Result, RelayURL: relayURL, FilterComplex: compiled.FilterComplex, AudioOutLabel: compiled.OutLabel, - AudioBitrate: row.AudioBitrate, - SampleRate: row.Profile.SampleRate, - CopyVideo: true, + // The second (VOD) audio track. Empty for every destination that has + // not opted in, which is nearly all of them, and empty produces byte + // for byte the command it produced before this existed. It reaches the + // backup feed through the same struct, so a redundant feed carries the + // same two tracks as the primary rather than silently dropping one -- + // which is the asymmetry this function's doc comment exists about. + SecondAudioOutLabel: compiled.SecondOutLabel, + AudioBitrate: row.AudioBitrate, + SampleRate: row.Profile.SampleRate, + CopyVideo: true, // A negative routing delay pulls audio ahead of picture, which no // audio filter can do, so the compiler hands the amount over here // and the video is held back instead. diff --git a/internal/engine/minted_key_secret_test.go b/internal/engine/minted_key_secret_test.go new file mode 100644 index 00000000..1575d853 --- /dev/null +++ b/internal/engine/minted_key_secret_test.go @@ -0,0 +1,126 @@ +package engine + +import ( + "strings" + "testing" + + "github.com/rainmanjam/polyemesis/internal/alerts" + "github.com/rainmanjam/polyemesis/internal/db" +) + +// The operator's own key, and the minted key Twitch answers a successful +// Enhanced Broadcasting negotiation with. +// +// The minted value is SYNTHETIC but its structure is the measured one, and the +// structure is the whole point of this test: v1_<64 hex signature>_<8 hex>__. The original is a SUFFIX of the +// minted key, which is what makes the naive protection fail quietly. +const ( + mintedOperatorKey = "live_2468013579_TheOperatorTypedThis" + mintedSignature = "v1_" + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + + "_a1b2c3d4_" + "7b2276223a312c2262223a343832307d_" + mintedKeyWhole = mintedSignature + mintedOperatorKey +) + +// TestTheMintedKeyIsMaskedWholeAndNotJustItsTail is the guard on a credential +// that only exists at go-live, and it is written against the specific way this +// protection fails: PARTIALLY. +// +// Twitch mints a 312-character stream key on a successful negotiation and it +// ends with the operator's own. polyemesis publishes with the minted value, so +// it reaches an FFmpeg command line and therefore reaches process.log, the +// monitoring page's argv, and every error the supervisor renders. +// supervisor.Process removes exactly the literals it was handed in Spec.Secrets +// and alerts.SecretSet.Scrub does that with strings.ReplaceAll. +// +// So registering the ORIGINAL key -- which destSecrets did, and which is what a +// reasonable person would assume covers "the stream key" -- masks the minted +// key's last segment and leaves +// +// v1_<64 hex signature>_<8 hex>__ +// +// standing in the log. That is a partially redacted live credential, and it +// reads as protection to anyone glancing at the file, which is worse than no +// redaction at all. #310 and #324 were both this class. +// +// The assertion is therefore NOT "the log does not contain the original key" -- +// that passes on the broken version. It is that the SIGNATURE PREFIX is gone +// too, which only happens if the minted value was registered in its own right. +// +// MUTATION: `out = append(out, extra...)` deleted from destSecrets, i.e. the +// minted key is never registered and only the original is. Observed: FAIL, +// "the minted key's signature survived scrubbing" with the masked-tail form +// above printed. Restored from /tmp backup; `git diff --stat` clean. +// MUTATION: destSecrets given the minted key but wireSpellings dropped. +// Observed: PASS -- so the test does NOT depend on wireSpellings, which is +// correct: that expands truncations and is a separate concern. +func TestTheMintedKeyIsMaskedWholeAndNotJustItsTail(t *testing.T) { + row := &db.Destination{ + Kind: db.DestRTMP, + URL: "rtmps://ingest.global-contribute.live-video.net/app", + StreamKey: mintedOperatorKey, + } + + // What the supervisor would be told to remove, WITH the minted key declared. + set := alerts.NewSecretSet(nil, destSecrets(row, mintedKeyWhole)...) + + // A log line of the shape FFmpeg actually produces when a publish fails: the + // whole publish URL, minted key and all. + line := "rtmps://ingest.global-contribute.live-video.net/app/" + mintedKeyWhole + + " Failed to open output" + got := set.Scrub(line) + + if strings.Contains(got, mintedKeyWhole) { + t.Fatalf("the whole minted key survived scrubbing:\n%s", got) + } + // THE LOAD-BEARING ASSERTION. The tail is masked even by the broken version, + // because the original is a suffix of the minted key. Only registering the + // minted value removes the signature. + if strings.Contains(got, mintedSignature) { + t.Errorf("the minted key's signature survived scrubbing, so the log carries a "+ + "partially redacted live credential:\n%s", got) + } + if strings.Contains(got, "0123456789abcdef") { + t.Errorf("the signature hex is still in the log line:\n%s", got) + } + // And the ordinary key is still covered, or this traded one leak for another. + if strings.Contains(got, mintedOperatorKey) { + t.Errorf("the operator's own key survived scrubbing:\n%s", got) + } +} + +// TestRegisteringOnlyTheOriginalKeyLeavesTheSignatureStanding is the negative +// control, and it exists because the test above could pass for the wrong +// reason -- alerts.Redact runs a residual pattern pass, and if THAT were what +// removed the signature then destSecrets would be untested and the protection +// would be an accident. +// +// This asserts the gap is real: with only the original registered, the +// signature IS still there. If this test ever starts failing, the protection +// has moved somewhere else and the comment above needs rewriting rather than +// the test deleting. +// +// MUTATION: not applicable in the usual direction -- this test asserts the +// ABSENCE of protection, so it is mutated by ADDING the minted key to the set, +// which is the fix. Observed with `destSecrets(row, mintedKeyWhole)`: FAIL, +// "the signature was already gone without registering the minted key". +// Restored from /tmp backup; `git diff --stat` clean. +func TestRegisteringOnlyTheOriginalKeyLeavesTheSignatureStanding(t *testing.T) { + row := &db.Destination{ + Kind: db.DestRTMP, + URL: "rtmps://ingest.global-contribute.live-video.net/app", + StreamKey: mintedOperatorKey, + } + // Deliberately NOT passing the minted key: this is the naive protection. + set := alerts.NewSecretSet(nil, destSecrets(row)...) + got := set.Scrub("publishing to /app/" + mintedKeyWhole) + + if !strings.Contains(got, mintedSignature) { + t.Fatalf("the signature was already gone without registering the minted key, so "+ + "TestTheMintedKeyIsMaskedWholeAndNotJustItsTail is not testing what it says:\n%s", got) + } + // And the tail IS masked, which is precisely why the gap is easy to miss. + if strings.Contains(got, mintedOperatorKey) { + t.Errorf("expected the original key to be masked even here: %s", got) + } +} diff --git a/internal/engine/secrets.go b/internal/engine/secrets.go index a264bae9..11ec4241 100644 --- a/internal/engine/secrets.go +++ b/internal/engine/secrets.go @@ -26,11 +26,28 @@ import ( // Both feeds get the same set. The backup argv is built from the same row and // splices the same expert text, so a set that covered only the primary would // leave dest:N:backup leaking on exactly the routes dest:N no longer does. -func destSecrets(row *db.Destination) []string { +// +// extra carries credentials that are NOT on the row because they did not exist +// until go-live. Today that is the Twitch Enhanced Broadcasting minted key, and +// it is a variadic rather than another row field because it is a fact about +// this run of this process, not about the destination -- a new one is minted +// per negotiation, and storing it would be storing a credential that is stale +// by the next broadcast. +// +// THE MINTED KEY NEEDS ITS OWN ENTRY AND CANNOT INHERIT THE ORIGINAL'S. +// SecretSet.Scrub is a substring replace, and the minted key ENDS WITH the +// operator's original -- v1___. Registering only +// the original therefore masks the last segment and leaves the signature and +// the manifest standing, which is a partially redacted credential in a log +// file: enough to identify the broadcast, and exactly the shape of half-fix +// that reads as protection. Measured and pinned by +// TestTheMintedKeyIsMaskedWholeAndNotJustItsTail. +func destSecrets(row *db.Destination, extra ...string) []string { if row == nil { return nil } out := []string{row.StreamKey, row.BackupStreamKey} + out = append(out, extra...) out = append(out, urlSecrets(row.URL)...) out = append(out, urlSecrets(row.BackupURL)...) out = append(out, expertArgsSecrets(row.ExtraInputArgs)...) diff --git a/internal/multitrack/negotiate.go b/internal/multitrack/negotiate.go new file mode 100644 index 00000000..62940942 --- /dev/null +++ b/internal/multitrack/negotiate.go @@ -0,0 +1,183 @@ +package multitrack + +import ( + "context" +) + +// Outcome is what one go-live negotiation decided, and it is deliberately not +// an (Outcome, error) pair. +// +// NOT NEGOTIATING IS AN ORDINARY RESULT, NOT A FAILURE. Twitch refuses any +// client without a supported GPU, and polyemesis is built to be installed on +// the operator's own server -- a rented VPS has none. So on most installs the +// fallback IS the path, every time, for ever. A function that returned an error +// for it would make the ordinary case look broken: it would be logged at error +// level, counted as a fault, and retried by somebody who assumed a non-nil +// error meant something had gone wrong. Nothing here can fail in a way the +// caller must handle -- the caller either publishes to Target or publishes to +// the destination's own URL, and both are correct. +type Outcome struct { + // Target is where to publish, valid only when Use is true. Its Key is a + // CREDENTIAL and is the minted one; see the Use comment. + Target Target + + // Use reports whether the caller must publish to Target instead of the + // destination's stored URL. + // + // A TRUE HERE IS NOT EVIDENCE THE STREAM KEY IS VALID. Measured: the live + // endpoint returned a successful negotiation, with a full ladder and a + // minted key, for a plainly invalid stream key. Validation happens at + // PUBLISH, not at negotiation. So a caller must not read a successful + // Negotiate as "the credential works" -- the failure will arrive later, at + // the ingest, and anything that reported the key as verified here will have + // made it harder to diagnose rather than easier. + // + // WHEN THIS IS TRUE THE MINTED KEY IS MANDATORY, not preferred. Twitch + // answers a successful negotiation with a new 312-character stream key that + // carries the agreed ladder signed inside it, ending with the operator's + // original. Publishing with the operator's own key instead would connect -- + // which is what makes this dangerous rather than merely wrong -- and send a + // ladder the ingest never agreed to. Target.Key is that minted value; it + // must not be reassembled from the destination row. + Use bool + + // Verdict is Twitch's answer, for a caller that wants to distinguish "we + // never asked" from "we asked and were refused". Refused whenever Use is + // false. + Verdict Verdict + + // Note is ONE sentence for the operator, and it is always set -- including + // on the quiet success path, where it says nothing happened. + // + // IT CARRIES NO CREDENTIAL, and that is enforced here rather than inherited. + // See the scrubbing closure in Negotiate for what is defence and what is a + // measurement -- the distinction matters, and an earlier version of this + // comment got it wrong in a way that would have sent a reader to the wrong + // field. + // + // It is a note rather than a warning on purpose. An operator on a GPU-less + // server has not misconfigured anything and must not be shown a fault. + Note string + + // Divergences are the places the negotiated configuration departs from what + // was asked for. ADVISORY ONLY: they annotate, they never block. An optional + // VOD track must never veto a working broadcast, so a divergence is reported + // beside a destination that is publishing, not instead of one. + Divergences []Divergence +} + +// noteNoGPU is the fallback sentence for the majority install, and its wording +// is the whole point: nothing here is a fault. +const noteNoGPU = "Enhanced Broadcasting was not requested: it needs a supported GPU, " + + "which this server has not been told it has, so this destination is publishing to the ordinary Twitch ingest." + +// Negotiate asks Twitch for an Enhanced Broadcasting configuration and decides +// whether to publish to it. +// +// It is called once per destination per go-live, on the path between the +// operator pressing the button and anything reaching a viewer, which is why +// Client's timeout matters and why the no-GPU case below does not make the call +// at all. +// +// THE NO-GPU SHORT CIRCUIT IS A MEASURED SHORTCUT, NOT AN ASSUMPTION. Twitch +// was observed refusing, by name, a request that sent no GPU information at all +// ("Your broadcast software (polyemesis) did not send GPU Information"), an +// Intel iGPU, an unrecognised vendor ID, and an out-of-date driver. A request +// with no GPU therefore has one possible answer, and spending a network round +// trip at go-live to be told it is the answer we already know would delay every +// broadcast on every GPU-less install to learn nothing. Delete these four lines +// and the behaviour is identical but slower -- which is what makes it safe to +// delete if Twitch ever changes its mind. +// +// Nothing here measures hardware; Capabilities explains why not. The GPU facts +// come from the operator's configuration, and their absence is the default. +func Negotiate(ctx context.Context, c *Client, streamKey string, a Ask) Outcome { + // EVERY note leaves through here, scrubbed. THIS IS DEFENCE, NOT A FIX FOR AN + // OBSERVED LEAK, and the difference is worth stating precisely because the + // first version of this comment claimed the latter and was wrong. + // + // WHAT WAS MEASURED, against the live endpoint with a distinctive canary + // sent as `authentication`: + // + // - status.html_en_us echoes client.name, NOT the stream key. A refusal + // for missing canvases came back naming the broadcast software, and the + // canary did not appear in it under any refusal that could be produced. + // - The key that does come back is in ingest_endpoints[].authentication: + // the 312-character MINTED key on the success path, which ends with the + // original. That is the credential this response carries. + // + // So html_en_us is not a known key channel. It is scrubbed anyway because it + // is ATTACKER-INFLUENCED TEXT FROM A THIRD PARTY that polyemesis renders to + // an operator, and it is built by quoting request fields back -- a habit that + // needs only one more field to become a leak. Scrubbing text we do not + // control is cheap; discovering later that the habit grew is not. + // + // WHAT THIS CLOSURE DOES NOT PROTECT is the minted key, and nothing here + // could: it never appears in a Note, only in Outcome.Target.Key. Registering + // the original key as a secret does NOT cover the minted one either -- + // SecretSet.Scrub is a substring replace, so the original masks only the + // minted key's last segment and leaves the signature and manifest in the + // clear. Whoever publishes with Target.Key must register that value in its + // own right; see engine.destSecrets and + // TestTheMintedKeyIsMaskedWholeAndNotJustItsTail. + out := func(o Outcome) Outcome { + o.Note = scrub(o.Note, streamKey) + return o + } + + if len(a.Hardware.GPU) == 0 { + return out(Outcome{Verdict: Refused, Note: noteNoGPU}) + } + + cfg, err := c.Fetch(ctx, streamKey, NewRequest(a)) + if err != nil { + // Client.Fetch has already scrubbed the key out of every error it + // builds, including the *url.Error the transport hands back -- which is + // the shape that leaked in #310 and #324. Carried through as it stands + // rather than re-wrapped with anything of ours, because the only things + // this function knows that Fetch does not are the key and the target, + // and neither may appear in a note. + return out(Outcome{ + Verdict: Refused, + Note: "Enhanced Broadcasting could not be negotiated, so this destination is publishing " + + "to the ordinary Twitch ingest: " + err.Error(), + }) + } + + verdict, advice := cfg.Verdict() + if verdict == Refused { + return out(Outcome{ + Verdict: Refused, + Note: "Enhanced Broadcasting is not available for this broadcast, so this destination is " + + "publishing to the ordinary Twitch ingest: " + advice, + }) + } + + target, rerr := cfg.Resolve(streamKey) + if rerr != nil { + // A configuration that passed Verdict but carries no endpoint anyone can + // publish to. Falling back rather than failing, for the same reason as + // everywhere else here: the ordinary ingest works. + return out(Outcome{ + Verdict: Refused, + Note: "Enhanced Broadcasting returned a configuration with no usable ingest, so this " + + "destination is publishing to the ordinary Twitch ingest: " + rerr.Error(), + }) + } + + note := "Enhanced Broadcasting is in use for this destination." + if advice != "" { + // Advisory: Twitch agreed and said something about it. Shown, not acted + // on -- obs-studio puts a modal here and offers to abort, and polyemesis + // has no operator standing at the machine when a scheduled broadcast + // starts. + note = "Enhanced Broadcasting is in use for this destination, and Twitch added: " + advice + } + return out(Outcome{ + Target: target, + Use: true, + Verdict: verdict, + Note: note, + Divergences: Reconcile(a, cfg), + }) +} diff --git a/internal/multitrack/negotiate_test.go b/internal/multitrack/negotiate_test.go new file mode 100644 index 00000000..1044cb7e --- /dev/null +++ b/internal/multitrack/negotiate_test.go @@ -0,0 +1,309 @@ +package multitrack + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" +) + +// operatorKey is the key an operator typed into a destination row. +const operatorKey = "live_123456789_OperatorTypedThisOne" + +// mintedKey is a SYNTHETIC stand-in for the 312-character credential Twitch +// mints on a successful negotiation. +// +// Synthetic on purpose and it cannot be otherwise: the captured fixture +// testdata/negotiated-one-rendition.json has its `authentication` emptied, +// because a real minted key is a live credential and committing one would put +// a working stream key in the repository -- which is the thing #310 and #324 +// were about. So the SHAPE is reproduced from the measurement written down in +// IngestEndpoint.Authentication -- v1_<64 hex>_<8 hex>__ -- and the property under test is structural: whatever came +// back in that field is what gets published, and the operator's own key is not. +// A test built on the real value would assert the same thing. +var mintedKey = "v1_" + strings.Repeat("a1b2c3d4", 8) + "_deadbeef_" + + "7b2276223a312c2262223a343832307d_" + operatorKey + +// gpuAsk is an Ask that will actually reach the network: Negotiate short +// circuits an Ask with no GPU without making the call at all, so every test of +// the request path has to supply one. +func gpuAsk() Ask { + return Ask{ + Version: "test", + VODAudio: true, + Canvas: Canvas{Width: 1920, Height: 1080, CanvasWidth: 1920, CanvasHeight: 1080, + Framerate: Framerate{Numerator: 30, Denominator: 1}}, + Hardware: Capabilities{GPU: []GPU{{ + Model: "NVIDIA GeForce RTX 4080", VendorID: 4318, DeviceID: 9988, + DedicatedVideoMemory: 16 * 1024 * 1024 * 1024, + }}}, + } +} + +// serveFixture answers every request with the named testdata file, and records +// the body it was sent so a test can assert on what went out. +func serveFixture(t *testing.T, name string, mutate func(map[string]any)) (*Client, *[]byte) { + t.Helper() + raw, err := os.ReadFile(filepath.Join("testdata", name)) + if err != nil { + t.Fatalf("read fixture %s: %v", name, err) + } + if mutate != nil { + var doc map[string]any + if err := json.Unmarshal(raw, &doc); err != nil { + t.Fatalf("fixture %s is not JSON: %v", name, err) + } + mutate(doc) + if raw, err = json.Marshal(doc); err != nil { + t.Fatalf("re-encode fixture: %v", err) + } + } + var sent []byte + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b := make([]byte, r.ContentLength) + _, _ = r.Body.Read(b) + sent = b + w.Header().Set("Content-Type", "application/json") + // 200 on every path, including the refusal, because that is what the + // live endpoint does and it is the whole hazard this package exists for. + w.WriteHeader(http.StatusOK) + _, _ = w.Write(raw) + })) + t.Cleanup(srv.Close) + return &Client{BaseURL: srv.URL}, &sent +} + +// withMintedKey puts a minted credential into both ingest endpoints of the +// negotiated fixture, which ships with the field emptied. +func withMintedKey(doc map[string]any) { + for _, e := range doc["ingest_endpoints"].([]any) { + e.(map[string]any)["authentication"] = mintedKey + } +} + +// TestASuccessfulNegotiationPublishesWithTheMintedKey is the most important +// assertion in this package, and it is about a failure that WORKS. +// +// On success Twitch mints a new stream key carrying the agreed ladder signed +// inside it, ending with the operator's original. Publishing with the +// operator's own key instead does not fail loudly: it CONNECTS, and sends a +// ladder the ingest never agreed to. So "did it publish?" cannot tell the two +// apart, and neither can a test that only checks the hostname moved. The +// assertion has to be that the minted value specifically is what came out, and +// that the operator's own key is not the whole of it. +// +// MUTATION: Resolve's `if ep.Authentication != ""` disabled, so the minted key +// is ignored and the operator's own is published. Observed: FAIL, "published +// key is neither the minted key nor the operator's: +// \"live_123456789_OperatorTypedThisOne?clientConfigId=...\"" -- the config id +// is still appended, which is precisely why the assertion checks the signed +// v1_ prefix rather than equality with the operator's key. Restored from /tmp +// backup; `git diff --stat` clean. +func TestASuccessfulNegotiationPublishesWithTheMintedKey(t *testing.T) { + c, _ := serveFixture(t, "negotiated-one-rendition.json", withMintedKey) + + out := Negotiate(context.Background(), c, operatorKey, gpuAsk()) + + if !out.Use { + t.Fatalf("a negotiated configuration was not used; note: %s", out.Note) + } + if out.Verdict != Negotiated { + t.Errorf("verdict = %q, want %q", out.Verdict, Negotiated) + } + if !strings.HasPrefix(out.Target.Key, mintedKey) { + if out.Target.Key == operatorKey || strings.HasPrefix(operatorKey, out.Target.Key) { + t.Fatalf("published key is the operator's own, not the minted one: %q", out.Target.Key) + } + t.Fatalf("published key is neither the minted key nor the operator's: %q", out.Target.Key) + } + // The minted key ENDS with the operator's own, so a test asserting only + // "contains the operator key" would pass on the wrong value too. What + // separates them is the signed prefix. + if !strings.HasPrefix(out.Target.Key, "v1_") { + t.Errorf("published key has lost its signature prefix: %q", out.Target.Key) + } + // The config id travels to the ingest on the key, which is how Twitch knows + // which negotiated ladder is arriving. + if !strings.Contains(out.Target.Key, "49456f79-a985-4011-941f-3cde9897a0c6") { + t.Errorf("the negotiated config id is not on the published key: %q", out.Target.Key) + } + // And the endpoint is the OTHER host -- not live.twitch.tv. + if !strings.Contains(out.Target.URL, "global-contribute") { + t.Errorf("target URL is not the multitrack ingest: %q", out.Target.URL) + } +} + +// TestAGPULessHostFallsBackQuietlyAndNeverCallsTwitch is the majority install: +// a rented VPS with no GPU. It must fall back, must not look like a fault, and +// must not spend a network round trip at go-live to be told what is already +// measured. +// +// The "never called" half is the load-bearing one. A test that only checked +// Use==false would pass on an implementation that made the call, waited for the +// timeout, and fell back -- which is the same answer arrived at slowly, on the +// path between the operator pressing go-live and anything reaching a viewer. +// +// MUTATION: the `len(a.Hardware.GPU) == 0` short circuit deleted from +// Negotiate. Observed: FAIL, "a GPU-less host still called Twitch". Restored +// from /tmp backup; `git diff --stat` clean. +func TestAGPULessHostFallsBackQuietlyAndNeverCallsTwitch(t *testing.T) { + var called bool + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{}`)) + })) + t.Cleanup(srv.Close) + + ask := gpuAsk() + ask.Hardware.GPU = nil // the VPS + + out := Negotiate(context.Background(), &Client{BaseURL: srv.URL}, operatorKey, ask) + + if called { + t.Error("a GPU-less host still called Twitch; the answer is already measured and the call costs go-live latency") + } + if out.Use { + t.Error("a GPU-less host was told to publish to the multitrack ingest") + } + if out.Target.Key != "" || out.Target.URL != "" { + t.Errorf("a fallback outcome carries a target: %+v", out.Target) + } + if out.Note == "" { + t.Fatal("the fallback said nothing at all; the operator has to be told which ingest is in use") + } + // The wording is the requirement, not decoration: this is the normal path + // and must not read as a fault. + for _, forbidden := range []string{"error", "failed", "fault", "invalid"} { + if strings.Contains(strings.ToLower(out.Note), forbidden) { + t.Errorf("the ordinary GPU-less fallback note reads as a fault (%q): %s", forbidden, out.Note) + } + } +} + +// TestARefusalIsATwoHundredAndFallsBack is the hazard the whole package exists +// for: Twitch answers a refusal with HTTP 200 and puts the verdict in +// status.result. A client that read the status code would publish to a +// configuration it was refused. +// +// MUTATION: Config.Verdict's StatusError case changed to fall through to +// Negotiated. Observed: FAIL, "a refused configuration was used". Restored from +// /tmp backup; `git diff --stat` clean. +func TestARefusalIsATwoHundredAndFallsBack(t *testing.T) { + c, _ := serveFixture(t, "refused-no-gpu.json", nil) + + out := Negotiate(context.Background(), c, operatorKey, gpuAsk()) + + if out.Use { + t.Fatal("a refused configuration was used; the HTTP status was 200 and the refusal is in status.result") + } + if out.Verdict != Refused { + t.Errorf("verdict = %q, want %q", out.Verdict, Refused) + } + // Twitch's own explanation has to reach the operator -- it is the only + // statement of why that exists. + if !strings.Contains(out.Note, "GPU") { + t.Errorf("the refusal note does not carry Twitch's explanation: %s", out.Note) + } + if strings.Contains(out.Note, operatorKey) { + t.Error("the operator's stream key is in the note") + } +} + +// TestNoOutcomeEverCarriesTheStreamKeyInItsNote is the security guard, and it is +// swept across every path rather than asserted on one: a note is built from +// Twitch's own text, which QUOTES REQUEST FIELDS BACK, and the request carries +// the stream key. That is the shape of leak #310 and #324 were. +// +// MUTATION: the transport-error branch changed to append the target URL and key +// to the note. Observed: FAIL on the "transport failure" subcase, "note carries +// the stream key". Restored from /tmp backup; `git diff --stat` clean. +func TestNoOutcomeEverCarriesTheStreamKeyInItsNote(t *testing.T) { + // A server that echoes the key back inside Twitch's own explanation field, + // which is exactly what the live endpoint does. + echo := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"meta":{"service":"IVS"},"status":{"result":"error",` + + `"html_en_us":"Your key ` + operatorKey + ` was rejected"},` + + `"encoder_configurations":[],"audio_configurations":{}}`)) + })) + t.Cleanup(echo.Close) + + // A server that is not there at all, for the transport-error path where the + // *url.Error carries the full URL. + dead := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + deadURL := dead.URL + dead.Close() + + for _, tc := range []struct { + name string + c *Client + }{ + {"twitch echoes the key back", &Client{BaseURL: echo.URL}}, + {"transport failure", &Client{BaseURL: deadURL}}, + {"malformed response", nil}, + } { + t.Run(tc.name, func(t *testing.T) { + c := tc.c + if c == nil { + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"meta":`)) + })) + t.Cleanup(bad.Close) + c = &Client{BaseURL: bad.URL} + } + out := Negotiate(context.Background(), c, operatorKey, gpuAsk()) + if out.Use { + t.Error("a broken negotiation was used") + } + if out.Note == "" { + t.Fatal("no note") + } + if strings.Contains(out.Note, operatorKey) { + t.Errorf("note carries the stream key: %s", out.Note) + } + }) + } +} + +// TestTheRequestAsksForTheVODTrack pins the one preference the whole feature +// turns on. Twitch populates audio_configurations.vod only when it is asked to, +// so a request that quietly dropped this would negotiate successfully, publish +// happily, and carry one audio track. +// +// MUTATION: NewRequest's `VODTrackAudio: a.VODAudio` changed to a literal +// false. Observed: FAIL, "the request did not ask for the VOD audio track". +// Restored from /tmp backup; `git diff --stat` clean. +func TestTheRequestAsksForTheVODTrack(t *testing.T) { + c, sent := serveFixture(t, "negotiated-one-rendition.json", withMintedKey) + + out := Negotiate(context.Background(), c, operatorKey, gpuAsk()) + if !out.Use { + t.Fatalf("negotiation did not succeed: %s", out.Note) + } + if len(*sent) == 0 { + t.Fatal("no request body was captured, so this test is asserting nothing") + } + + var body map[string]any + if err := json.Unmarshal(*sent, &body); err != nil { + t.Fatalf("request body is not JSON: %v\n%s", err, *sent) + } + prefs, ok := body["preferences"].(map[string]any) + if !ok { + t.Fatalf("no preferences in the request: %s", *sent) + } + if vod, _ := prefs["vod_track_audio"].(bool); !vod { + t.Errorf("the request did not ask for the VOD audio track: %v", prefs["vod_track_audio"]) + } + // And the key really did travel, or the fixture proves nothing about auth. + if body["authentication"] != operatorKey { + t.Errorf("authentication = %v, want the operator's stream key", body["authentication"]) + } +} diff --git a/internal/routing/filtergraph.go b/internal/routing/filtergraph.go index 9f151470..bc45e9fd 100644 --- a/internal/routing/filtergraph.go +++ b/internal/routing/filtergraph.go @@ -12,6 +12,21 @@ import ( // destination command builder maps it with -map "[aout]". const OutLabel = "aout" +// ns is a label namespace. Every intermediate label a compile emits goes +// through it, so that two graphs can share one filter_complex without colliding. +// +// The empty namespace reproduces the labels this package emitted before it +// existed, BYTE FOR BYTE -- that is asserted, not assumed, because a destination +// that gains a second mix must not have its first mix quietly rewritten. See +// TestTheEmptyNamespaceIsByteIdenticalToTheSingleMixGraph. +type ns struct{ prefix string } + +// of namespaces a fixed label. +func (n ns) of(name string) string { return n.prefix + name } + +// track names the per-track pan output for ingest track t. +func (n ns) track(t int) string { return fmt.Sprintf("%sa_t%d", n.prefix, t) } + // Result is a compiled routing profile. type Result struct { // FilterComplex is the full -filter_complex argument, ready to hand to @@ -19,6 +34,16 @@ type Result struct { FilterComplex string `json:"filterComplex"` // OutLabel is the label to -map for the destination's audio. OutLabel string `json:"outLabel"` + // SecondOutLabel is the label to -map for a SECOND finished mix -- the VOD + // track -- or "" when this destination has only one, which is nearly all of + // them. Set only by CompilePair. + // + // It lives on Result rather than only on Pair so that the engine's one + // description of an output does not have to change type to carry it: every + // signature between the compiler and ffmpeg.DestSpec already passes a + // Result, and "" is exactly what DestSpec.SecondAudioOutLabel means by "not + // opted in". + SecondOutLabel string `json:"secondOutLabel,omitempty"` // Summary is the human sentence shown on a destination card, // e.g. "Tracks 1, 2, 4 → stereo". Summary string `json:"summary"` @@ -62,7 +87,7 @@ type Result struct { // (or, for denoise, the source annotation) asked for it, so a profile that uses // none of them produces the string above byte for byte. func Compile(p Profile, src Source) (Result, error) { - return compile(p, src, false) + return compile(p, src, false, ns{}) } // CompileProvisional builds the same graph for a layout that has NOT been @@ -78,15 +103,15 @@ func Compile(p Profile, src Source) (Result, error) { // one, because an operator has to know the mix is being decided by FFmpeg at // runtime rather than by the matrix they drew. func CompileProvisional(p Profile, src Source) (Result, error) { - return compile(p, src, true) + return compile(p, src, true, ns{}) } -func compile(p Profile, src Source, provisional bool) (Result, error) { +func compile(p Profile, src Source, provisional bool, n ns) (Result, error) { if err := p.Validate(); err != nil { return Result{}, err } - res := Result{OutLabel: OutLabel} + res := Result{OutLabel: n.of(OutLabel)} cells, warns := resolveCells(p, src) if provisional { @@ -131,7 +156,7 @@ func compile(p Profile, src Source, provisional bool) (Result, error) { var chains []string label := make(map[int]string, len(tracks)) for _, t := range tracks { - label[t] = fmt.Sprintf("a_t%d", t) + label[t] = n.track(t) chain := trackChain(src, t, byTrack[t]) if provisional { chain = provisionalChain(src, t, trackGain(p, t)) @@ -143,7 +168,7 @@ func compile(p Profile, src Source, provisional bool) (Result, error) { // trigger down along with everything else. legs := make([]string, 0, len(tracks)) if d, ok := p.EffectiveDucking(); ok { - duckChains, duckLegs, duckWarns := duckGraph(d, src, tracks, label, provisional) + duckChains, duckLegs, duckWarns := duckGraph(d, src, tracks, label, provisional, n) chains = append(chains, duckChains...) legs = duckLegs if len(duckWarns) > 0 { @@ -161,9 +186,9 @@ func compile(p Profile, src Source, provisional bool) (Result, error) { // the resulting clip risk explicitly, below. cur := legs[0] if len(legs) > 1 { - chains = append(chains, fmt.Sprintf("%samix=inputs=%d:duration=longest:normalize=0[a_mix]", - bracket(legs), len(legs))) - cur = "a_mix" + chains = append(chains, fmt.Sprintf("%samix=inputs=%d:duration=longest:normalize=0[%s]", + bracket(legs), len(legs), n.of("a_mix"))) + cur = n.of("a_mix") } norm := resolveNorm(p.Normalize, len(tracks), peakGain(cells)) @@ -177,8 +202,8 @@ func compile(p Profile, src Source, provisional bool) (Result, error) { } res.Normalization = norm if f := normFilterFor(norm, loud, loudOK); f != "" { - chains = append(chains, fmt.Sprintf("[%s]%s[a_norm]", cur, f)) - cur = "a_norm" + chains = append(chains, fmt.Sprintf("[%s]%s[%s]", cur, f, n.of("a_norm"))) + cur = n.of("a_norm") } // Delay last but one: holding the finished mix is what "this destination is @@ -186,8 +211,8 @@ func compile(p Profile, src Source, provisional bool) (Result, error) { // loudness measurement looking at the same samples it always did. switch { case p.DelayMS > 0: - chains = append(chains, fmt.Sprintf("[%s]adelay=delays=%d:all=1[a_delay]", cur, p.DelayMS)) - cur = "a_delay" + chains = append(chains, fmt.Sprintf("[%s]adelay=delays=%d:all=1[%s]", cur, p.DelayMS, n.of("a_delay"))) + cur = n.of("a_delay") case p.DelayMS < 0: res.VideoDelayMS = -p.DelayMS } @@ -198,7 +223,7 @@ func compile(p Profile, src Source, provisional bool) (Result, error) { if rate == 0 { rate = 48000 } - chains = append(chains, fmt.Sprintf("[%s]aresample=%d:async=1:first_pts=0[%s]", cur, rate, OutLabel)) + chains = append(chains, fmt.Sprintf("[%s]aresample=%d:async=1:first_pts=0[%s]", cur, rate, n.of(OutLabel))) res.FilterComplex = strings.Join(chains, ";") res.Summary = summarize(tracks) @@ -325,7 +350,7 @@ const DenoiseFilter = "afftdn=nr=12:nf=-25:tn=1" // legs. Returning no legs means nothing was ducked and the caller should mix as // usual; that is the deliberate response to a duck that cannot be built, since // an un-ducked mix is still the operator's audio and a broken graph is silence. -func duckGraph(d Ducking, src Source, tracks []int, label map[int]string, provisional bool) (chains, legs, warns []string) { +func duckGraph(d Ducking, src Source, tracks []int, label map[int]string, provisional bool, n ns) (chains, legs, warns []string) { inMix := map[int]bool{} for _, t := range tracks { inMix[t] = true @@ -359,15 +384,15 @@ func duckGraph(d Ducking, src Source, tracks []int, label map[int]string, provis var keys []string for _, t := range triggers { if inMix[t] { - mixLbl := fmt.Sprintf("a_t%d_mix", t) - keyLbl := fmt.Sprintf("a_t%d_key", t) + mixLbl := fmt.Sprintf("%sa_t%d_mix", n.prefix, t) + keyLbl := fmt.Sprintf("%sa_t%d_key", n.prefix, t) chains = append(chains, fmt.Sprintf("[%s]asplit=2[%s][%s]", label[t], mixLbl, keyLbl)) label[t] = mixLbl keys = append(keys, keyLbl) continue } tr, _ := src.TrackByIndex(t) - keyLbl := fmt.Sprintf("a_k%d", t) + keyLbl := fmt.Sprintf("%sa_k%d", n.prefix, t) // Downmix the tap the same way a contributing track would, so the two // sidechaincompress inputs always agree on channel layout, and denoise // it if it is annotated: room noise opening the duck is precisely the @@ -387,9 +412,9 @@ func duckGraph(d Ducking, src Source, tracks []int, label map[int]string, provis key := keys[0] if len(keys) > 1 { - chains = append(chains, fmt.Sprintf("%samix=inputs=%d:duration=longest:normalize=0[a_duckkey]", - bracket(keys), len(keys))) - key = "a_duckkey" + chains = append(chains, fmt.Sprintf("%samix=inputs=%d:duration=longest:normalize=0[%s]", + bracket(keys), len(keys), n.of("a_duckkey"))) + key = n.of("a_duckkey") } bus := label[targets[0]] @@ -402,17 +427,17 @@ func duckGraph(d Ducking, src Source, tracks []int, label map[int]string, provis // amix=normalize=0 is a plain sum, so summing early is arithmetically // identical, and one compressor means one gain-reduction envelope // instead of several that could drift apart. - chains = append(chains, fmt.Sprintf("%samix=inputs=%d:duration=longest:normalize=0[a_duckin]", - bracket(in), len(in))) - bus = "a_duckin" + chains = append(chains, fmt.Sprintf("%samix=inputs=%d:duration=longest:normalize=0[%s]", + bracket(in), len(in), n.of("a_duckin"))) + bus = n.of("a_duckin") } - chains = append(chains, fmt.Sprintf("[%s][%s]sidechaincompress=%s[a_duck]", bus, key, duckParams(d))) + chains = append(chains, fmt.Sprintf("[%s][%s]sidechaincompress=%s[%s]", bus, key, duckParams(d), n.of("a_duck"))) placed := false for _, t := range tracks { if isTarget[t] { if !placed { - legs = append(legs, "a_duck") + legs = append(legs, n.of("a_duck")) placed = true } continue diff --git a/internal/routing/pair.go b/internal/routing/pair.go new file mode 100644 index 00000000..ad613c0f --- /dev/null +++ b/internal/routing/pair.go @@ -0,0 +1,140 @@ +package routing + +import ( + "fmt" + "strings" +) + +// SecondaryPrefix is the label namespace of the second mix in a paired graph. +// +// It is spelled for a reader of `ffmpeg -h`, not for brevity: an operator +// staring at a filter_complex in the UI should be able to tell which half is +// which without knowing this package exists. Every label in the second half +// carries it -- vod_a_t0, vod_a_mix, vod_aout -- and no label in the first half +// can, because the first half is compiled in the EMPTY namespace and every label +// it emits begins with "a". +const SecondaryPrefix = "vod_" + +// Pair is two finished mixes sharing ONE filter graph, for a destination that +// carries a second audio track. +// +// WHY THIS TYPE EXISTS. ffmpeg.DestSpec.SecondAudioOutLabel has been able to map +// and encode a second mix since #331, and it was measured arriving as two +// distinct tracks through this project's own RTMP ingest. Nothing could ask for +// one, because Compile emitted a single mix whose internal labels were fixed +// constants -- a_t0, a_mix, aout -- so concatenating two compiled graphs +// collided on every one of them. Namespacing those labels is the whole of the +// fix, and this type is how a caller asks for the result. +// +// TWO INPUT TAPS OF THE SAME INGEST TRACK ARE FINE, which is the fact that +// decided the shape of this. Both halves emit [0:a:N] for any track they share, +// and a filter output pad feeds exactly one input pad -- so the obvious reading +// is that a shared tap needs an explicit asplit, the way duckGraph splits a +// trigger that also has to reach the mix. It does not: an INPUT STREAM is not a +// filter pad, and FFmpeg inserts the split itself. Measured on FFmpeg 6.0.1 +// (Alpine 3.18 -- the 6.0 floor internal/ffmpeg/detect.go enforces) and on 8.1.2 +// (Homebrew), both of which built the two-tap graph and produced two mixes whose +// tone content differed. An asplit here would have been dead weight carried on a +// guess; see TestAPairedGraphReachesFFmpegAsTwoDistinctMixes, which builds the +// real graph with the real binary and reads the tones back rather than counting +// tracks. +type Pair struct { + // Result is the PRIMARY mix -- its OutLabel, Tracks, Summary, Normalization + // and VideoDelayMS all describe the first track, exactly as a plain Compile + // would -- with TWO exceptions: FilterComplex carries BOTH halves, because + // that is the single string FFmpeg is handed, and SecondOutLabel names the + // second mix. Callers that already know what to do with a Result therefore + // need to learn one new field, not a new type -- which is what lets the + // engine carry a VOD track without changing a single signature. + Result + + // Second describes the second mix on its own terms -- which tracks reached + // it, what it was normalized to, what it warns about. Nil when there is no + // second mix, which includes the case where one was ASKED for and could not + // be built; see CompilePair for why that is a warning and not an error. + Second *Result +} + +// CompilePair compiles a primary mix and an optional secondary mix into one +// filter graph. +// +// The primary is compiled in the empty namespace, so a destination that gains a +// second track does not have its first track silently rewritten: the primary +// half of the returned FilterComplex is byte for byte what Compile(primary, src) +// returns on its own. That is asserted by +// TestTheEmptyNamespaceIsByteIdenticalToTheSingleMixGraph, because "the live mix +// is unchanged" is the promise an operator is actually relying on when they tick +// a VOD box, and a promise nothing checks is a promise that decays. +// +// A NIL SECONDARY IS THE ORDINARY CASE and returns a Pair with SecondOutLabel "" +// and Second nil -- i.e. a Result, wearing a different hat. Callers do not need +// to branch before calling. +// +// A SECONDARY THAT WILL NOT COMPILE IS A WARNING, NOT AN ERROR. This is the +// owner's standing decision that an optional VOD track must never veto a working +// broadcast, applied at the earliest point it can be: if the secondary profile +// selects nothing this ingest carries, or every track it selects is excluded by +// a role policy, the operator gets their live stream plus a sentence saying why +// there is no VOD track. The alternative -- returning an error -- would take a +// destination that was publishing fine yesterday off the air because an OPTIONAL +// extra could not be built, which is precisely backwards. +// +// A PRIMARY THAT WILL NOT COMPILE IS STILL AN ERROR. There is no stream without +// it, and pretending otherwise would publish silence. +func CompilePair(primary Profile, secondary *Profile, src Source) (Pair, error) { + first, err := compile(primary, src, false, ns{}) + if err != nil { + return Pair{}, err + } + + p := Pair{Result: first} + if secondary == nil { + return p, nil + } + + second, serr := compile(*secondary, src, false, ns{prefix: SecondaryPrefix}) + if serr != nil { + // Name the destination-level consequence, not the Go error. "no audio" + // on its own reads as though the whole destination is silent, which is + // the opposite of what has happened. + p.Warnings = dedupe(append(p.Warnings, fmt.Sprintf( + "the second (VOD) audio track could not be built and is not being sent, so this destination is publishing its live mix only: %v", serr))) + return p, nil + } + + // Prefix the second mix's own warnings. Unprefixed they are indistinguishable + // from the live mix's -- "track 4 is not present on the ingest" is a + // different problem depending on which mix dropped it, and an operator + // reading a destination card cannot tell them apart otherwise. + for _, w := range second.Warnings { + p.Warnings = append(p.Warnings, "second (VOD) audio track: "+w) + } + p.Warnings = dedupe(p.Warnings) + + p.FilterComplex = first.FilterComplex + ";" + second.FilterComplex + p.SecondOutLabel = second.OutLabel + p.Second = &second + return p, nil +} + +// SecondaryLabels reports every label the secondary namespace claims in g, in +// the order they are defined. It exists for the collision test and for anyone +// debugging a graph by eye; nothing in the compile path calls it. +func SecondaryLabels(g string) []string { + var out []string + for _, f := range strings.Split(g, ";") { + for { + i := strings.Index(f, "["+SecondaryPrefix) + if i < 0 { + break + } + j := strings.Index(f[i:], "]") + if j < 0 { + break + } + out = append(out, f[i+1:i+j]) + f = f[i+j:] + } + } + return dedupe(out) +} diff --git a/internal/routing/pair_test.go b/internal/routing/pair_test.go new file mode 100644 index 00000000..e71cd641 --- /dev/null +++ b/internal/routing/pair_test.go @@ -0,0 +1,401 @@ +package routing + +import ( + "os/exec" + "regexp" + "strconv" + "strings" + "testing" + + "github.com/rainmanjam/polyemesis/internal/testenv" +) + +// pairSource is two stereo ingest tracks, which is the smallest source that can +// tell two mixes apart: a mix carrying both is distinguishable from a mix +// carrying one. +func pairSource() Source { + return Source{Tracks: []Track{ + {Index: 0, Channels: 2, Layout: "stereo", Codec: "aac"}, + {Index: 1, Channels: 2, Layout: "stereo", Codec: "aac"}, + }} +} + +func pairProfile(tracks ...int) Profile { + p := DefaultProfile() + p.Tracks = nil + for _, t := range tracks { + p.Tracks = append(p.Tracks, TrackSel{Track: t, Enabled: true, Gain: 1}) + } + return p +} + +// TestTheEmptyNamespaceIsByteIdenticalToTheSingleMixGraph is the compatibility +// half of the label namespacing: a destination that gains a VOD track must not +// have its LIVE mix rewritten, and every destination that never asks for one +// must produce the exact command it produced before namespacing existed. +// +// It compares the primary half of a paired graph against a solo Compile of the +// same profile, byte for byte, across profiles that exercise every label site +// there is -- the per-track pans, the amix, the loudness stage, the delay, the +// resample, and all six ducking labels. A namespace accidentally applied to the +// primary, or a label site missed and left as a bare constant while its +// neighbours moved, both show up here as a diff. +// +// MUTATION: `ns{}` -> `ns{prefix: "x_"}` in Compile (filtergraph.go). Observed: +// FAIL, "solo and paired primary differ" on every subcase. +// MUTATION: `n.of("a_mix")` -> `"a_mix"` in compile. Observed: FAIL on the +// ducking and multi-track subcases (a_mix defined in the vod_ namespace but +// referenced bare). Restored from /tmp backup; `git diff --stat` clean. +func TestTheEmptyNamespaceIsByteIdenticalToTheSingleMixGraph(t *testing.T) { + src := pairSource() + + ducked := pairProfile(0, 1) + ducked.Ducking = &Ducking{Target: []int{0}, Trigger: []int{1}, ThresholdDB: -24, Ratio: 8, AttackMS: 20, ReleaseMS: 300} + + loud := pairProfile(0, 1) + loud.Loudness = &Loudness{TargetLUFS: -16, TruePeakDB: -1.5, RangeLU: 11} + + delayed := pairProfile(0, 1) + delayed.DelayMS = 250 + + for _, tc := range []struct { + name string + prof Profile + }{ + {"one track", pairProfile(0)}, + {"two tracks", pairProfile(0, 1)}, + {"ducking", ducked}, + {"loudness", loud}, + {"delay", delayed}, + } { + t.Run(tc.name, func(t *testing.T) { + solo, err := Compile(tc.prof, src) + if err != nil { + t.Fatalf("solo compile: %v", err) + } + vod := pairProfile(0) + paired, err := CompilePair(tc.prof, &vod, src) + if err != nil { + t.Fatalf("paired compile: %v", err) + } + // The secondary half is appended after the primary, so the primary + // half is a literal prefix. Checking the prefix rather than cutting + // on a marker keeps the assertion on the actual claim -- "the live + // graph is unchanged" -- and does not depend on what the first + // secondary chain happens to start with (it starts with an input + // tap, [0:a:N], not with the namespace). + if !strings.HasPrefix(paired.FilterComplex, solo.FilterComplex+";") { + t.Errorf("the paired graph does not start with the solo graph\n solo: %s\npaired: %s", solo.FilterComplex, paired.FilterComplex) + } + if !strings.Contains(paired.FilterComplex, SecondaryPrefix) { + t.Fatalf("paired graph has no secondary half at all: %s", paired.FilterComplex) + } + if paired.OutLabel != solo.OutLabel { + t.Errorf("primary out label moved: solo %q, paired %q", solo.OutLabel, paired.OutLabel) + } + }) + } +} + +// labelDefs returns every label a filter graph DEFINES, i.e. every [x] that sits +// at the end of a chain rather than at its start. Those are the ones that +// collide: FFmpeg refuses a graph that defines the same label twice, and this +// whole change exists because Compile defined a_t0/a_mix/aout unconditionally. +// +// Input taps ([0:a:0]) are deliberately NOT collected. A shared input tap is +// legal and is measured to be legal -- see TestAPairedGraphReachesFFmpegAsTwo +// DistinctMixes -- so counting it as a collision would fail a working graph. +func labelDefs(graph string) []string { + var out []string + re := regexp.MustCompile(`\[([A-Za-z_][A-Za-z0-9_]*)\]`) + for _, chain := range strings.Split(graph, ";") { + // Everything after the last filter argument: the trailing [x][y] run. + idx := strings.LastIndex(chain, "]") + if idx < 0 { + continue + } + // Walk back over a contiguous run of [..] groups at the end. + tail := chain + start := len(tail) + for start > 0 && tail[start-1] == ']' { + open := strings.LastIndex(tail[:start], "[") + if open < 0 { + break + } + start = open + } + for _, m := range re.FindAllStringSubmatch(tail[start:], -1) { + out = append(out, m[1]) + } + } + return out +} + +// TestAPairedGraphDefinesNoLabelTwice is the structural statement of the bug +// this change fixes: before namespacing, concatenating two compiled graphs +// defined a_t0, a_mix and aout twice each, and FFmpeg refuses such a graph. +// +// This is deliberately NOT the only test of pairing, because on its own it is +// exactly the kind of guard this repo keeps getting burned by: it could pass +// against a graph FFmpeg still refuses for some other reason, and it cannot see +// whether the two mixes actually carry different audio. The measurement that +// closes both holes is TestAPairedGraphReachesFFmpegAsTwoDistinctMixes; this one +// exists to name the specific defect and to fail fast without an FFmpeg binary. +// +// MUTATION: `ns{prefix: SecondaryPrefix}` -> `ns{}` in CompilePair. Observed: +// FAIL, "label a_t0 is defined 2 times" (and a_mix, aout). Restored from /tmp +// backup; `git diff --stat` clean. +func TestAPairedGraphDefinesNoLabelTwice(t *testing.T) { + src := pairSource() + live := pairProfile(0, 1) + live.Ducking = &Ducking{Target: []int{0}, Trigger: []int{1}, ThresholdDB: -24, Ratio: 8, AttackMS: 20, ReleaseMS: 300} + vod := pairProfile(0, 1) + vod.Ducking = &Ducking{Target: []int{0}, Trigger: []int{1}, ThresholdDB: -24, Ratio: 8, AttackMS: 20, ReleaseMS: 300} + + paired, err := CompilePair(live, &vod, src) + if err != nil { + t.Fatalf("compile pair: %v", err) + } + + seen := map[string]int{} + for _, l := range labelDefs(paired.FilterComplex) { + seen[l]++ + } + if len(seen) == 0 { + t.Fatalf("no labels found at all -- labelDefs is not reading this graph: %s", paired.FilterComplex) + } + for l, n := range seen { + if n > 1 { + t.Errorf("label %s is defined %d times in one graph; FFmpeg refuses that", l, n) + } + } + // The ducking labels are the ones most easily missed, because they are + // generated in a different function. Assert the namespace actually reached + // them rather than trusting the count above, which a graph with no ducking + // would also satisfy. + // a_t1_mix, not a_t0_mix: the asplit lands on the TRIGGER track (1), which is + // the one that has to reach both the detector and the mix. + for _, want := range []string{SecondaryPrefix + "a_duck", SecondaryPrefix + "a_t1_mix", SecondaryPrefix + "a_mix"} { + if seen[want] == 0 { + t.Errorf("expected the secondary namespace to claim %q, but it is not defined; labels: %v", want, seen) + } + } + if seen["a_duck"] == 0 { + t.Errorf("the PRIMARY ducking label a_duck is missing -- the namespace leaked onto the primary; labels: %v", seen) + } +} + +// TestASecondaryThatCannotCompileLeavesTheLiveMixPublishing is the owner's +// standing decision made testable: an optional VOD track must never veto a +// working broadcast. +// +// The secondary here selects track 7, which this ingest does not carry, so it +// compiles to ErrNoAudio. The live mix must still come back intact, with a +// warning naming the consequence, and with no second label for the egress to +// map. +// +// MUTATION: `return p, nil` -> `return Pair{}, serr` in the serr branch of +// CompilePair. Observed: FAIL, "a failed VOD mix took the live mix down with +// it: routing profile selects no audio". Restored from /tmp backup; `git diff +// --stat` clean. +func TestASecondaryThatCannotCompileLeavesTheLiveMixPublishing(t *testing.T) { + src := pairSource() + live := pairProfile(0, 1) + vod := pairProfile(7) + + paired, err := CompilePair(live, &vod, src) + if err != nil { + t.Fatalf("a failed VOD mix took the live mix down with it: %v", err) + } + solo, err := Compile(live, src) + if err != nil { + t.Fatalf("solo compile: %v", err) + } + if paired.FilterComplex != solo.FilterComplex { + t.Errorf("the live graph changed because an optional extra failed\n want: %s\n got: %s", solo.FilterComplex, paired.FilterComplex) + } + if paired.SecondOutLabel != "" { + t.Errorf("SecondOutLabel = %q, want empty -- the egress would map a label the graph never defines", paired.SecondOutLabel) + } + if paired.Second != nil { + t.Errorf("Second = %+v, want nil", paired.Second) + } + var found bool + for _, w := range paired.Warnings { + if strings.Contains(w, "VOD") && strings.Contains(w, "live mix only") { + found = true + } + } + if !found { + t.Errorf("no warning explains why there is no VOD track; warnings: %v", paired.Warnings) + } +} + +// TestANilSecondaryIsExactlyASingleMix keeps the ordinary case honest: nearly +// every destination has no VOD mix, and asking for a pair without one must be +// indistinguishable from Compile -- same graph, no second label, and critically +// no warning, because there is nothing to warn about. +// +// MUTATION: deleted the `if secondary == nil { return p, nil }` early return so +// a nil secondary fell through to compile(*secondary, ...). Observed: panic +// (nil dereference) -> FAIL. Restored from /tmp backup; `git diff --stat` clean. +func TestANilSecondaryIsExactlyASingleMix(t *testing.T) { + src := pairSource() + live := pairProfile(0, 1) + + paired, err := CompilePair(live, nil, src) + if err != nil { + t.Fatalf("compile pair: %v", err) + } + solo, err := Compile(live, src) + if err != nil { + t.Fatalf("solo compile: %v", err) + } + if paired.FilterComplex != solo.FilterComplex { + t.Errorf("graph differs\n want: %s\n got: %s", solo.FilterComplex, paired.FilterComplex) + } + if paired.SecondOutLabel != "" || paired.Second != nil { + t.Errorf("a nil secondary produced a second mix: label %q, second %+v", paired.SecondOutLabel, paired.Second) + } + if len(paired.Warnings) != len(solo.Warnings) { + t.Errorf("warnings differ: solo %v, paired %v", solo.Warnings, paired.Warnings) + } +} + +// TestAPrimaryThatCannotCompileIsStillAnError is the other side of the +// never-veto rule, and it is here because "never fail" is the easy overcorrection. +// There is no stream without the live mix; returning a Pair anyway would publish +// silence and report success. +// +// MUTATION: made CompilePair swallow the primary error and return an empty Pair +// with nil error. Observed: FAIL, "a primary that selects nothing compiled +// without error". Restored from /tmp backup; `git diff --stat` clean. +func TestAPrimaryThatCannotCompileIsStillAnError(t *testing.T) { + src := pairSource() + vod := pairProfile(0) + if _, err := CompilePair(pairProfile(7), &vod, src); err == nil { + t.Fatal("a primary that selects nothing compiled without error") + } +} + +// TestAPairedGraphReachesFFmpegAsTwoDistinctMixes is the measurement the rest of +// this file is scaffolding for, and the only test here that can tell the +// difference between "two labels" and "two tracks of different audio". +// +// It puts a 300 Hz tone on ingest track 0 and 5 kHz on track 1, compiles a LIVE +// mix carrying both and a VOD mix carrying only track 0, hands the real graph to +// the real FFmpeg binary, and reads the tone content back off each output track +// with a bandpass and volumedetect. +// +// WHAT THIS CATCHES THAT A LABEL COUNT CANNOT, and the reason it is written this +// way round: the failure this feature is most likely to ship with is TWO TRACKS +// THAT ARE THE SAME MIX. A track count sees two tracks and passes. The +// per-track tone content is the only assertion that separates them, so the +// 5 kHz reading on the VOD track is the load-bearing line -- it must be ABSENT +// there and PRESENT on the live track. +// +// Both graphs tap [0:a:0]. That is the shared-input case, and it is the reason +// this runs a binary at all: whether FFmpeg splits an input stream implicitly +// was the open question that decided against emitting an asplit. Answered yes on +// 6.0.1 (Alpine 3.18, the floor internal/ffmpeg/detect.go enforces) and 8.1.2. +// +// MUTATION: `p.SecondOutLabel = second.OutLabel` -> `= first.OutLabel`, i.e. map +// the live mix twice, which is the exact "two tracks, one mix" defect. Observed: +// FFmpeg refused the duplicate map outright -> FAIL at the build step. Then +// mutated the VOD profile to select tracks 0 and 1 (a legal graph that is +// nonetheless the wrong mix): FAIL, "VOD track carries 5 kHz at -18.1 dB, want +// it absent -- the two tracks are the same mix, not two mixes". Restored from +// /tmp backup; `git diff --stat` clean. +func TestAPairedGraphReachesFFmpegAsTwoDistinctMixes(t *testing.T) { + // testenv.FFmpegBinary rather than a local exec.LookPath + t.Skip: its skip + // lives inside internal/testenv, so it is not a free pass this package has to + // account for, and it FAILS instead of skipping when POLYEMESIS_REQUIRE_FFMPEG + // says the environment undertook to provide a binary (#187). + ffmpegBin := testenv.FFmpegBinary(t, "ffmpeg", + "no ffmpeg on PATH; this test measures a real filter graph and has nothing to measure without one") + + dir := t.TempDir() + srcFile := dir + "/src.nut" + // Track 0 = 300 Hz, track 1 = 5 kHz. Distinct enough that a bandpass at one + // reads the other as absent by ~55 dB. + // + // PCM in NUT, and no video at all. Every codec here is built into any FFmpeg + // -- lavfi, pcm_s16le, nut -- so there is no "this build cannot mux h264/aac" + // case to skip on, which is the second free pass this test would otherwise + // have needed. It also keeps the measurement off a lossy round trip: the + // absent tone reads -73 dB rather than whatever AAC would have left behind. + build := exec.Command(ffmpegBin, "-y", "-v", "error", + "-f", "lavfi", "-i", "sine=frequency=300:duration=2:sample_rate=48000", + "-f", "lavfi", "-i", "sine=frequency=5000:duration=2:sample_rate=48000", + "-map", "0:a", "-map", "1:a", + "-c:a", "pcm_s16le", "-f", "nut", srcFile) + if out, err := build.CombinedOutput(); err != nil { + t.Fatalf("building the two-track fixture failed (%v); every codec it uses is built into FFmpeg, so this is not an environment problem: %s", err, out) + } + + src := pairSource() + live := pairProfile(0, 1) // both tones + vod := pairProfile(0) // 300 Hz only + + paired, err := CompilePair(live, &vod, src) + if err != nil { + t.Fatalf("compile pair: %v", err) + } + if paired.SecondOutLabel == "" { + t.Fatal("no second mix was compiled, so there is nothing to measure") + } + + outFile := dir + "/out.nut" + run := exec.Command(ffmpegBin, "-y", "-v", "error", "-i", srcFile, + "-filter_complex", paired.FilterComplex, + "-map", "["+paired.OutLabel+"]", + "-map", "["+paired.SecondOutLabel+"]", + "-c:a", "pcm_s16le", "-f", "nut", outFile) + if out, err := run.CombinedOutput(); err != nil { + t.Fatalf("FFmpeg refused the paired graph: %v\ngraph: %s\n%s", err, paired.FilterComplex, out) + } + + const ( + liveTrack = 0 + vodTrack = 1 + // A tone that is present reads around -18 dB; one that is absent reads + // below -70. Anything under this threshold is absent by any reading. + presentAbove = -40.0 + ) + measure := func(track int, hz int) float64 { + t.Helper() + cmd := exec.Command(ffmpegBin, "-hide_banner", "-nostats", "-i", outFile, + "-map", "0:a:"+strconv.Itoa(track), + "-af", "bandpass=f="+strconv.Itoa(hz)+":width_type=h:w=80,volumedetect", + "-f", "null", "-") + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("measuring track %d at %d Hz: %v\n%s", track, hz, err, out) + } + m := regexp.MustCompile(`max_volume:\s*(-?[0-9.]+) dB`).FindStringSubmatch(string(out)) + if m == nil { + t.Fatalf("volumedetect printed no max_volume for track %d at %d Hz; output:\n%s", track, hz, out) + } + v, err := strconv.ParseFloat(m[1], 64) + if err != nil { + t.Fatalf("unparseable max_volume %q: %v", m[1], err) + } + return v + } + + // The live mix carries BOTH tones. + if v := measure(liveTrack, 300); v < presentAbove { + t.Errorf("live track is missing its 300 Hz tone: %.1f dB", v) + } + if v := measure(liveTrack, 5000); v < presentAbove { + t.Errorf("live track is missing its 5 kHz tone: %.1f dB", v) + } + // The VOD mix carries ONLY 300 Hz. This is the assertion that separates two + // distinct mixes from the same mix sent twice. + if v := measure(vodTrack, 300); v < presentAbove { + t.Errorf("VOD track is missing its 300 Hz tone: %.1f dB", v) + } + if v := measure(vodTrack, 5000); v >= presentAbove { + t.Errorf("VOD track carries 5 kHz at %.1f dB, want it absent -- the two tracks are the same mix, not two mixes", v) + } +} diff --git a/ui/src/components/DestinationDialog.tsx b/ui/src/components/DestinationDialog.tsx index 76d372e6..b610da99 100644 --- a/ui/src/components/DestinationDialog.tsx +++ b/ui/src/components/DestinationDialog.tsx @@ -602,6 +602,12 @@ export function DestinationDialog({ open, onOpenChange, destination, onSaved }: // control is still rendered inside the Facebook box because Facebook is the // only platform that hands out a backup endpoint today. const [backupIngestWanted, setBackupIngestWanted] = useState(false); + // "Negotiate Enhanced Broadcasting with Twitch at go-live." Top-level and + // platform-neutral in the row, the same way backupIngestWanted is, but the + // control is rendered only in the Twitch box because Twitch is the one + // platform that publishes this negotiation today — see + // db.Destination.Multitrack. + const [multitrack, setMultitrack] = useState(false); const [accountId, setAccountId] = useState("none"); const [accounts, setAccounts] = useState([]); const [renditionId, setRenditionId] = useState(PASSTHROUGH); @@ -686,6 +692,7 @@ export function DestinationDialog({ open, onOpenChange, destination, onSaved }: setCompliance(destination.compliance ?? {}); setFacebook(destination.facebook ?? {}); setBackupIngestWanted(destination.backupIngestWanted ?? false); + setMultitrack(destination.multitrack ?? false); setAccountId(destination.accountId ? String(destination.accountId) : "none"); // A destination saved before renditions existed has no rendition id at // all, which is exactly passthrough — the same thing it has always done. @@ -697,6 +704,7 @@ export function DestinationDialog({ open, onOpenChange, destination, onSaved }: setCompliance({}); setFacebook({}); setBackupIngestWanted(false); + setMultitrack(false); setName(""); setPlatform("custom"); setPresetId(""); @@ -859,6 +867,11 @@ export function DestinationDialog({ open, onOpenChange, destination, onSaved }: compliance, facebook, backupIngestWanted, + // The bare boolean, for the reason spelled out at the backup toggle + // below: the PUT is decoded OVER the stored row, so a key + // JSON.stringify omits is a key the server leaves alone. `false` has + // to travel or switching this off saves nothing. + multitrack, }; // The stream key travels ONLY when this dialog is what changed it. // @@ -1839,6 +1852,45 @@ export function DestinationDialog({ open, onOpenChange, destination, onSaved }: )} + {/* Gated the same way the Facebook box below is: by the selected + platform, not by whether an account is connected. Twitch is the + one platform that publishes this negotiation, and offering it + elsewhere would be a control that cannot do anything. + + Copy is inline English rather than catalogue keys, following the + audio copy toggle and the Twitch content-labels help directly + above -- both are per-destination switches in this same dialog. */} + {platform === "twitch" && ( +
+ +
+ + + {multitrack ? "Negotiate at go-live" : "Use the ordinary Twitch ingest"} + +
+ {/* Not a warning, and deliberately not styled as one. A + negotiation that does not succeed is the EXPECTED outcome on + the machine most operators run this on, so saying it in amber + would train them to read a normal broadcast as broken. */} + + Asks Twitch at go-live for an ingest endpoint, a stream key it mints for this + broadcast, and the audio tracks it will accept — which is what a second + (VOD) audio mix needs, because the ordinary Twitch ingest carries one track. + Twitch only grants this to a client with a supported GPU, and a rented server + usually has none: where it is not granted the destination simply publishes to + the ordinary Twitch ingest and says so once. Nothing is lost by leaving it on, + and nothing is wrong when it falls back. + +
+ )} + {/* Not compliance: neither field is an obligation, and both are create-time-only, same as Audience above -- so this is its own box rather than folded into the amber one. See db.FacebookSettings. */} diff --git a/ui/src/lib/types.ts b/ui/src/lib/types.ts index abbd4cea..dea6f005 100644 --- a/ui/src/lib/types.ts +++ b/ui/src/lib/types.ts @@ -241,6 +241,31 @@ export interface Destination { enabled: boolean; audioBitrate: number; profile: RoutingProfile; + /** Opt into Twitch Enhanced Broadcasting — what Amazon's IVS calls Multitrack + * Video: a negotiation at go-live that answers with an ingest endpoint, a + * minted stream key, and the audio tracks Twitch will accept. + * + * Absent or false on nearly every destination, and that stays the common + * case rather than being a gap. Twitch refuses any client without a + * supported GPU, and polyemesis is installed on the operator's own + * server — a rented VPS has none. A negotiation that does not succeed is + * not a fault: the destination falls back to the ordinary ingest and says + * so once. Opt-in only because a network round trip at go-live should be + * something the operator asked for. */ + multitrack?: boolean; + /** The SECOND audio mix — the VOD track, separate from the live one. Same + * shape as `profile`, because it is the same kind of thing. + * + * Absent or null for every destination that has not opted in, which is + * nearly all of them, and null produces exactly the filter graph the + * destination produced before this field existed. Null rather than an empty + * profile: "no second mix" and "a second mix that happens to be the zero + * profile" are different, and the zero profile is not valid anyway. + * + * On Twitch this needs `multitrack`. The ordinary Twitch RTMP ingest takes + * one audio track; Enhanced Broadcasting is the only published path that + * takes two. Nothing enforces that pairing — the engine reports it. */ + vodProfile?: RoutingProfile | null; position: number; /** The shared video encode this destination reads. null or absent means * passthrough: no encode, straight off the ingest relay. */