config: reject unknown send_side_bwe_pacer, warn on unwired fallback (#4804)

send_side_bwe_pacer accepted any string and silently ran no-queue. leaky-bucket
is a defined PacerBehavior with a full LeakyBucket implementation, but the
selection switch in NewPCTransport only handled pass-through and no-queue, so
both leaky-bucket and outright typos fell through to the default with no
diagnostic.

leaky-bucket is deliberately left unwired here. LeakyBucket paces to a bitrate
supplied via Pacer.SetBitrate, and no production code calls SetBitrate or
SetInterval -- Base no-ops both and the only caller in the tree is the
test-only capturingPacer. Wiring it as-is would pace at a static bitrate and
ignore the bandwidth estimate, which is worse than the fallback.

Instead, make the existing behaviour visible: CongestionControlConfig.Validate
rejects unrecognised values at startup, the fallback logs a warning naming the
requested pacer, and config-sample.yaml documents send_side_bwe_pacer and
use_send_side_bwe for the first time.

Refs #4724

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

config-sample.yaml

@@ -117,6 +117,16 @@ rtc:
   #   # in the unlikely event of highly congested networks, SFU may choose to pause some tracks
   #   # in order to allow others to stream smoothly. You can disable this behavior here
   #   allow_pause: true
+  #   # use send-side bandwidth estimation instead of relying on receiver reports, default false
+  #   use_send_side_bwe: false
+  #   # how egress is paced when use_send_side_bwe is enabled. one of:
+  #   #   no-queue     - hand packets to a send worker that drains them without rate
+  #   #                  limiting, so the sender never blocks on a write (default)
+  #   #   pass-through - send each packet inline on the calling goroutine
+  #   #   leaky-bucket - accepted, but not currently wired to the bandwidth estimate; it
+  #   #                  logs a warning and falls back to no-queue. see livekit/livekit#4724
+  #   # an unrecognized value is rejected at startup.
+  #   send_side_bwe_pacer: no-queue
   # # allows automatic connection fallback to TCP and TURN/TLS (if configured) when UDP has been unstable, default true
   # allow_tcp_fallback: true
   # # signaling RTT (in milliseconds) below which ICE/TCP is attempted on a UDP failure; at or above it,

pkg/config/config.go

@@ -220,6 +220,22 @@ type CongestionControlConfig struct {
 	SendSideBWE      sendsidebwe.SendSideBWEConfig `yaml:"send_side_bwe,omitempty"`
 }
 
+// Validate reports configuration that would otherwise be ignored at runtime.
+func (c CongestionControlConfig) Validate() error {
+	switch pacer.PacerBehavior(c.SendSideBWEPacer) {
+	case "", pacer.PacerBehaviorPassThrough, pacer.PacerBehaviorNoQueue, pacer.PacerBehaviorLeakybucket:
+		return nil
+	default:
+		return fmt.Errorf(
+			"invalid send_side_bwe_pacer %q, must be one of %q, %q, %q",
+			c.SendSideBWEPacer,
+			pacer.PacerBehaviorPassThrough,
+			pacer.PacerBehaviorNoQueue,
+			pacer.PacerBehaviorLeakybucket,
+		)
+	}
+}
+
 type PlayoutDelayConfig struct {
 	Enabled bool `yaml:"enabled,omitempty"`
 	Min     int  `yaml:"min,omitempty"`
@@ -671,6 +687,10 @@ func NewConfig(confString string, strictMode bool, c *cli.Command, baseFlags []c
 		return nil, fmt.Errorf("could not validate RTC config: %v", err)
 	}
 
+	if err := conf.RTC.CongestionControl.Validate(); err != nil {
+		return nil, fmt.Errorf("could not validate congestion control config: %v", err)
+	}
+
 	conf.NormalizeTURNTTLs()
 
 	// expand env vars in filenames

pkg/config/config_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/urfave/cli/v3"
 
 	"github.com/livekit/livekit-server/pkg/config/configtest"
+	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 )
 
 func TestConfig_UnmarshalKeys(t *testing.T) {
@@ -182,6 +183,40 @@ func TestNewConfigNormalizesTURNTTL(t *testing.T) {
 	require.Equal(t, DefaultTURNTTLSeconds, conf.TURN.TTLSeconds)
 }
 
+func TestNewConfigValidatesSendSideBWEPacer(t *testing.T) {
+	cases := []struct {
+		name    string
+		pacer   string
+		wantErr bool
+	}{
+		{name: "unset keeps the default", pacer: "", wantErr: false},
+		{name: "pass-through", pacer: string(pacer.PacerBehaviorPassThrough), wantErr: false},
+		{name: "no-queue", pacer: string(pacer.PacerBehaviorNoQueue), wantErr: false},
+		{name: "leaky-bucket", pacer: string(pacer.PacerBehaviorLeakybucket), wantErr: false},
+		{name: "typo is rejected", pacer: "leakybucket", wantErr: true},
+		{name: "unknown value is rejected", pacer: "not-a-pacer", wantErr: true},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			content := "rtc:\n  congestion_control:\n    send_side_bwe_pacer: " + tc.pacer
+			conf, err := NewConfig(content, true, nil, nil)
+			if tc.wantErr {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tc.pacer)
+				return
+			}
+
+			require.NoError(t, err)
+			if tc.pacer == "" {
+				require.Equal(t, string(pacer.PacerBehaviorNoQueue), conf.RTC.CongestionControl.SendSideBWEPacer)
+				return
+			}
+			require.Equal(t, tc.pacer, conf.RTC.CongestionControl.SendSideBWEPacer)
+		})
+	}
+}
+
 func writeSecretFile(t *testing.T, content string) string {
 	t.Helper()
 	path := filepath.Join(t.TempDir(), "secret")

pkg/rtc/transport.go

@@ -582,12 +582,25 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
 				Config: params.CongestionControlConfig.SendSideBWE,
 				Logger: params.Logger,
 			})
-			switch pacer.PacerBehavior(params.CongestionControlConfig.SendSideBWEPacer) {
+			switch behavior := pacer.PacerBehavior(params.CongestionControlConfig.SendSideBWEPacer); behavior {
 			case pacer.PacerBehaviorPassThrough:
 				t.pacer = pacer.NewPassThrough(params.Logger, t.bwe)
 			case pacer.PacerBehaviorNoQueue:
 				t.pacer = pacer.NewNoQueue(params.Logger, t.bwe)
 			default:
+				// leaky-bucket lands here: it is a recognized value, but LeakyBucket paces
+				// to a bitrate set through Pacer.SetBitrate and nothing drives that from
+				// the bandwidth estimate, so it would pace at a fixed rate and ignore
+				// congestion. Unknown values are rejected by CongestionControlConfig.
+				// Validate, so they only reach here from callers building TransportParams
+				// directly.
+				if behavior != "" {
+					params.Logger.Warnw(
+						"send side BWE pacer unavailable, falling back", nil,
+						"pacerBehavior", behavior,
+						"fallback", pacer.PacerBehaviorNoQueue,
+					)
+				}
 				t.pacer = pacer.NewNoQueue(params.Logger, t.bwe)
 			}
 		} else {