From eb3de092f0980140b3abe07915775171d94949d0 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Thu, 25 Jun 2026 21:43:07 +0800 Subject: [PATCH 01/15] update pion/sctp (#4623) * update pion/sctp * webrtc & datachannel --- go.mod | 6 +++--- go.sum | 14 ++++++-------- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/go.mod b/go.mod index 89a90de11..c91e34abc 100644 --- a/go.mod +++ b/go.mod @@ -30,17 +30,17 @@ require ( github.com/moby/moby/client v0.4.1 github.com/olekukonko/tablewriter v1.1.4 github.com/ory/dockertest/v4 v4.0.0 - github.com/pion/datachannel v1.6.0 + github.com/pion/datachannel v1.6.2 github.com/pion/dtls/v3 v3.1.4 github.com/pion/ice/v4 v4.2.7 github.com/pion/interceptor v0.1.45 github.com/pion/rtcp v1.2.16 github.com/pion/rtp v1.10.2 - github.com/pion/sctp v1.9.5 + github.com/pion/sctp v1.10.2 github.com/pion/sdp/v3 v3.0.19 github.com/pion/transport/v4 v4.0.2 github.com/pion/turn/v5 v5.0.10 - github.com/pion/webrtc/v4 v4.2.11 + github.com/pion/webrtc/v4 v4.2.15 github.com/pkg/errors v0.9.1 github.com/prometheus/client_golang v1.23.2 github.com/redis/go-redis/v9 v9.21.0 diff --git a/go.sum b/go.sum index 29ca2a832..040fa11dd 100644 --- a/go.sum +++ b/go.sum @@ -229,8 +229,8 @@ github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJw github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= github.com/ory/dockertest/v4 v4.0.0 h1:i19aFsO/VXE0VrMk4ifnKW4G/KIJ93PCjLOslxXoPME= github.com/ory/dockertest/v4 v4.0.0/go.mod h1:b5Ofu8VIxWNhXFvQcLu17pRNQdoUBKtXBW74G4Ygzx8= -github.com/pion/datachannel v1.6.0 h1:XecBlj+cvsxhAMZWFfFcPyUaDZtd7IJvrXqlXD/53i0= -github.com/pion/datachannel v1.6.0/go.mod h1:ur+wzYF8mWdC+Mkis5Thosk+u/VOL287apDNEbFpsIk= +github.com/pion/datachannel v1.6.2 h1:7EXQ8TH3vTouBUdRWYbcX2edSx9Yj6k5zl5P+qyxEPc= +github.com/pion/datachannel v1.6.2/go.mod h1:pzbdAZvyGtXbcHM1hBbsFaOTf40lZizU/dNlvVOak6E= github.com/pion/dtls/v3 v3.1.4 h1:QhvtMflMfu9Kf0RcDC5BJBle4caPskByrKQR6uuYqpY= github.com/pion/dtls/v3 v3.1.4/go.mod h1:cr/qotLISUw/9C1m83ZPNZtj9WnXkYLpfCptPqbkInc= github.com/pion/ice/v4 v4.2.7 h1:zDEbC6MiEdhQpF8TxBOTws+NU6ZgGpveHrQq4Lc1kao= @@ -247,8 +247,8 @@ github.com/pion/rtcp v1.2.16 h1:fk1B1dNW4hsI78XUCljZJlC4kZOPk67mNRuQ0fcEkSo= github.com/pion/rtcp v1.2.16/go.mod h1:/as7VKfYbs5NIb4h6muQ35kQF/J0ZVNz2Z3xKoCBYOo= github.com/pion/rtp v1.10.2 h1:l+f6tTDcAH6xwepaAoW791ddhuYsJlqRATOzirO04Mo= github.com/pion/rtp v1.10.2/go.mod h1:Au8fc6cEByy8RLTwKTQTEeQqDB/SJDxwL4mZuxYA5Pk= -github.com/pion/sctp v1.9.5 h1:QoSFB/drmAsmSeSFNQNI3xx010nW4HsycCZckRVWWag= -github.com/pion/sctp v1.9.5/go.mod h1:N20Dq6LY+JvJDAh9VVh1JELngb2rQ8dPgds5yBWiPgw= +github.com/pion/sctp v1.10.2 h1:6aezYsMrHAwpjJ6kUdyCiWPqZwgToT00ponT7seJ6a4= +github.com/pion/sctp v1.10.2/go.mod h1:7KFmTwLcoYgJs/Z+99nJvsWL0qDpuyloSI0RbAqlrz0= github.com/pion/sdp/v3 v3.0.19 h1:1VMKs3gIkTQV5M3hNKfTAPrDXSNrYtOlmOD8+mSZUGQ= github.com/pion/sdp/v3 v3.0.19/go.mod h1:dE5WOSlzXrtiE/iuZqe9n+AcEbOjtAd3k5m5NtlV/qU= github.com/pion/srtp/v3 v3.0.12 h1:U7V17bckl7sI4mb3sepiojByDuBY0wNCqQE+6IlQBbc= @@ -259,12 +259,10 @@ github.com/pion/transport/v3 v3.1.1 h1:Tr684+fnnKlhPceU+ICdrw6KKkTms+5qHMgw6bIkY github.com/pion/transport/v3 v3.1.1/go.mod h1:+c2eewC5WJQHiAA46fkMMzoYZSuGzA/7E2FPrOYHctQ= github.com/pion/transport/v4 v4.0.2 h1:ifYlPqNwsy6aKQ9y8yzxXlHae5431ZrH2avkD/Rn6Tk= github.com/pion/transport/v4 v4.0.2/go.mod h1:06hFI+jCFcok2X2MekVufNZ/uzNZXivGBPfviSVcjgM= -github.com/pion/turn/v4 v4.1.4 h1:EU11yMXKIsK43FhcUnjLlrhE4nboHZq+TXBIi3QpcxQ= -github.com/pion/turn/v4 v4.1.4/go.mod h1:ES1DXVFKnOhuDkqn9hn5VJlSWmZPaRJLyBXoOeO/BmQ= github.com/pion/turn/v5 v5.0.10 h1:mOMZjudflXpte5OsCnXztpUKwNXcpXIAzMBnq9TXOSQ= github.com/pion/turn/v5 v5.0.10/go.mod h1:u3XjBqy2Z4+NhCUpDoOSsNuQDrPLvKStlCGWk6sTQ1E= -github.com/pion/webrtc/v4 v4.2.11 h1:QUX1QZKlNIn4O7U5JxLPGP0sV5RTncZkzu9SPR3jVNU= -github.com/pion/webrtc/v4 v4.2.11/go.mod h1:s/rAiyy77GyRFrZMx+Ls6aua26dIBPudH8/ZHYbIRWY= +github.com/pion/webrtc/v4 v4.2.15 h1:Ir/MauNFCfg+kgyBYPQLiGdVWFlzEcLxqtuzAkYkky0= +github.com/pion/webrtc/v4 v4.2.15/go.mod h1:CPTcyLfIzC4scOkQ4UY4pj6WvbUGhcNLIpK28cP5h6M= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= From 23090163ce42cddad6c42336c12d54f3419a26a6 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 26 Jun 2026 20:01:32 +0530 Subject: [PATCH 02/15] Configurable migration wait duration for longer waits in simulation. (#4624) Only applies if it is more than the default 3 seconds. --- pkg/rtc/participant.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 1cf5bb5b2..915a7022e 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -227,6 +227,7 @@ type ParticipantParams struct { DisableTransceiverReuseForE2EE bool EnableParticipantDataBlob bool EnableStartAtDesiredQuality bool + MigrationWaitDuration time.Duration } type ParticipantImpl struct { @@ -1572,7 +1573,7 @@ func (p *ParticipantImpl) setupMigrationTimerLocked() { // to try and succeed. If not, close the subscriber peer connection // and help the remote side to narrow down its ICE candidate pool. // - p.migrationTimer = time.AfterFunc(migrationWaitDuration, func() { + p.migrationTimer = time.AfterFunc(max(p.params.MigrationWaitDuration, migrationWaitDuration), func() { p.clearMigrationTimer() if p.IsClosed() || p.IsDisconnected() { From a81f06b9603e474fc4d6142c91cf27684edb330e Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 27 Jun 2026 12:06:30 +0530 Subject: [PATCH 03/15] Release v1.13.2. (#4625) Co-authored-by: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 33 +++++++++++++++++++++++++++++++++ version/version.go | 2 +- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 233153522..9f4f6bc55 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,39 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.13.2] - 2026-06-27 + +### Added + +- Add Prometheus metrics for join latency and peer connection state (#4574, #4616) +- Preserve original expiry when refreshing token (#4580) +- Add grants expiry to Auth context (#4581) +- Add ability to run pprof on dedicated HTTP server (#4584) +- Add API to get latest node stats (#4589) +- Enforce subscription permission to data track (#4588) +- egress v2 api (#4592) +- agent: thread attributes map from dispatch to job (#4598) +- Log subscription limit breaches (#4603) +- Acquire requested video layer directly at HIGH quality by default (#4595) +- Report participant capabilities in ParticipantInfo (#4606) +- Add option to force drain rtcService/agentService connections (#4618) +- Add support for data blob (a. k. a. async participant attributes) (#4619) + +### Changed + +- Update dependencies: pion/sctp, DTLS v3.1.4, protocol (#4587, #4601, #4623) +- rtc: add RestartSessionTimer to re-anchor participant session duration (#4566) +- Record more RTC cancellation points (#4600) +- Cap all metadata at 512 KiB; enforce on join, agent dispatch, and embedded agents (#4602) +- Tighten up publish latency stat (#4615) +- Echo offered audio payload types in single-PC subscriber answer (#4614) + +### Fixed + +- Fix skipped packets accounting (#4604) +- Do not log due to negative getting interpreted as large unsigned positive (#4605) +- Do not call nil callback (#4607) + ## [1.13.1] - 2026-06-08 ### Fixed diff --git a/version/version.go b/version/version.go index fbc9dc1e2..3f1dbda2b 100644 --- a/version/version.go +++ b/version/version.go @@ -14,4 +14,4 @@ package version -const Version = "1.13.1" +const Version = "1.13.2" From bfb75d1fc30d53761236b28d465e48abf1a23cb6 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 28 Jun 2026 09:28:18 +0200 Subject: [PATCH 04/15] feat: mock API server for testing server SDKs (#4627) * feat: mock API server for testing server SDKs * fixed lint --- .github/workflows/test-server-docker.yaml | 53 +++++ cmd/test-server/Dockerfile | 40 ++++ cmd/test-server/README.md | 84 ++++++++ cmd/test-server/handlers.go | 228 ++++++++++++++++++++ cmd/test-server/main.go | 247 ++++++++++++++++++++++ magefile.go | 5 + 6 files changed, 657 insertions(+) create mode 100644 .github/workflows/test-server-docker.yaml create mode 100644 cmd/test-server/Dockerfile create mode 100644 cmd/test-server/README.md create mode 100644 cmd/test-server/handlers.go create mode 100644 cmd/test-server/main.go diff --git a/.github/workflows/test-server-docker.yaml b/.github/workflows/test-server-docker.yaml new file mode 100644 index 000000000..a0a854eaf --- /dev/null +++ b/.github/workflows/test-server-docker.yaml @@ -0,0 +1,53 @@ +# Copyright 2026 LiveKit, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Builds and publishes the SDK conformance test server (cmd/test-server) to +# Docker Hub as livekit/test-server:latest on every push to master. The server +# SDK repos boot this image in their CI to run region-failover tests. +name: Release Test Server to Docker + +permissions: + contents: read + +on: + workflow_dispatch: + push: + branches: + - master + +jobs: + docker: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4 + + - name: Login to DockerHub + uses: docker/login-action@650006c6eb7dba73a995cc03b0b2d7f5ca915bee # v4 + with: + username: ${{ secrets.DOCKERHUB_USERNAME }} + password: ${{ secrets.DOCKERHUB_TOKEN }} + + - name: Build and push + uses: docker/build-push-action@f9f3042f7e2789586610d6e8b85c8f03e5195baf # v7 + with: + context: . + file: cmd/test-server/Dockerfile + push: true + platforms: linux/amd64,linux/arm64 + # Each build overrides the latest tag so SDK CI always boots the + # current mock server. + tags: livekit/test-server:latest diff --git a/cmd/test-server/Dockerfile b/cmd/test-server/Dockerfile new file mode 100644 index 000000000..0182aadea --- /dev/null +++ b/cmd/test-server/Dockerfile @@ -0,0 +1,40 @@ +# Copyright 2026 LiveKit, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Mock LiveKit API server for SDK conformance testing. Build from the repo +# root: docker build -f cmd/test-server/Dockerfile -t livekit/test-server . +FROM golang:1.26-alpine AS builder + +ARG TARGETARCH + +WORKDIR /workspace + +COPY go.mod go.mod +COPY go.sum go.sum +RUN go mod download + +COPY cmd/ cmd/ +COPY pkg/ pkg/ +COPY version/ version/ + +RUN CGO_ENABLED=0 GOOS=linux GOARCH=$TARGETARCH GO111MODULE=on go build -a -o livekit-test-server ./cmd/test-server + +FROM alpine + +COPY --from=builder /workspace/livekit-test-server /livekit-test-server + +# region-0 (primary, 9999) + 3 fallback regions +EXPOSE 9999 10000 10001 10002 + +ENTRYPOINT ["/livekit-test-server"] diff --git a/cmd/test-server/README.md b/cmd/test-server/README.md new file mode 100644 index 000000000..08dd75122 --- /dev/null +++ b/cmd/test-server/README.md @@ -0,0 +1,84 @@ +# LiveKit SDK test server + +A stateless, per-request programmable mock of the LiveKit server HTTP API. It +exists so the server SDKs (Go, Rust, Python, Node, Kotlin, Ruby) can exercise +client-side behavior against one shared implementation, published as a Docker +image and booted by each SDK's CI. + +## Why it looks the way it does + +- **Stateless.** All behavior is selected by per-request `X-Lk-Mock-*` headers, + so the server holds no mutable state and tests run in parallel. +- **Multi-port = multi-region.** The process binds one listener per simulated + region (`--ports`). A port's position in the list is its **region index**; + index `0` is the primary the SDK is initially pointed at. `GET + /settings/regions` advertises all of them in order. +- **One header drives every attempt.** The SDK sends the same control header on + the initial request *and* every failover retry. Each listener decides what to + do from its **own** index, so a single `X-Lk-Mock-Fail-Regions: 0` makes the + primary fail while the first fallback succeeds — no coordination needed. +- **The whole API is mocked with populated responses.** Every RoomService, + Egress, Ingress, SIP, and Connector method returns a type-correct, populated + response: scalar fields that share a name with the request are echoed (e.g. + `name`, `metadata`, `identity`, timeouts), `id`/`sid` fields get placeholder + values, and list endpoints return one element. Both protobuf and JSON Twirp + clients are supported. A client can override the response entirely with the + `X-Lk-Mock-Response` header (see below). Unregistered/future methods fall back + to an empty (all-default) message, which still decodes cleanly. + +## Running + +```bash +go run ./cmd/test-server # primary :9999, regions :10000-10002 +go run ./cmd/test-server --ports 9999,10000 # primary + one fallback + +# Docker +docker build -f cmd/test-server/Dockerfile -t livekit/test-server . +docker run -p 9999-10002:9999-10002 livekit/test-server +``` + +| Flag | Env | Default | Meaning | +|---|---|---|---| +| `--ports` | `LK_TEST_SERVER_PORTS` | `9999,10000,10001,10002` | listener ports; index = position | +| `--advertise-host` | `LK_TEST_SERVER_ADVERTISE_HOST` | `http://127.0.0.1` | base URL used in `/settings/regions` | +| `--bind` | `LK_TEST_SERVER_BIND` | `0.0.0.0` | bind address | +| `--twirp-prefix` | `LK_TEST_SERVER_TWIRP_PREFIX` | `/twirp` | Twirp path prefix | + +## Control protocol + +Request headers (sent by the SDK on API calls; the SDK must forward +client-configured custom headers onto the `/settings/regions` fetch and every +failover retry): + +| Header | Default | Effect | +|---|---|---| +| `X-Lk-Mock-Fail-Regions` | — | comma list of region indices that fail this request, e.g. `0` or `0,1`. Each listener fails only if its own index is listed. | +| `X-Lk-Mock-Fail-Mode` | `status` | how a failing region fails: `status`, `drop` (close connection → transport error), `delay`. | +| `X-Lk-Mock-Fail-Status` | `503` | HTTP status when failing with `status`/`delay`. | +| `X-Lk-Mock-Fail-Twirp-Code` | derived from status | Twirp error code string in the failure body. | +| `X-Lk-Mock-Delay-Ms` | `30000` | delay before a `delay`-mode region responds (for timeout tests). | +| `X-Lk-Mock-Regions-Status` | `200` | override the status of `GET /settings/regions`. | +| `X-Lk-Mock-Response` | — | protojson of the response message for the called method; replaces the populated default, giving full control over the returned payload. | + +Response headers: + +| Header | Meaning | +|---|---| +| `X-Lk-Mock-Region` | index of the region that served the response (blank on a failed region). Assert on this to confirm which region a failover landed on. | + +## Common recipes + +| Goal | Headers | +|---|---| +| Happy path | _(none)_ → 200 from region `0` | +| Failover succeeds on region 1 | `X-Lk-Mock-Fail-Regions: 0` | +| Exhaust to region 2 | `X-Lk-Mock-Fail-Regions: 0,1` | +| All regions down | `X-Lk-Mock-Fail-Regions: 0,1,2,3` | +| 4xx, no retry | `X-Lk-Mock-Fail-Regions: 0` + `X-Lk-Mock-Fail-Status: 400` | +| Transport-error failover | `X-Lk-Mock-Fail-Regions: 0` + `X-Lk-Mock-Fail-Mode: drop` | +| Region discovery unreachable | `X-Lk-Mock-Regions-Status: 500` | +| Custom response payload | `X-Lk-Mock-Response: {"sid":"RM_x","name":"my-room"}` | + +Note: SDK region failover normally only engages for `*.livekit.cloud` hosts. +Since tests point at `127.0.0.1`, set the SDK's failover-enable option to its +forced-on value so failover engages against localhost. diff --git a/cmd/test-server/handlers.go b/cmd/test-server/handlers.go new file mode 100644 index 000000000..f1568db95 --- /dev/null +++ b/cmd/test-server/handlers.go @@ -0,0 +1,228 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "io" + "net/http" + "strconv" + "strings" + + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/types/known/emptypb" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/utils/protojson" +) + +// apiSpec captures the request and response message types for one Twirp method, +// so the mock can decode the incoming request and build a typed response. +type apiSpec struct { + newReq func() proto.Message + newResp func() proto.Message +} + +// ptrMsg constrains a pointer type that is also a proto.Message, letting reg +// construct fresh request/response values generically. +type ptrMsg[T any] interface { + *T + proto.Message +} + +// apiHandlers maps "./" to its message types. It +// covers the full LiveKit API surface; see init below. +var apiHandlers = map[string]apiSpec{} + +func reg[ReqT, RespT any, Req ptrMsg[ReqT], Resp ptrMsg[RespT]](key string) { + apiHandlers[key] = apiSpec{ + newReq: func() proto.Message { return Req(new(ReqT)) }, + newResp: func() proto.Message { return Resp(new(RespT)) }, + } +} + +func init() { + // RoomService + reg[livekit.CreateRoomRequest, livekit.Room]("livekit.RoomService/CreateRoom") + reg[livekit.ListRoomsRequest, livekit.ListRoomsResponse]("livekit.RoomService/ListRooms") + reg[livekit.DeleteRoomRequest, livekit.DeleteRoomResponse]("livekit.RoomService/DeleteRoom") + reg[livekit.ListParticipantsRequest, livekit.ListParticipantsResponse]("livekit.RoomService/ListParticipants") + reg[livekit.RoomParticipantIdentity, livekit.ParticipantInfo]("livekit.RoomService/GetParticipant") + reg[livekit.RoomParticipantIdentity, livekit.RemoveParticipantResponse]("livekit.RoomService/RemoveParticipant") + reg[livekit.MuteRoomTrackRequest, livekit.MuteRoomTrackResponse]("livekit.RoomService/MutePublishedTrack") + reg[livekit.UpdateParticipantRequest, livekit.ParticipantInfo]("livekit.RoomService/UpdateParticipant") + reg[livekit.UpdateSubscriptionsRequest, livekit.UpdateSubscriptionsResponse]("livekit.RoomService/UpdateSubscriptions") + reg[livekit.SendDataRequest, livekit.SendDataResponse]("livekit.RoomService/SendData") + reg[livekit.UpdateRoomMetadataRequest, livekit.Room]("livekit.RoomService/UpdateRoomMetadata") + reg[livekit.ForwardParticipantRequest, livekit.ForwardParticipantResponse]("livekit.RoomService/ForwardParticipant") + reg[livekit.MoveParticipantRequest, livekit.MoveParticipantResponse]("livekit.RoomService/MoveParticipant") + reg[livekit.PerformRpcRequest, livekit.PerformRpcResponse]("livekit.RoomService/PerformRpc") + + // Egress + reg[livekit.StartEgressRequest, livekit.EgressInfo]("livekit.Egress/StartEgress") + reg[livekit.UpdateLayoutRequest, livekit.EgressInfo]("livekit.Egress/UpdateLayout") + reg[livekit.UpdateStreamRequest, livekit.EgressInfo]("livekit.Egress/UpdateStream") + reg[livekit.ListEgressRequest, livekit.ListEgressResponse]("livekit.Egress/ListEgress") + reg[livekit.StopEgressRequest, livekit.EgressInfo]("livekit.Egress/StopEgress") + reg[livekit.RoomCompositeEgressRequest, livekit.EgressInfo]("livekit.Egress/StartRoomCompositeEgress") + reg[livekit.WebEgressRequest, livekit.EgressInfo]("livekit.Egress/StartWebEgress") + reg[livekit.ParticipantEgressRequest, livekit.EgressInfo]("livekit.Egress/StartParticipantEgress") + reg[livekit.TrackCompositeEgressRequest, livekit.EgressInfo]("livekit.Egress/StartTrackCompositeEgress") + reg[livekit.TrackEgressRequest, livekit.EgressInfo]("livekit.Egress/StartTrackEgress") + + // Ingress + reg[livekit.CreateIngressRequest, livekit.IngressInfo]("livekit.Ingress/CreateIngress") + reg[livekit.UpdateIngressRequest, livekit.IngressInfo]("livekit.Ingress/UpdateIngress") + reg[livekit.ListIngressRequest, livekit.ListIngressResponse]("livekit.Ingress/ListIngress") + reg[livekit.DeleteIngressRequest, livekit.IngressInfo]("livekit.Ingress/DeleteIngress") + + // SIP + reg[livekit.ListSIPTrunkRequest, livekit.ListSIPTrunkResponse]("livekit.SIP/ListSIPTrunk") + reg[livekit.CreateSIPInboundTrunkRequest, livekit.SIPInboundTrunkInfo]("livekit.SIP/CreateSIPInboundTrunk") + reg[livekit.CreateSIPOutboundTrunkRequest, livekit.SIPOutboundTrunkInfo]("livekit.SIP/CreateSIPOutboundTrunk") + reg[livekit.UpdateSIPInboundTrunkRequest, livekit.SIPInboundTrunkInfo]("livekit.SIP/UpdateSIPInboundTrunk") + reg[livekit.UpdateSIPOutboundTrunkRequest, livekit.SIPOutboundTrunkInfo]("livekit.SIP/UpdateSIPOutboundTrunk") + reg[livekit.GetSIPInboundTrunkRequest, livekit.GetSIPInboundTrunkResponse]("livekit.SIP/GetSIPInboundTrunk") + reg[livekit.GetSIPOutboundTrunkRequest, livekit.GetSIPOutboundTrunkResponse]("livekit.SIP/GetSIPOutboundTrunk") + reg[livekit.ListSIPInboundTrunkRequest, livekit.ListSIPInboundTrunkResponse]("livekit.SIP/ListSIPInboundTrunk") + reg[livekit.ListSIPOutboundTrunkRequest, livekit.ListSIPOutboundTrunkResponse]("livekit.SIP/ListSIPOutboundTrunk") + reg[livekit.DeleteSIPTrunkRequest, livekit.SIPTrunkInfo]("livekit.SIP/DeleteSIPTrunk") + reg[livekit.CreateSIPDispatchRuleRequest, livekit.SIPDispatchRuleInfo]("livekit.SIP/CreateSIPDispatchRule") + reg[livekit.UpdateSIPDispatchRuleRequest, livekit.SIPDispatchRuleInfo]("livekit.SIP/UpdateSIPDispatchRule") + reg[livekit.ListSIPDispatchRuleRequest, livekit.ListSIPDispatchRuleResponse]("livekit.SIP/ListSIPDispatchRule") + reg[livekit.DeleteSIPDispatchRuleRequest, livekit.SIPDispatchRuleInfo]("livekit.SIP/DeleteSIPDispatchRule") + reg[livekit.CreateSIPParticipantRequest, livekit.SIPParticipantInfo]("livekit.SIP/CreateSIPParticipant") + reg[livekit.TransferSIPParticipantRequest, emptypb.Empty]("livekit.SIP/TransferSIPParticipant") + + // Connector + reg[livekit.DialWhatsAppCallRequest, livekit.DialWhatsAppCallResponse]("livekit.Connector/DialWhatsAppCall") + reg[livekit.DisconnectWhatsAppCallRequest, livekit.DisconnectWhatsAppCallResponse]("livekit.Connector/DisconnectWhatsAppCall") + reg[livekit.ConnectWhatsAppCallRequest, livekit.ConnectWhatsAppCallResponse]("livekit.Connector/ConnectWhatsAppCall") + reg[livekit.AcceptWhatsAppCallRequest, livekit.AcceptWhatsAppCallResponse]("livekit.Connector/AcceptWhatsAppCall") + reg[livekit.ConnectTwilioCallRequest, livekit.ConnectTwilioCallResponse]("livekit.Connector/ConnectTwilioCall") +} + +// writeAPIResponse serves a populated, type-correct response for a known API +// method. The response is the reflection-populated default unless the request +// carries an X-Lk-Mock-Response header (protojson), which overrides it +// entirely. Content type (protobuf vs JSON) mirrors the request. +func (h *mockHandler) writeAPIResponse(w http.ResponseWriter, r *http.Request) { + json := strings.Contains(r.Header.Get("Content-Type"), "json") + w.Header().Set(headerRegion, strconv.Itoa(h.regionIndex)) + + key := strings.TrimPrefix(r.URL.Path, h.twirpPrefix+"/") + spec, ok := apiHandlers[key] + if !ok { + // Unknown/future method: an empty body still decodes to a valid default + // message in every Twirp client. + writeEmptySuccess(w, json) + return + } + + body, _ := io.ReadAll(r.Body) + req := spec.newReq() + if json { + _ = protojson.Unmarshal(body, req) + } else { + _ = proto.Unmarshal(body, req) + } + + resp := spec.newResp() + if override := r.Header.Get(headerResponse); override != "" { + if err := protojson.Unmarshal([]byte(override), resp); err != nil { + // Malformed override: fall back to the populated default. + resp = spec.newResp() + populateMessage(resp.ProtoReflect(), req.ProtoReflect(), 1) + } + } else { + populateMessage(resp.ProtoReflect(), req.ProtoReflect(), 1) + } + + if json { + out, _ := protojson.Marshal(resp) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(out) + } else { + out, _ := proto.Marshal(resp) + w.Header().Set("Content-Type", "application/protobuf") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(out) + } +} + +func writeEmptySuccess(w http.ResponseWriter, json bool) { + if json { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("{}")) + } else { + w.Header().Set("Content-Type", "application/protobuf") + w.WriteHeader(http.StatusOK) + } +} + +// populateMessage fills a response message with plausible values: it echoes +// scalar fields that share a name with the request, assigns placeholder values +// to id/sid fields, and adds one element to repeated-message (list) fields so +// list endpoints return non-empty results. depth bounds list-element nesting. +func populateMessage(m protoreflect.Message, req protoreflect.Message, depth int) { + fields := m.Descriptor().Fields() + for i := 0; i < fields.Len(); i++ { + fd := fields.Get(i) + + // Echo a same-named scalar field from the request (e.g. name, metadata, + // identity, room, timeouts). + if req != nil && fd.Cardinality() != protoreflect.Repeated && isScalarKind(fd.Kind()) { + if rf := req.Descriptor().Fields().ByName(fd.Name()); rf != nil && + rf.Kind() == fd.Kind() && rf.Cardinality() != protoreflect.Repeated && req.Has(rf) { + m.Set(fd, req.Get(rf)) + continue + } + } + + // Give id/sid-like string fields a deterministic placeholder. + if fd.Kind() == protoreflect.StringKind && fd.Cardinality() != protoreflect.Repeated && !m.Has(fd) { + n := string(fd.Name()) + if n == "id" || n == "sid" || strings.HasSuffix(n, "_id") || strings.HasSuffix(n, "_sid") { + m.Set(fd, protoreflect.ValueOfString("MOCK_"+strings.ToUpper(n))) + continue + } + } + + // Populate list endpoints with a single element so clients see results. + if depth > 0 && fd.IsList() && fd.Kind() == protoreflect.MessageKind { + list := m.Mutable(fd).List() + elem := list.NewElement() + populateMessage(elem.Message(), nil, depth-1) + list.Append(elem) + } + } +} + +func isScalarKind(k protoreflect.Kind) bool { + switch k { + case protoreflect.BoolKind, + protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Uint32Kind, + protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Uint64Kind, + protoreflect.Sfixed32Kind, protoreflect.Fixed32Kind, + protoreflect.Sfixed64Kind, protoreflect.Fixed64Kind, + protoreflect.FloatKind, protoreflect.DoubleKind, + protoreflect.StringKind, protoreflect.BytesKind: + return true + default: + return false + } +} diff --git a/cmd/test-server/main.go b/cmd/test-server/main.go new file mode 100644 index 000000000..66b756d47 --- /dev/null +++ b/cmd/test-server/main.go @@ -0,0 +1,247 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Command test-server is a programmable mock of the LiveKit server HTTP API, +// used by the server SDKs to test client behavior. See cmd/test-server/README.md. +package main + +import ( + "errors" + "fmt" + "net/http" + "os" + "os/signal" + "strconv" + "strings" + "syscall" + "time" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/utils/protojson" +) + +// X-Lk-Mock-* request headers control the mock's behavior; see the README. +const ( + headerFailRegions = "X-Lk-Mock-Fail-Regions" + headerFailMode = "X-Lk-Mock-Fail-Mode" + headerFailStatus = "X-Lk-Mock-Fail-Status" + headerFailTwirpCode = "X-Lk-Mock-Fail-Twirp-Code" + headerDelayMs = "X-Lk-Mock-Delay-Ms" + headerRegionsStatus = "X-Lk-Mock-Regions-Status" + headerResponse = "X-Lk-Mock-Response" + // headerRegion is set on responses to the index of the region that served it. + headerRegion = "X-Lk-Mock-Region" +) + +const defaultDelayMs = 30_000 + +func main() { + portsFlag := flagValue("--ports", "LK_TEST_SERVER_PORTS", "9999,10000,10001,10002") + advertiseHost := flagValue("--advertise-host", "LK_TEST_SERVER_ADVERTISE_HOST", "http://127.0.0.1") + bindAddr := flagValue("--bind", "LK_TEST_SERVER_BIND", "0.0.0.0") + twirpPrefix := flagValue("--twirp-prefix", "LK_TEST_SERVER_TWIRP_PREFIX", "/twirp") + + ports, err := parsePorts(portsFlag) + if err != nil { + fmt.Fprintf(os.Stderr, "invalid --ports: %v\n", err) + os.Exit(1) + } + advertiseHost = strings.TrimRight(advertiseHost, "/") + + regions := &livekit.RegionSettings{} + for i, p := range ports { + regions.Regions = append(regions.Regions, &livekit.RegionInfo{ + Region: fmt.Sprintf("region-%d", i), + Url: fmt.Sprintf("%s:%d", advertiseHost, p), + Distance: int64(i), + }) + } + + errCh := make(chan error, len(ports)) + for i, p := range ports { + srv := &http.Server{ + Addr: fmt.Sprintf("%s:%d", bindAddr, p), + Handler: &mockHandler{regionIndex: i, regions: regions, twirpPrefix: twirpPrefix}, + } + go func() { errCh <- srv.ListenAndServe() }() + fmt.Printf("test-server: region-%d listening on %s:%d (advertised as %s:%d)\n", i, bindAddr, p, advertiseHost, p) + } + + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + select { + case err := <-errCh: + fmt.Fprintf(os.Stderr, "listener failed: %v\n", err) + os.Exit(1) + case <-sigCh: + fmt.Println("test-server: shutting down") + } +} + +type mockHandler struct { + regionIndex int + regions *livekit.RegionSettings + twirpPrefix string +} + +func (h *mockHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/settings/regions": + h.handleRegions(w, r) + case strings.HasPrefix(r.URL.Path, h.twirpPrefix+"/"): + h.handleTwirp(w, r) + case r.URL.Path == "/" || r.URL.Path == "/_test/health": + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + default: + http.NotFound(w, r) + } +} + +func (h *mockHandler) handleRegions(w http.ResponseWriter, r *http.Request) { + if status := parseStatus(r.Header.Get(headerRegionsStatus), 0); status != 0 && status != http.StatusOK { + w.WriteHeader(status) + return + } + body, err := protojson.Marshal(h.regions) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Cache-Control", "max-age=0") + w.Header().Set(headerRegion, strconv.Itoa(h.regionIndex)) + _, _ = w.Write(body) +} + +func (h *mockHandler) handleTwirp(w http.ResponseWriter, r *http.Request) { + if h.shouldFail(r) { + h.fail(w, r) + return + } + h.writeAPIResponse(w, r) +} + +func (h *mockHandler) shouldFail(r *http.Request) bool { + for _, part := range strings.Split(r.Header.Get(headerFailRegions), ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + if idx, err := strconv.Atoi(part); err == nil && idx == h.regionIndex { + return true + } + } + return false +} + +func (h *mockHandler) fail(w http.ResponseWriter, r *http.Request) { + switch strings.ToLower(r.Header.Get(headerFailMode)) { + case "drop": + if hj, ok := w.(http.Hijacker); ok { + if conn, _, err := hj.Hijack(); err == nil { + _ = conn.Close() + return + } + } + w.WriteHeader(http.StatusServiceUnavailable) + case "delay": + delay := defaultDelayMs + if ms, err := strconv.Atoi(r.Header.Get(headerDelayMs)); err == nil && ms >= 0 { + delay = ms + } + time.Sleep(time.Duration(delay) * time.Millisecond) + writeTwirpError(w, r, parseStatus(r.Header.Get(headerFailStatus), http.StatusServiceUnavailable)) + default: + writeTwirpError(w, r, parseStatus(r.Header.Get(headerFailStatus), http.StatusServiceUnavailable)) + } +} + +func writeTwirpError(w http.ResponseWriter, r *http.Request, status int) { + code := r.Header.Get(headerFailTwirpCode) + if code == "" { + code = twirpCodeForStatus(status) + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set(headerRegion, "") + w.WriteHeader(status) + _, _ = fmt.Fprintf(w, `{"code":%q,"msg":%q}`, code, fmt.Sprintf("mock failure (status %d)", status)) +} + +func twirpCodeForStatus(status int) string { + switch { + case status == http.StatusBadRequest: + return "invalid_argument" + case status == http.StatusUnauthorized: + return "unauthenticated" + case status == http.StatusForbidden: + return "permission_denied" + case status == http.StatusNotFound: + return "not_found" + case status == http.StatusTooManyRequests: + return "resource_exhausted" + case status >= 500: + return "unavailable" + default: + return "internal" + } +} + +func parseStatus(s string, def int) int { + if s == "" { + return def + } + if v, err := strconv.Atoi(strings.TrimSpace(s)); err == nil && v >= 100 && v <= 599 { + return v + } + return def +} + +func parsePorts(s string) ([]int, error) { + var ports []int + for _, part := range strings.Split(s, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + v, err := strconv.Atoi(part) + if err != nil { + return nil, fmt.Errorf("%q is not a port number", part) + } + ports = append(ports, v) + } + if len(ports) == 0 { + return nil, errors.New("at least one port is required") + } + return ports, nil +} + +// flagValue resolves a setting from a --flag, then an environment variable, then a default. +func flagValue(flag, env, def string) string { + prefix := flag + "=" + for i, arg := range os.Args[1:] { + if arg == flag { + if i+2 <= len(os.Args[1:]) { + return os.Args[1:][i+1] + } + } + if strings.HasPrefix(arg, prefix) { + return strings.TrimPrefix(arg, prefix) + } + } + if v := os.Getenv(env); v != "" { + return v + } + return def +} diff --git a/magefile.go b/magefile.go index 4423b01c9..d4d42fa68 100644 --- a/magefile.go +++ b/magefile.go @@ -177,6 +177,11 @@ func TestAll() error { return mageutil.Run(context.Background(), "go test ./... -count=1 -timeout=4m -v") } +// runs the SDK test server (cmd/test-server) in the foreground +func TestServer() error { + return mageutil.Run(context.Background(), "go run ./cmd/test-server") +} + // runs golangci-lint func Lint() error { if _, err := exec.LookPath("golangci-lint"); err != nil { From 930a2b6ad770df4ac7dc4e3e07b2162dff94e42c Mon Sep 17 00:00:00 2001 From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com> Date: Sun, 28 Jun 2026 14:57:35 -0700 Subject: [PATCH 05/15] Update module github.com/urfave/cli/v3 to v3.10.0 (#4612) Generated by renovateBot Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index c91e34abc..57c4003fb 100644 --- a/go.mod +++ b/go.mod @@ -140,7 +140,7 @@ require ( github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.69.0 // indirect github.com/prometheus/procfs v0.20.1 // indirect - github.com/urfave/cli/v3 v3.9.0 + github.com/urfave/cli/v3 v3.10.0 github.com/wlynxg/anet v0.0.5 // indirect github.com/zeebo/xxh3 v1.1.0 // indirect go.uber.org/zap/exp v0.3.0 // indirect diff --git a/go.sum b/go.sum index 040fa11dd..d8ab175ef 100644 --- a/go.sum +++ b/go.sum @@ -303,8 +303,8 @@ github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJX github.com/twitchtv/twirp v8.1.3+incompatible/go.mod h1:RRJoFSAmTEh2weEqWtpPE3vFK5YBhA6bqp2l1kfCC5A= github.com/ua-parser/uap-go v0.0.0-20260529044130-17c35e68e58c h1:XbG4n3OWA1PcRTpbBA22E2ChPLvJCuwYRXO12tIyVL0= github.com/ua-parser/uap-go v0.0.0-20260529044130-17c35e68e58c/go.mod h1:gwANdYmo9R8LLwGnyDFWK2PMsaXXX2HhAvCnb/UhZsM= -github.com/urfave/cli/v3 v3.9.0 h1:AV9lIiPv3ukYnxunaCUsHnEozptYmDN2F0+yWqLMn/c= -github.com/urfave/cli/v3 v3.9.0/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso= +github.com/urfave/cli/v3 v3.10.0 h1:0aU8yOObVDMkM13Cj4G+zb4P0PdeJMec65f81Ak1ioM= +github.com/urfave/cli/v3 v3.10.0/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso= github.com/urfave/negroni/v3 v3.1.1 h1:6MS4nG9Jk/UuCACaUlNXCbiKa0ywF9LXz5dGu09v8hw= github.com/urfave/negroni/v3 v3.1.1/go.mod h1:jWvnX03kcSjDBl/ShB0iHvx5uOs7mAzZXW+JvJ5XYAs= github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= From 2aec61c11b545ea5d9575224ee7e9f1dbe95a243 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Mon, 29 Jun 2026 10:33:51 +0800 Subject: [PATCH 06/15] update webrtc to fix interop issue with bundled datachannel (#4631) --- go.mod | 4 ++-- go.sum | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index 57c4003fb..84378d367 100644 --- a/go.mod +++ b/go.mod @@ -36,11 +36,11 @@ require ( github.com/pion/interceptor v0.1.45 github.com/pion/rtcp v1.2.16 github.com/pion/rtp v1.10.2 - github.com/pion/sctp v1.10.2 + github.com/pion/sctp v1.10.3 github.com/pion/sdp/v3 v3.0.19 github.com/pion/transport/v4 v4.0.2 github.com/pion/turn/v5 v5.0.10 - github.com/pion/webrtc/v4 v4.2.15 + github.com/pion/webrtc/v4 v4.2.16 github.com/pkg/errors v0.9.1 github.com/prometheus/client_golang v1.23.2 github.com/redis/go-redis/v9 v9.21.0 diff --git a/go.sum b/go.sum index d8ab175ef..3a0b13d74 100644 --- a/go.sum +++ b/go.sum @@ -247,8 +247,8 @@ github.com/pion/rtcp v1.2.16 h1:fk1B1dNW4hsI78XUCljZJlC4kZOPk67mNRuQ0fcEkSo= github.com/pion/rtcp v1.2.16/go.mod h1:/as7VKfYbs5NIb4h6muQ35kQF/J0ZVNz2Z3xKoCBYOo= github.com/pion/rtp v1.10.2 h1:l+f6tTDcAH6xwepaAoW791ddhuYsJlqRATOzirO04Mo= github.com/pion/rtp v1.10.2/go.mod h1:Au8fc6cEByy8RLTwKTQTEeQqDB/SJDxwL4mZuxYA5Pk= -github.com/pion/sctp v1.10.2 h1:6aezYsMrHAwpjJ6kUdyCiWPqZwgToT00ponT7seJ6a4= -github.com/pion/sctp v1.10.2/go.mod h1:7KFmTwLcoYgJs/Z+99nJvsWL0qDpuyloSI0RbAqlrz0= +github.com/pion/sctp v1.10.3 h1:1gBtLMA9lmwNuJkZSZJCdD5/Hz4yJs+7dAqi6ZY97QI= +github.com/pion/sctp v1.10.3/go.mod h1:7KFmTwLcoYgJs/Z+99nJvsWL0qDpuyloSI0RbAqlrz0= github.com/pion/sdp/v3 v3.0.19 h1:1VMKs3gIkTQV5M3hNKfTAPrDXSNrYtOlmOD8+mSZUGQ= github.com/pion/sdp/v3 v3.0.19/go.mod h1:dE5WOSlzXrtiE/iuZqe9n+AcEbOjtAd3k5m5NtlV/qU= github.com/pion/srtp/v3 v3.0.12 h1:U7V17bckl7sI4mb3sepiojByDuBY0wNCqQE+6IlQBbc= @@ -261,8 +261,8 @@ github.com/pion/transport/v4 v4.0.2 h1:ifYlPqNwsy6aKQ9y8yzxXlHae5431ZrH2avkD/Rn6 github.com/pion/transport/v4 v4.0.2/go.mod h1:06hFI+jCFcok2X2MekVufNZ/uzNZXivGBPfviSVcjgM= github.com/pion/turn/v5 v5.0.10 h1:mOMZjudflXpte5OsCnXztpUKwNXcpXIAzMBnq9TXOSQ= github.com/pion/turn/v5 v5.0.10/go.mod h1:u3XjBqy2Z4+NhCUpDoOSsNuQDrPLvKStlCGWk6sTQ1E= -github.com/pion/webrtc/v4 v4.2.15 h1:Ir/MauNFCfg+kgyBYPQLiGdVWFlzEcLxqtuzAkYkky0= -github.com/pion/webrtc/v4 v4.2.15/go.mod h1:CPTcyLfIzC4scOkQ4UY4pj6WvbUGhcNLIpK28cP5h6M= +github.com/pion/webrtc/v4 v4.2.16 h1:oK1GAg0TWJtZWYB8J/BgTgGWPoV2148gQWocH12vr3Q= +github.com/pion/webrtc/v4 v4.2.16/go.mod h1:y4HjLAkX90LH+C/qPqGOUgz8RA8CbDj3Iar3d+2hdKQ= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= From ad76898f0351b5d2bfe336079ed6a5909034974b Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 29 Jun 2026 08:19:05 +0200 Subject: [PATCH 07/15] support auth checks with mock server (#4629) * support auth checks with mock server * simplify unit tests --- cmd/test-server/README.md | 37 ++++++- cmd/test-server/auth.go | 212 ++++++++++++++++++++++++++++++++++++ cmd/test-server/handlers.go | 50 ++++++--- cmd/test-server/main.go | 21 ++-- 4 files changed, 299 insertions(+), 21 deletions(-) create mode 100644 cmd/test-server/auth.go diff --git a/cmd/test-server/README.md b/cmd/test-server/README.md index 08dd75122..1c4eb247a 100644 --- a/cmd/test-server/README.md +++ b/cmd/test-server/README.md @@ -59,6 +59,7 @@ failover retry): | `X-Lk-Mock-Delay-Ms` | `30000` | delay before a `delay`-mode region responds (for timeout tests). | | `X-Lk-Mock-Regions-Status` | `200` | override the status of `GET /settings/regions`. | | `X-Lk-Mock-Response` | — | protojson of the response message for the called method; replaces the populated default, giving full control over the returned payload. | +| `X-Lk-Mock-Skip-Auth` | — | `true` disables permission enforcement for the request (use for tests that aren't about authz, e.g. failover tests with a placeholder token). | Response headers: @@ -66,11 +67,45 @@ Response headers: |---|---| | `X-Lk-Mock-Region` | index of the region that served the response (blank on a failed region). Assert on this to confirm which region a failover landed on. | +## Permission enforcement + +Every API method requires the same token grants the real LiveKit server checks +(see `pkg/service/auth.go`), so the mock doubles as a conformance check that an +SDK attaches the right permissions automatically. Tokens are parsed and verified +with the protocol's own `auth` helpers — the same code path the real server uses +— against the mock's configured API secret (default `secret`, matching +`livekit-server --dev`; override with `--api-secret` / `LK_TEST_SERVER_API_SECRET`). + +- Missing, malformed, or wrongly-signed `Authorization` → `401 unauthenticated`. +- Validly-signed token without the required grant → `403 permission_denied`. +- `roomAdmin`-scoped methods also require the token's `room` to match the + request's room; `ForwardParticipant`/`MoveParticipant` additionally require + `destinationRoom` to match. + +SDKs exercising permissions should sign tokens with the same API secret the mock +is configured with (`secret` by default). + +| Grant | Methods | +|---|---| +| `video.roomCreate` | `CreateRoom`, `DeleteRoom`, all `Connector` calls | +| `video.roomList` | `ListRooms` | +| `video.roomRecord` | all `Egress` methods | +| `video.ingressAdmin` | all `Ingress` methods | +| `video.roomAdmin` (+ `room`) | room participant/data/metadata methods, `AgentDispatchService` methods | +| `video.roomAdmin` (+ `room` + `destinationRoom`) | `ForwardParticipant`, `MoveParticipant` | +| `sip.admin` | SIP trunk & dispatch-rule CRUD | +| `sip.call` | `CreateSIPParticipant`; `TransferSIPParticipant` (also needs `roomAdmin`) | + +Send `X-Lk-Mock-Skip-Auth: true` to bypass enforcement for tests that aren't +about permissions. + ## Common recipes | Goal | Headers | |---|---| -| Happy path | _(none)_ → 200 from region `0` | +| Happy path | valid token with the method's grant → 200 from region `0` | +| Bypass auth (failover tests) | `X-Lk-Mock-Skip-Auth: true` | +| Missing-permission error | token without the required grant → 403 | | Failover succeeds on region 1 | `X-Lk-Mock-Fail-Regions: 0` | | Exhaust to region 2 | `X-Lk-Mock-Fail-Regions: 0,1` | | All regions down | `X-Lk-Mock-Fail-Regions: 0,1,2,3` | diff --git a/cmd/test-server/auth.go b/cmd/test-server/auth.go new file mode 100644 index 000000000..d526e6777 --- /dev/null +++ b/cmd/test-server/auth.go @@ -0,0 +1,212 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "net/http" + "strings" + + "github.com/livekit/protocol/auth" + + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" +) + +// The mock enforces the same token permissions the real LiveKit server requires +// for each API method (see pkg/service/auth.go in livekit/livekit). The point is +// to verify that SDKs attach the correct grants automatically. Tokens are parsed +// and verified with the protocol's own auth helpers (the same code the server +// uses), against the mock's configured API secret (default "secret", matching +// `livekit-server --dev`); set --api-secret / LK_TEST_SERVER_API_SECRET to change it. +// +// Set X-Lk-Mock-Skip-Auth: true to bypass enforcement for tests that aren't +// about permissions (e.g. region-failover tests using a placeholder token). + +// perm describes the grants a method requires. roomAdmin additionally requires +// the token's room to match the request's room; destRoom further requires the +// token's destinationRoom to match the request's destination_room. +type perm struct { + roomCreate bool + roomList bool + roomRecord bool + ingressAdmin bool + roomAdmin bool + destRoom bool + sipAdmin bool + sipCall bool +} + +// methodPerms maps "./" to its required grants, +// matching the Ensure*Permission checks in the real server's services. +var methodPerms = map[string]perm{ + // RoomService + "livekit.RoomService/CreateRoom": {roomCreate: true}, + "livekit.RoomService/DeleteRoom": {roomCreate: true}, + "livekit.RoomService/ListRooms": {roomList: true}, + "livekit.RoomService/ListParticipants": {roomAdmin: true}, + "livekit.RoomService/GetParticipant": {roomAdmin: true}, + "livekit.RoomService/RemoveParticipant": {roomAdmin: true}, + "livekit.RoomService/MutePublishedTrack": {roomAdmin: true}, + "livekit.RoomService/UpdateParticipant": {roomAdmin: true}, + "livekit.RoomService/UpdateSubscriptions": {roomAdmin: true}, + "livekit.RoomService/SendData": {roomAdmin: true}, + "livekit.RoomService/UpdateRoomMetadata": {roomAdmin: true}, + "livekit.RoomService/ForwardParticipant": {destRoom: true}, + "livekit.RoomService/MoveParticipant": {destRoom: true}, + "livekit.RoomService/PerformRpc": {roomAdmin: true}, + + // Egress — all require record permission + "livekit.Egress/StartEgress": {roomRecord: true}, + "livekit.Egress/StartRoomCompositeEgress": {roomRecord: true}, + "livekit.Egress/StartWebEgress": {roomRecord: true}, + "livekit.Egress/StartParticipantEgress": {roomRecord: true}, + "livekit.Egress/StartTrackCompositeEgress": {roomRecord: true}, + "livekit.Egress/StartTrackEgress": {roomRecord: true}, + "livekit.Egress/UpdateLayout": {roomRecord: true}, + "livekit.Egress/UpdateStream": {roomRecord: true}, + "livekit.Egress/ListEgress": {roomRecord: true}, + "livekit.Egress/StopEgress": {roomRecord: true}, + + // Ingress — all require ingress admin + "livekit.Ingress/CreateIngress": {ingressAdmin: true}, + "livekit.Ingress/UpdateIngress": {ingressAdmin: true}, + "livekit.Ingress/ListIngress": {ingressAdmin: true}, + "livekit.Ingress/DeleteIngress": {ingressAdmin: true}, + + // SIP — trunk/dispatch administration requires sip.admin + "livekit.SIP/CreateSIPInboundTrunk": {sipAdmin: true}, + "livekit.SIP/CreateSIPOutboundTrunk": {sipAdmin: true}, + "livekit.SIP/UpdateSIPInboundTrunk": {sipAdmin: true}, + "livekit.SIP/UpdateSIPOutboundTrunk": {sipAdmin: true}, + "livekit.SIP/GetSIPInboundTrunk": {sipAdmin: true}, + "livekit.SIP/GetSIPOutboundTrunk": {sipAdmin: true}, + "livekit.SIP/ListSIPTrunk": {sipAdmin: true}, + "livekit.SIP/ListSIPInboundTrunk": {sipAdmin: true}, + "livekit.SIP/ListSIPOutboundTrunk": {sipAdmin: true}, + "livekit.SIP/DeleteSIPTrunk": {sipAdmin: true}, + "livekit.SIP/CreateSIPDispatchRule": {sipAdmin: true}, + "livekit.SIP/UpdateSIPDispatchRule": {sipAdmin: true}, + "livekit.SIP/ListSIPDispatchRule": {sipAdmin: true}, + "livekit.SIP/DeleteSIPDispatchRule": {sipAdmin: true}, + // Placing a call requires sip.call; transfer also requires room admin. + "livekit.SIP/CreateSIPParticipant": {sipCall: true}, + "livekit.SIP/TransferSIPParticipant": {sipCall: true, roomAdmin: true}, + + // AgentDispatch — room admin scoped to the dispatch's room + "livekit.AgentDispatchService/CreateDispatch": {roomAdmin: true}, + "livekit.AgentDispatchService/DeleteDispatch": {roomAdmin: true}, + "livekit.AgentDispatchService/ListDispatch": {roomAdmin: true}, + + // Connector (cloud) — initiating a call requires room create + "livekit.Connector/DialWhatsAppCall": {roomCreate: true}, + "livekit.Connector/DisconnectWhatsAppCall": {roomCreate: true}, + "livekit.Connector/ConnectWhatsAppCall": {roomCreate: true}, + "livekit.Connector/AcceptWhatsAppCall": {roomCreate: true}, + "livekit.Connector/ConnectTwilioCall": {roomCreate: true}, +} + +// authorize enforces the permissions a method requires. It returns the HTTP +// status and Twirp error code to send (0, "" means authorized / not enforced). +func (h *mockHandler) authorize(key string, r *http.Request, req proto.Message) (int, string) { + if strings.EqualFold(r.Header.Get(headerSkipAuth), "true") { + return 0, "" + } + p, known := methodPerms[key] + if !known { + return 0, "" // unknown/future method: don't enforce + } + + grants, err := h.verifyToken(r.Header.Get("Authorization")) + if err != nil { + // Missing, malformed, or improperly-signed token, like the real server. + return http.StatusUnauthorized, "unauthenticated" + } + if !p.satisfiedBy(grants, req) { + return http.StatusForbidden, "permission_denied" + } + return 0, "" +} + +// verifyToken parses and verifies a "Bearer " header using the protocol's +// own auth helpers (the same path the real server uses), returning the grants. +func (h *mockHandler) verifyToken(authorization string) (*auth.ClaimGrants, error) { + token := strings.TrimSpace(strings.TrimPrefix(authorization, "Bearer ")) + v, err := auth.ParseAPIToken(token) + if err != nil { + return nil, err + } + _, grants, err := v.Verify(h.apiSecret) + if err != nil { + return nil, err + } + return grants, nil +} + +func (p perm) satisfiedBy(g *auth.ClaimGrants, req proto.Message) bool { + v, s := g.Video, g.SIP + if p.roomCreate && (v == nil || !v.RoomCreate) { + return false + } + if p.roomList && (v == nil || !v.RoomList) { + return false + } + if p.roomRecord && (v == nil || !v.RoomRecord) { + return false + } + if p.ingressAdmin && (v == nil || !v.IngressAdmin) { + return false + } + if p.sipAdmin && (s == nil || !s.Admin) { + return false + } + if p.sipCall && (s == nil || !s.Call) { + return false + } + if p.roomAdmin || p.destRoom { + if v == nil || !v.RoomAdmin { + return false + } + if room := requestRoom(req); room != "" && v.Room != room { + return false + } + } + if p.destRoom { + if dest := requestString(req, "destination_room"); dest != "" && v.DestinationRoom != dest { + return false + } + } + return true +} + +// requestRoom reads the room name a request targets, trying the common "room" +// and "room_name" fields. +func requestRoom(req proto.Message) string { + if v := requestString(req, "room"); v != "" { + return v + } + return requestString(req, "room_name") +} + +func requestString(req proto.Message, field string) string { + if req == nil { + return "" + } + m := req.ProtoReflect() + fd := m.Descriptor().Fields().ByName(protoreflect.Name(field)) + if fd == nil || fd.Kind() != protoreflect.StringKind || fd.IsList() { + return "" + } + return m.Get(fd).String() +} diff --git a/cmd/test-server/handlers.go b/cmd/test-server/handlers.go index f1568db95..7de083bc4 100644 --- a/cmd/test-server/handlers.go +++ b/cmd/test-server/handlers.go @@ -114,31 +114,55 @@ func init() { reg[livekit.ConnectTwilioCallRequest, livekit.ConnectTwilioCallResponse]("livekit.Connector/ConnectTwilioCall") } +// serveAPI handles a Twirp call end to end: decode the request, enforce the +// method's required permissions (like the real server's auth middleware), apply +// any region-failure injection, then serve a populated response. +func (h *mockHandler) serveAPI(w http.ResponseWriter, r *http.Request) { + json := strings.Contains(r.Header.Get("Content-Type"), "json") + key := strings.TrimPrefix(r.URL.Path, h.twirpPrefix+"/") + spec, known := apiHandlers[key] + + // Decode the request up front — needed both to enforce room-scoped grants + // and to build the echoed response. + var req proto.Message + if known { + body, _ := io.ReadAll(r.Body) + req = spec.newReq() + if json { + _ = protojson.Unmarshal(body, req) + } else { + _ = proto.Unmarshal(body, req) + } + } + + // Permission enforcement comes first, mirroring the real server. + if status, code := h.authorize(key, r, req); status != 0 { + writeTwirpErrorCode(w, status, code, "mock: "+code) + return + } + + if h.shouldFail(r) { + h.fail(w, r) + return + } + + h.writeAPIResponse(w, r, json, known, req, spec) +} + // writeAPIResponse serves a populated, type-correct response for a known API // method. The response is the reflection-populated default unless the request // carries an X-Lk-Mock-Response header (protojson), which overrides it // entirely. Content type (protobuf vs JSON) mirrors the request. -func (h *mockHandler) writeAPIResponse(w http.ResponseWriter, r *http.Request) { - json := strings.Contains(r.Header.Get("Content-Type"), "json") +func (h *mockHandler) writeAPIResponse(w http.ResponseWriter, r *http.Request, json, known bool, req proto.Message, spec apiSpec) { w.Header().Set(headerRegion, strconv.Itoa(h.regionIndex)) - key := strings.TrimPrefix(r.URL.Path, h.twirpPrefix+"/") - spec, ok := apiHandlers[key] - if !ok { + if !known { // Unknown/future method: an empty body still decodes to a valid default // message in every Twirp client. writeEmptySuccess(w, json) return } - body, _ := io.ReadAll(r.Body) - req := spec.newReq() - if json { - _ = protojson.Unmarshal(body, req) - } else { - _ = proto.Unmarshal(body, req) - } - resp := spec.newResp() if override := r.Header.Get(headerResponse); override != "" { if err := protojson.Unmarshal([]byte(override), resp); err != nil { diff --git a/cmd/test-server/main.go b/cmd/test-server/main.go index 66b756d47..07449539d 100644 --- a/cmd/test-server/main.go +++ b/cmd/test-server/main.go @@ -40,6 +40,8 @@ const ( headerDelayMs = "X-Lk-Mock-Delay-Ms" headerRegionsStatus = "X-Lk-Mock-Regions-Status" headerResponse = "X-Lk-Mock-Response" + // headerSkipAuth disables permission enforcement for a request. + headerSkipAuth = "X-Lk-Mock-Skip-Auth" // headerRegion is set on responses to the index of the region that served it. headerRegion = "X-Lk-Mock-Region" ) @@ -51,6 +53,9 @@ func main() { advertiseHost := flagValue("--advertise-host", "LK_TEST_SERVER_ADVERTISE_HOST", "http://127.0.0.1") bindAddr := flagValue("--bind", "LK_TEST_SERVER_BIND", "0.0.0.0") twirpPrefix := flagValue("--twirp-prefix", "LK_TEST_SERVER_TWIRP_PREFIX", "/twirp") + // API secret used to verify request tokens for permission enforcement. + // Defaults to the `livekit-server --dev` secret. + apiSecret := flagValue("--api-secret", "LK_TEST_SERVER_API_SECRET", "secret") ports, err := parsePorts(portsFlag) if err != nil { @@ -72,7 +77,7 @@ func main() { for i, p := range ports { srv := &http.Server{ Addr: fmt.Sprintf("%s:%d", bindAddr, p), - Handler: &mockHandler{regionIndex: i, regions: regions, twirpPrefix: twirpPrefix}, + Handler: &mockHandler{regionIndex: i, regions: regions, twirpPrefix: twirpPrefix, apiSecret: apiSecret}, } go func() { errCh <- srv.ListenAndServe() }() fmt.Printf("test-server: region-%d listening on %s:%d (advertised as %s:%d)\n", i, bindAddr, p, advertiseHost, p) @@ -93,6 +98,7 @@ type mockHandler struct { regionIndex int regions *livekit.RegionSettings twirpPrefix string + apiSecret string } func (h *mockHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -126,11 +132,7 @@ func (h *mockHandler) handleRegions(w http.ResponseWriter, r *http.Request) { } func (h *mockHandler) handleTwirp(w http.ResponseWriter, r *http.Request) { - if h.shouldFail(r) { - h.fail(w, r) - return - } - h.writeAPIResponse(w, r) + h.serveAPI(w, r) } func (h *mockHandler) shouldFail(r *http.Request) bool { @@ -173,10 +175,15 @@ func writeTwirpError(w http.ResponseWriter, r *http.Request, status int) { if code == "" { code = twirpCodeForStatus(status) } + writeTwirpErrorCode(w, status, code, fmt.Sprintf("mock failure (status %d)", status)) +} + +// writeTwirpErrorCode writes a Twirp JSON error with an explicit code and message. +func writeTwirpErrorCode(w http.ResponseWriter, status int, code, msg string) { w.Header().Set("Content-Type", "application/json") w.Header().Set(headerRegion, "") w.WriteHeader(status) - _, _ = fmt.Fprintf(w, `{"code":%q,"msg":%q}`, code, fmt.Sprintf("mock failure (status %d)", status)) + _, _ = fmt.Fprintf(w, `{"code":%q,"msg":%q}`, code, msg) } func twirpCodeForStatus(status int) string { From a47e21b6cb945aabee88c98650366cef8cbf7a99 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 29 Jun 2026 10:44:08 -0700 Subject: [PATCH 08/15] Data track schema metadata (#4622) * Async attributes on participant. How it is different from existing participant attributes? 1. Async attribute can be added one at a time. 2. These are not included in `ParticipantInfo`. 3. Get an attribute bt participant identity and async attribute ID as and when needed. * clean up * get full definitions, not just ids * listener OnDataTrackSchema * name length config * data blob * deps * static check * Add missing request ID * Update protocol commit * Wire up StoreDataBlobResponse * Pass request ID through in GetDataBlobResponse * Pin protocol for schema metadata * Pass through schema and frame encoding * Support custom encoding identifiers * Rename config key * Increase default length to 32 * Make log messages more generic * Use getters with built-in null check * Do not bump deps * Rename function * Use protocol v1.48.1 release --------- Co-authored-by: boks1971 --- go.mod | 2 +- go.sum | 4 +-- pkg/config/config.go | 37 +++++++++++++++++++----- pkg/rtc/participant_data_blob_handler.go | 10 +++++++ pkg/rtc/participant_data_track.go | 15 ++++++++++ 5 files changed, 58 insertions(+), 10 deletions(-) diff --git a/go.mod b/go.mod index 84378d367..06e261c39 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 - github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63 + github.com/livekit/protocol v1.48.1 github.com/livekit/psrpc v0.7.2 github.com/mackerelio/go-osstat v0.2.7 github.com/magefile/mage v1.17.2 diff --git a/go.sum b/go.sum index 3a0b13d74..4e0188b7a 100644 --- a/go.sum +++ b/go.sum @@ -160,8 +160,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 h1:XHNNzebIKZRkLimla/hFGrAIX5EMWHctrgt3hLw7s+I= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63 h1:Rj9/54oztXeioAwUkukXBwcPE/GxL97WylFd1m7V2pQ= -github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= +github.com/livekit/protocol v1.48.1 h1:vFnMqGeknJo6jDwW4wU+x5KuxgA95fPuz7MriAnLzCc= +github.com/livekit/protocol v1.48.1/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= github.com/livekit/psrpc v0.7.2 h1:6oZ+NODJ2pLyaT6VqDq1F4Qc/3TpDUSpyphj/P9MhQc= github.com/livekit/psrpc v0.7.2/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/mackerelio/go-osstat v0.2.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94= diff --git a/pkg/config/config.go b/pkg/config/config.go index f85f6a841..90135c781 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -298,6 +298,8 @@ type LimitConfig struct { MaxDataBlobKeyLength int `yaml:"max_data_blob_key_length,omitempty"` MaxDataBlobSize uint32 `yaml:"max_data_blobs_size,omitempty"` + + MaxDataTrackCustomEncodingLength int `yaml:"max_data_track_custom_encoding_length,omitempty"` } func (l LimitConfig) CheckRoomNameLength(name string) bool { @@ -332,6 +334,26 @@ func (l LimitConfig) CheckDataBlobKeyLength(key string) bool { return l.MaxDataBlobKeyLength == 0 || len(key) <= l.MaxDataBlobKeyLength } +func (l LimitConfig) CheckDataTrackCustomEncodingLength(identifier string) bool { + return l.MaxDataTrackCustomEncodingLength == 0 || len(identifier) <= l.MaxDataTrackCustomEncodingLength +} + +func (l LimitConfig) CheckDataTrackFrameEncoding(encoding *livekit.DataTrackFrameEncoding) bool { + custom, ok := encoding.GetValue().(*livekit.DataTrackFrameEncoding_Custom) + if !ok { + return true + } + return len(custom.Custom) != 0 && l.CheckDataTrackCustomEncodingLength(custom.Custom) +} + +func (l LimitConfig) CheckDataTrackSchemaID(schema *livekit.DataTrackSchemaId) bool { + custom, ok := schema.GetEncoding().GetValue().(*livekit.DataTrackSchemaEncoding_Custom) + if !ok { + return true + } + return len(custom.Custom) != 0 && l.CheckDataTrackCustomEncodingLength(custom.Custom) +} + func (l LimitConfig) CheckDataBlobsSize(dataBlobs []*livekit.DataBlob) bool { if l.MaxDataBlobSize == 0 { return true @@ -475,13 +497,14 @@ var DefaultConfig = Config{ UpdateBatchTargetSize: 128 * 1024, }, Limit: LimitConfig{ - MaxMetadataSize: 512 * 1024, - MaxAttributesSize: 64 * 1024, - MaxRoomNameLength: 256, - MaxParticipantIdentityLength: 256, - MaxParticipantNameLength: 256, - MaxDataBlobKeyLength: 256, - MaxDataBlobSize: 64000, + MaxMetadataSize: 512 * 1024, + MaxAttributesSize: 64 * 1024, + MaxRoomNameLength: 256, + MaxParticipantIdentityLength: 256, + MaxParticipantNameLength: 256, + MaxDataBlobKeyLength: 256, + MaxDataBlobSize: 64000, + MaxDataTrackCustomEncodingLength: 32, }, Logging: LoggingConfig{ PionLevel: "error", diff --git a/pkg/rtc/participant_data_blob_handler.go b/pkg/rtc/participant_data_blob_handler.go index e6ebcc8ed..8c19ee765 100644 --- a/pkg/rtc/participant_data_blob_handler.go +++ b/pkg/rtc/participant_data_blob_handler.go @@ -41,6 +41,16 @@ func (p *ParticipantImpl) HandleStoreDataBlobRequest(req *livekit.StoreDataBlobR return } + if !p.params.LimitConfig.CheckDataTrackSchemaID(req.Blob.Key.GetSchemaId()) { + p.pubLogger.Warnw("data blob key schema id is invalid", nil, "req", logger.Proto(req)) + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "encoding identifier is empty or exceeds the maximum length", + }) + return + } + if len(req.Blob.Contents) == 0 { p.sendRequestResponse(&livekit.RequestResponse{ RequestId: req.RequestId, diff --git a/pkg/rtc/participant_data_track.go b/pkg/rtc/participant_data_track.go index dbda1ede0..df9854744 100644 --- a/pkg/rtc/participant_data_track.go +++ b/pkg/rtc/participant_data_track.go @@ -60,6 +60,19 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData return } + if !p.params.LimitConfig.CheckDataTrackFrameEncoding(req.FrameEncoding) || + !p.params.LimitConfig.CheckDataTrackSchemaID(req.Schema) { + p.pubLogger.Warnw("invalid encoding identifier", nil, "req", logger.Proto(req)) + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "encoding identifier is empty or exceeds the maximum length", + Request: &livekit.RequestResponse_PublishDataTrack{ + PublishDataTrack: utils.CloneProto(req), + }, + }) + return + } + publishedDataTracks := p.UpDataTrackManager.GetPublishedDataTracks() for _, dt := range publishedDataTracks { message := "" @@ -95,6 +108,8 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData Name: req.Name, Encryption: req.Encryption, } + dti.FrameEncoding = utils.CloneProto(req.GetFrameEncoding()) + dti.Schema = utils.CloneProto(req.GetSchema()) dt := NewDataTrack( DataTrackParams{ Logger: p.params.Logger.WithValues("trackID", dti.Sid), From fcc46d3c45497142d8fe2c6c5756da2ab7d800c3 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 29 Jun 2026 21:33:55 -0700 Subject: [PATCH 09/15] Use camel case log name in `DataBlobKey` (#4633) --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 06e261c39..1fcc70246 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 - github.com/livekit/protocol v1.48.1 + github.com/livekit/protocol v1.48.2 github.com/livekit/psrpc v0.7.2 github.com/mackerelio/go-osstat v0.2.7 github.com/magefile/mage v1.17.2 diff --git a/go.sum b/go.sum index 4e0188b7a..c969726a2 100644 --- a/go.sum +++ b/go.sum @@ -160,8 +160,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 h1:XHNNzebIKZRkLimla/hFGrAIX5EMWHctrgt3hLw7s+I= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.48.1 h1:vFnMqGeknJo6jDwW4wU+x5KuxgA95fPuz7MriAnLzCc= -github.com/livekit/protocol v1.48.1/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= +github.com/livekit/protocol v1.48.2 h1:1Jv1Eckf2jMN7SgJm7fQkSRQoMkGpmuN63mFeh3gd1U= +github.com/livekit/protocol v1.48.2/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= github.com/livekit/psrpc v0.7.2 h1:6oZ+NODJ2pLyaT6VqDq1F4Qc/3TpDUSpyphj/P9MhQc= github.com/livekit/psrpc v0.7.2/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/mackerelio/go-osstat v0.2.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94= From 46e5caedbe0d53217d1b4c84398deb687737f6b6 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 1 Jul 2026 17:06:49 +0800 Subject: [PATCH 10/15] Report average bitrates for whip ingress (#4634) --- pkg/service/roommanager_service.go | 78 +++++++++++++++++++----------- 1 file changed, 50 insertions(+), 28 deletions(-) diff --git a/pkg/service/roommanager_service.go b/pkg/service/roommanager_service.go index 0b5ce6d79..5986fa997 100644 --- a/pkg/service/roommanager_service.go +++ b/pkg/service/roommanager_service.go @@ -12,6 +12,7 @@ import ( "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/rtc/types" + "github.com/livekit/livekit-server/pkg/sfu/rtpstats" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -204,52 +205,73 @@ func (s whipService) sendConnectionNotify(ctx context.Context, participant types } func getMediaStateForParticipant(participant types.Participant) (*livekit.InputVideoState, *livekit.InputAudioState) { - pParticipant := participant.ToProto() - var video *livekit.InputVideoState var audio *livekit.InputAudioState - for _, v := range pParticipant.Tracks { - if v == nil { + for _, t := range participant.GetPublishedTracks() { + if t == nil { continue } - if v.Type != livekit.TrackType_VIDEO { + ti := t.ToProto() + if ti == nil { continue } - video = &livekit.InputVideoState{} + switch t.Kind() { + case livekit.TrackType_VIDEO: + if video != nil { + continue + } - video.MimeType = v.MimeType - video.Height = v.Height - video.Width = v.Width + video = &livekit.InputVideoState{ + MimeType: ti.MimeType, + Width: ti.Width, + Height: ti.Height, + AverageBitrate: trackAverageBitrate(t), + } - break - } + case livekit.TrackType_AUDIO: + if audio != nil { + continue + } - for _, a := range pParticipant.Tracks { - if a == nil { - continue + channels := uint32(1) + if ti.Stereo { + channels = 2 + } + + audio = &livekit.InputAudioState{ + MimeType: ti.MimeType, + Channels: channels, + AverageBitrate: trackAverageBitrate(t), + } } - - if a.Type != livekit.TrackType_AUDIO { - continue - } - - audio = &livekit.InputAudioState{} - - audio.MimeType = a.MimeType - audio.Channels = 1 - if a.Stereo { - audio.Channels = 2 - } - - break } return video, audio } +func trackAverageBitrate(t types.MediaTrack) uint32 { + var allStats []*livekit.RTPStats + for _, r := range t.Receivers() { + if r == nil { + continue + } + + if s := r.GetTrackStats(); s != nil { + allStats = append(allStats, s) + } + } + + agg := rtpstats.AggregateRTPStats(allStats) + if agg == nil { + return 0 + } + + return uint32(agg.Bitrate) +} + // ------------------------------------------- type whipParticipantService struct { From d1b031a9dd4a8e511348ad4f657960edd49f221d Mon Sep 17 00:00:00 2001 From: Benjamin Pracht Date: Thu, 2 Jul 2026 14:07:21 +0100 Subject: [PATCH 11/15] Stop WHIP session notifier when participant leaves (#4637) The WHIP connection-notify loop kept issuing RPCs after the participant had left the room. Guard sendConnectionNotify against a closed participant (returning ErrParticipantNotFound) and treat that error as a clean loop exit. Also reorder DeleteSession so the participant is removed before its WHIP OnClose entry is cleared. Adds tests covering loop termination on participant leave, context cancellation, and the closed-participant guard. Co-authored-by: Claude Opus 4.8 (1M context) --- pkg/service/roommanager_service.go | 13 ++- pkg/service/roommanager_service_test.go | 125 ++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 5 deletions(-) create mode 100644 pkg/service/roommanager_service_test.go diff --git a/pkg/service/roommanager_service.go b/pkg/service/roommanager_service.go index 5986fa997..d8ea05433 100644 --- a/pkg/service/roommanager_service.go +++ b/pkg/service/roommanager_service.go @@ -20,9 +20,8 @@ import ( "github.com/livekit/psrpc" ) -const ( - whipSessionNotifyInterval = 10 * time.Second -) +// whipSessionNotifyInterval is a var (rather than a const) so tests can shorten it. +var whipSessionNotifyInterval = 10 * time.Second type whipService struct { *RoomManager @@ -181,7 +180,7 @@ func (s whipService) notifySession(ctx context.Context, participant types.Partic case <-ticker.C: err := s.sendConnectionNotify(ctx, participant) if err != nil { - if errors.Is(err, context.Canceled) { + if errors.Is(err, context.Canceled) || errors.Is(err, ErrParticipantNotFound) { return nil } } @@ -193,6 +192,10 @@ func (s whipService) notifySession(ctx context.Context, participant types.Partic } func (s whipService) sendConnectionNotify(ctx context.Context, participant types.Participant) error { + if participant.IsClosed() { + return ErrParticipantNotFound + } + video, audio := getMediaStateForParticipant(participant) _, err := s.ingressRpcCli.WHIPRTCConnectionNotify(ctx, string(participant.ID()), &rpc.WHIPRTCConnectionNotifyRequest{ @@ -345,12 +348,12 @@ func (r whipParticipantService) DeleteSession(ctx context.Context, req *rpc.WHIP lp := room.GetParticipantByID(livekit.ParticipantID(req.ParticipantId)) if lp != nil { - lp.AddOnClose(types.ParticipantCloseKeyWHIP, nil) room.RemoveParticipant( lp.Identity(), lp.ID(), types.ParticipantCloseReasonClientRequestLeave, ) + lp.AddOnClose(types.ParticipantCloseKeyWHIP, nil) } return &emptypb.Empty{}, nil diff --git a/pkg/service/roommanager_service_test.go b/pkg/service/roommanager_service_test.go new file mode 100644 index 000000000..e0d588080 --- /dev/null +++ b/pkg/service/roommanager_service_test.go @@ -0,0 +1,125 @@ +package service + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/atomic" + "google.golang.org/protobuf/types/known/emptypb" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/rpc" + "github.com/livekit/psrpc" + + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" +) + +// fakeIngressHandlerClient records WHIPRTCConnectionNotify calls. It embeds the +// interface so only the method under test needs to be implemented; any other +// call would panic (and we assert none happen). +type fakeIngressHandlerClient struct { + rpc.IngressHandlerClient + notifyCount atomic.Int32 +} + +func (f *fakeIngressHandlerClient) WHIPRTCConnectionNotify( + _ context.Context, + _ string, + _ *rpc.WHIPRTCConnectionNotifyRequest, + _ ...psrpc.RequestOption, +) (*emptypb.Empty, error) { + f.notifyCount.Inc() + return &emptypb.Empty{}, nil +} + +// TestWhipNotifySessionStopsWhenParticipantLeaves verifies the notifier loop +// terminates once the WHIP participant leaves the room (i.e. IsClosed becomes +// true), and stops issuing further connection notifications. +func TestWhipNotifySessionStopsWhenParticipantLeaves(t *testing.T) { + origInterval := whipSessionNotifyInterval + whipSessionNotifyInterval = 5 * time.Millisecond + t.Cleanup(func() { whipSessionNotifyInterval = origInterval }) + + var closed atomic.Bool + participant := &typesfakes.FakeParticipant{} + participant.IsClosedStub = func() bool { return closed.Load() } + participant.IDReturns(livekit.ParticipantID("PA_test")) + participant.ToProtoReturns(&livekit.ParticipantInfo{}) + + cli := &fakeIngressHandlerClient{} + s := whipService{ingressRpcCli: cli} + + done := make(chan error, 1) + go func() { + done <- s.notifySession(context.Background(), participant) + }() + + // while the participant is connected the loop should keep notifying + require.Eventually(t, func() bool { + return cli.notifyCount.Load() > 0 + }, time.Second, time.Millisecond, "expected notifications while participant is connected") + + // the participant leaves the room + closed.Store(true) + + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("notifySession did not stop after the participant left the room") + } + + // no further notifications should be attempted after it stops + countAtStop := cli.notifyCount.Load() + time.Sleep(50 * time.Millisecond) + require.Equal(t, countAtStop, cli.notifyCount.Load(), "should not notify after the participant left") +} + +// TestWhipNotifySessionStopsOnContextCancel verifies the loop exits when the +// aliveCtx (cancelled from the participant's OnClose callback) is done. +func TestWhipNotifySessionStopsOnContextCancel(t *testing.T) { + origInterval := whipSessionNotifyInterval + whipSessionNotifyInterval = 5 * time.Millisecond + t.Cleanup(func() { whipSessionNotifyInterval = origInterval }) + + participant := &typesfakes.FakeParticipant{} + participant.IsClosedReturns(false) + participant.IDReturns(livekit.ParticipantID("PA_test")) + participant.ToProtoReturns(&livekit.ParticipantInfo{}) + + cli := &fakeIngressHandlerClient{} + s := whipService{ingressRpcCli: cli} + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- s.notifySession(ctx, participant) + }() + + cancel() + + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("notifySession did not stop after context was cancelled") + } +} + +// TestWhipSendConnectionNotifySkipsClosedParticipant verifies the guard that +// short-circuits the RPC (and drives loop termination) for a closed participant. +func TestWhipSendConnectionNotifySkipsClosedParticipant(t *testing.T) { + participant := &typesfakes.FakeParticipant{} + participant.IsClosedReturns(true) + participant.IDReturns(livekit.ParticipantID("PA_test")) + participant.ToProtoReturns(&livekit.ParticipantInfo{}) + + cli := &fakeIngressHandlerClient{} + s := whipService{ingressRpcCli: cli} + + err := s.sendConnectionNotify(context.Background(), participant) + require.ErrorIs(t, err, ErrParticipantNotFound) + require.Zero(t, cli.notifyCount.Load(), "should not issue an RPC for a closed participant") +} From 8f6a9cb8b735549f0c5770df8ea70ac51f860ecb Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 3 Jul 2026 11:03:59 +0530 Subject: [PATCH 12/15] Release v1.13.3. (#4640) Co-authored-by: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 19 +++++++++++++++++++ version/version.go | 2 +- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f4f6bc55..9ad30c3cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,25 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.13.3] - 2026-07-03 + +### Added + +- feat: mock API server for testing server SDKs (#4627) +- support auth checks with mock server (#4629) +- Data track schema metadata (#4622) +- Report average bitrates for whip ingress (#4634) + +### Changed + +- Update webrtc to fix interop issue with bundled datachannel (#4631) +- Update module github.com/urfave/cli/v3 to v3.10.0 (#4612) +- Use camel case log name in `DataBlobKey` (#4633) + +### Fixed + +- Stop WHIP session notifier when participant leaves (#4637) + ## [1.13.2] - 2026-06-27 ### Added diff --git a/version/version.go b/version/version.go index 3f1dbda2b..2ee36a37b 100644 --- a/version/version.go +++ b/version/version.go @@ -14,4 +14,4 @@ package version -const Version = "1.13.2" +const Version = "1.13.3" From 00348c1299b77aeca8678152f7d3faab76fdebbd Mon Sep 17 00:00:00 2001 From: David Zhao Date: Fri, 3 Jul 2026 13:17:59 +0200 Subject: [PATCH 13/15] simpler mock protocol (#4641) enabling us to build a comprehensive set of API tests for our clients --- cmd/test-server/README.md | 74 ++++++++++++-------- cmd/test-server/auth.go | 20 ++++-- cmd/test-server/config.go | 132 ++++++++++++++++++++++++++++++++++++ cmd/test-server/handlers.go | 54 ++++++++++++--- cmd/test-server/main.go | 71 ++++++------------- 5 files changed, 256 insertions(+), 95 deletions(-) create mode 100644 cmd/test-server/config.go diff --git a/cmd/test-server/README.md b/cmd/test-server/README.md index 1c4eb247a..02991a934 100644 --- a/cmd/test-server/README.md +++ b/cmd/test-server/README.md @@ -7,23 +7,27 @@ image and booted by each SDK's CI. ## Why it looks the way it does -- **Stateless.** All behavior is selected by per-request `X-Lk-Mock-*` headers, - so the server holds no mutable state and tests run in parallel. +- **Stateless.** All behavior is selected by a single per-request `X-Lk-Mock` + header (a JSON object), so the server holds no mutable state and tests run in + parallel. - **Multi-port = multi-region.** The process binds one listener per simulated region (`--ports`). A port's position in the list is its **region index**; index `0` is the primary the SDK is initially pointed at. `GET /settings/regions` advertises all of them in order. - **One header drives every attempt.** The SDK sends the same control header on the initial request *and* every failover retry. Each listener decides what to - do from its **own** index, so a single `X-Lk-Mock-Fail-Regions: 0` makes the - primary fail while the first fallback succeeds — no coordination needed. + do from its **own** index, so a single `X-Lk-Mock: {"failRegions":[0]}` makes + the primary fail while the first fallback succeeds — no coordination needed. +- **Realistic latency.** Methods that block in the real server block here too: + `CreateSIPParticipant` with `wait_until_answered` and `TransferSIPParticipant` + take ~11s before responding, so SDKs can exercise their timeouts. - **The whole API is mocked with populated responses.** Every RoomService, Egress, Ingress, SIP, and Connector method returns a type-correct, populated response: scalar fields that share a name with the request are echoed (e.g. `name`, `metadata`, `identity`, timeouts), `id`/`sid` fields get placeholder values, and list endpoints return one element. Both protobuf and JSON Twirp clients are supported. A client can override the response entirely with the - `X-Lk-Mock-Response` header (see below). Unregistered/future methods fall back + `response` field (see below). Unregistered/future methods fall back to an empty (all-default) message, which still decodes cleanly. ## Running @@ -46,20 +50,31 @@ docker run -p 9999-10002:9999-10002 livekit/test-server ## Control protocol -Request headers (sent by the SDK on API calls; the SDK must forward -client-configured custom headers onto the `/settings/regions` fetch and every -failover retry): +All behavior is driven by a single `X-Lk-Mock` request header whose value is a +JSON object. The SDK sends the same header on API calls, on the +`/settings/regions` fetch, and on every failover retry (it must forward +client-configured custom headers onto all of them). Omit the header — or any +field — for normal behavior. Every field is optional: -| Header | Default | Effect | +| Field | Default | Effect | |---|---|---| -| `X-Lk-Mock-Fail-Regions` | — | comma list of region indices that fail this request, e.g. `0` or `0,1`. Each listener fails only if its own index is listed. | -| `X-Lk-Mock-Fail-Mode` | `status` | how a failing region fails: `status`, `drop` (close connection → transport error), `delay`. | -| `X-Lk-Mock-Fail-Status` | `503` | HTTP status when failing with `status`/`delay`. | -| `X-Lk-Mock-Fail-Twirp-Code` | derived from status | Twirp error code string in the failure body. | -| `X-Lk-Mock-Delay-Ms` | `30000` | delay before a `delay`-mode region responds (for timeout tests). | -| `X-Lk-Mock-Regions-Status` | `200` | override the status of `GET /settings/regions`. | -| `X-Lk-Mock-Response` | — | protojson of the response message for the called method; replaces the populated default, giving full control over the returned payload. | -| `X-Lk-Mock-Skip-Auth` | — | `true` disables permission enforcement for the request (use for tests that aren't about authz, e.g. failover tests with a placeholder token). | +| `failRegions` | — | array of region indices that fail this request, e.g. `[0]` or `[0,1]`. Each listener fails only if its own index is listed. | +| `failMode` | `status` | how a failing region fails: `status` (write a Twirp error), or `drop` (close the connection → transport error). | +| `failStatus` | `503` | HTTP status for a `status`-mode failure. | +| `failTwirpCode` | derived from status | Twirp error code string in the failure body. | +| `delayMs` | — | delay (ms) before responding, on success or failure. Overrides a method's natural latency — use it for timeout tests, or set it to skip a SIP method's built-in ~11s wait. | +| `regionsStatus` | `200` | override the status of `GET /settings/regions`. | +| `response` | — | the response message for the called method (a JSON object, protojson-shaped); replaces the populated default, giving full control over the returned payload. | +| `skipAuth` | `false` | `true` disables permission enforcement for the request (use for tests that aren't about authz, e.g. failover tests with a placeholder token). | + +Example: `X-Lk-Mock: {"skipAuth":true,"failRegions":[0],"failStatus":400}` + +> **Deprecated:** the older per-setting headers — `X-Lk-Mock-Fail-Regions`, +> `X-Lk-Mock-Fail-Mode` (incl. the `delay` mode), `X-Lk-Mock-Fail-Status`, +> `X-Lk-Mock-Fail-Twirp-Code`, `X-Lk-Mock-Delay-Ms`, `X-Lk-Mock-Regions-Status`, +> `X-Lk-Mock-Response`, `X-Lk-Mock-Skip-Auth` — are still honored for existing +> clients and will be removed later. When `X-Lk-Mock` is also present, its fields +> take precedence per-field. New clients should use `X-Lk-Mock` only. Response headers: @@ -96,23 +111,24 @@ is configured with (`secret` by default). | `sip.admin` | SIP trunk & dispatch-rule CRUD | | `sip.call` | `CreateSIPParticipant`; `TransferSIPParticipant` (also needs `roomAdmin`) | -Send `X-Lk-Mock-Skip-Auth: true` to bypass enforcement for tests that aren't +Send `X-Lk-Mock: {"skipAuth":true}` to bypass enforcement for tests that aren't about permissions. ## Common recipes -| Goal | Headers | +| Goal | `X-Lk-Mock` value | |---|---| -| Happy path | valid token with the method's grant → 200 from region `0` | -| Bypass auth (failover tests) | `X-Lk-Mock-Skip-Auth: true` | -| Missing-permission error | token without the required grant → 403 | -| Failover succeeds on region 1 | `X-Lk-Mock-Fail-Regions: 0` | -| Exhaust to region 2 | `X-Lk-Mock-Fail-Regions: 0,1` | -| All regions down | `X-Lk-Mock-Fail-Regions: 0,1,2,3` | -| 4xx, no retry | `X-Lk-Mock-Fail-Regions: 0` + `X-Lk-Mock-Fail-Status: 400` | -| Transport-error failover | `X-Lk-Mock-Fail-Regions: 0` + `X-Lk-Mock-Fail-Mode: drop` | -| Region discovery unreachable | `X-Lk-Mock-Regions-Status: 500` | -| Custom response payload | `X-Lk-Mock-Response: {"sid":"RM_x","name":"my-room"}` | +| Happy path | (no header) — valid token with the method's grant → 200 from region `0` | +| Bypass auth (failover tests) | `{"skipAuth":true}` | +| Missing-permission error | (no header) — token without the required grant → 403 | +| Failover succeeds on region 1 | `{"failRegions":[0]}` | +| Exhaust to region 2 | `{"failRegions":[0,1]}` | +| All regions down | `{"failRegions":[0,1,2,3]}` | +| 4xx, no retry | `{"failRegions":[0],"failStatus":400}` | +| Transport-error failover | `{"failRegions":[0],"failMode":"drop"}` | +| Timeout test | `{"delayMs":30000}` | +| Region discovery unreachable | `{"regionsStatus":500}` | +| Custom response payload | `{"response":{"sid":"RM_x","name":"my-room"}}` | Note: SDK region failover normally only engages for `*.livekit.cloud` hosts. Since tests point at `127.0.0.1`, set the SDK's failover-enable option to its diff --git a/cmd/test-server/auth.go b/cmd/test-server/auth.go index d526e6777..6e270cd00 100644 --- a/cmd/test-server/auth.go +++ b/cmd/test-server/auth.go @@ -31,8 +31,8 @@ import ( // uses), against the mock's configured API secret (default "secret", matching // `livekit-server --dev`); set --api-secret / LK_TEST_SERVER_API_SECRET to change it. // -// Set X-Lk-Mock-Skip-Auth: true to bypass enforcement for tests that aren't -// about permissions (e.g. region-failover tests using a placeholder token). +// Set `"skipAuth": true` in the X-Lk-Mock header to bypass enforcement for tests +// that aren't about permissions (e.g. region-failover tests with a placeholder token). // perm describes the grants a method requires. roomAdmin additionally requires // the token's room to match the request's room; destRoom further requires the @@ -119,8 +119,8 @@ var methodPerms = map[string]perm{ // authorize enforces the permissions a method requires. It returns the HTTP // status and Twirp error code to send (0, "" means authorized / not enforced). -func (h *mockHandler) authorize(key string, r *http.Request, req proto.Message) (int, string) { - if strings.EqualFold(r.Header.Get(headerSkipAuth), "true") { +func (h *mockHandler) authorize(key string, r *http.Request, cfg *mockConfig, req proto.Message) (int, string) { + if cfg.SkipAuth { return 0, "" } p, known := methodPerms[key] @@ -210,3 +210,15 @@ func requestString(req proto.Message, field string) string { } return m.Get(fd).String() } + +func requestBool(req proto.Message, field string) bool { + if req == nil { + return false + } + m := req.ProtoReflect() + fd := m.Descriptor().Fields().ByName(protoreflect.Name(field)) + if fd == nil || fd.Kind() != protoreflect.BoolKind || fd.IsList() { + return false + } + return m.Get(fd).Bool() +} diff --git a/cmd/test-server/config.go b/cmd/test-server/config.go new file mode 100644 index 000000000..2e1aefe40 --- /dev/null +++ b/cmd/test-server/config.go @@ -0,0 +1,132 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "encoding/json" + "net/http" + "strconv" + "strings" +) + +const ( + // headerMock carries the whole mock control config as a JSON object; see + // mockConfig and the README. Sent by the SDK on API calls (and forwarded onto + // the /settings/regions fetch and every failover retry). + headerMock = "X-Lk-Mock" + // headerRegion is set on responses to the index of the region that served it. + headerRegion = "X-Lk-Mock-Region" +) + +// Deprecated: the individual X-Lk-Mock-* control headers predate the unified +// X-Lk-Mock JSON header. They are still honored for existing clients; new +// clients should send X-Lk-Mock instead. When X-Lk-Mock is present its fields +// take precedence over any legacy header. +const ( + legacyHeaderFailRegions = "X-Lk-Mock-Fail-Regions" + legacyHeaderFailMode = "X-Lk-Mock-Fail-Mode" + legacyHeaderFailStatus = "X-Lk-Mock-Fail-Status" + legacyHeaderFailTwirpCode = "X-Lk-Mock-Fail-Twirp-Code" + legacyHeaderDelayMs = "X-Lk-Mock-Delay-Ms" + legacyHeaderRegionsStatus = "X-Lk-Mock-Regions-Status" + legacyHeaderResponse = "X-Lk-Mock-Response" + legacyHeaderSkipAuth = "X-Lk-Mock-Skip-Auth" +) + +// legacyDefaultDelayMs is the sleep used by the deprecated "delay" fail mode when +// no X-Lk-Mock-Delay-Ms is given (long enough to trip client timeouts). +const legacyDefaultDelayMs = 30_000 + +// mockConfig is the JSON value of the X-Lk-Mock request header. Every field is +// optional; the zero value means "behave normally". A single object keeps the +// control protocol simple — the SDK serializes one struct instead of juggling a +// header per knob. +type mockConfig struct { + // FailRegions lists region indices that should fail this request. A listener + // fails only if its own region index appears here, so one config can make the + // primary fail while a fallback succeeds. + FailRegions []int `json:"failRegions,omitempty"` + // FailMode selects how a failing region fails: "status" (default) writes a + // Twirp error; "drop" closes the connection to force a transport error. + // ("delay" is a deprecated legacy mode; new clients use DelayMs instead.) + FailMode string `json:"failMode,omitempty"` + // FailStatus is the HTTP status for a "status"-mode failure (default 503). + FailStatus int `json:"failStatus,omitempty"` + // FailTwirpCode overrides the Twirp error code string in the failure body + // (default derived from FailStatus). + FailTwirpCode string `json:"failTwirpCode,omitempty"` + // DelayMs delays the response by this many milliseconds before returning, + // whether the region succeeds or fails. It overrides a method's natural + // latency (see methodLatency): set it high for timeout tests, or to 0 to skip + // a SIP method's built-in wait. Nil means "use the natural latency". + DelayMs *int `json:"delayMs,omitempty"` + // RegionsStatus overrides the HTTP status of GET /settings/regions (default 200). + RegionsStatus int `json:"regionsStatus,omitempty"` + // Response is the protojson of the response message for the called method; it + // replaces the populated default, giving full control over the payload. + Response json.RawMessage `json:"response,omitempty"` + // SkipAuth disables permission enforcement for this request (for tests that + // aren't about authz, e.g. failover tests with a placeholder token). + SkipAuth bool `json:"skipAuth,omitempty"` + + // legacyDelayMs is the sleep used by the deprecated "delay" fail mode. It is + // populated only from the legacy X-Lk-Mock-Delay-Ms header, never from JSON. + legacyDelayMs int +} + +// parseMockConfig builds the request's config. Deprecated individual X-Lk-Mock-* +// headers form the base; the unified X-Lk-Mock JSON header (if present) is +// overlaid on top, so its fields win per-field while absent fields keep the +// legacy value. +func parseMockConfig(r *http.Request) mockConfig { + cfg := parseLegacyConfig(r) + if v := r.Header.Get(headerMock); v != "" { + // Unmarshal overwrites only the fields present in the JSON; the unexported + // legacyDelayMs is untouched. + _ = json.Unmarshal([]byte(v), &cfg) + } + return cfg +} + +// parseLegacyConfig reads the deprecated per-setting headers into a config. +func parseLegacyConfig(r *http.Request) mockConfig { + var cfg mockConfig + for _, part := range strings.Split(r.Header.Get(legacyHeaderFailRegions), ",") { + if idx, err := strconv.Atoi(strings.TrimSpace(part)); err == nil { + cfg.FailRegions = append(cfg.FailRegions, idx) + } + } + cfg.FailMode = r.Header.Get(legacyHeaderFailMode) + cfg.FailStatus = parseStatus(r.Header.Get(legacyHeaderFailStatus)) + cfg.FailTwirpCode = r.Header.Get(legacyHeaderFailTwirpCode) + cfg.RegionsStatus = parseStatus(r.Header.Get(legacyHeaderRegionsStatus)) + if resp := r.Header.Get(legacyHeaderResponse); resp != "" { + cfg.Response = json.RawMessage(resp) + } + cfg.SkipAuth = strings.EqualFold(r.Header.Get(legacyHeaderSkipAuth), "true") + cfg.legacyDelayMs = legacyDefaultDelayMs + if ms, err := strconv.Atoi(r.Header.Get(legacyHeaderDelayMs)); err == nil && ms >= 0 { + cfg.legacyDelayMs = ms + } + return cfg +} + +// parseStatus returns a valid HTTP status from s, or 0 if absent/invalid. +func parseStatus(s string) int { + if v, err := strconv.Atoi(strings.TrimSpace(s)); err == nil && v >= 100 && v <= 599 { + return v + } + return 0 +} diff --git a/cmd/test-server/handlers.go b/cmd/test-server/handlers.go index 7de083bc4..7878b7856 100644 --- a/cmd/test-server/handlers.go +++ b/cmd/test-server/handlers.go @@ -19,6 +19,7 @@ import ( "net/http" "strconv" "strings" + "time" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/reflect/protoreflect" @@ -135,25 +136,58 @@ func (h *mockHandler) serveAPI(w http.ResponseWriter, r *http.Request) { } } + cfg := parseMockConfig(r) + // Permission enforcement comes first, mirroring the real server. - if status, code := h.authorize(key, r, req); status != 0 { + if status, code := h.authorize(key, r, &cfg, req); status != 0 { writeTwirpErrorCode(w, status, code, "mock: "+code) return } - if h.shouldFail(r) { - h.fail(w, r) + // Delay before responding (success or failure). An explicit delayMs overrides + // the method's natural latency — e.g. CreateSIPParticipant blocking until the + // callee answers. + delay := methodLatency(key, req) + if cfg.DelayMs != nil { + delay = time.Duration(*cfg.DelayMs) * time.Millisecond + } + if delay > 0 { + time.Sleep(delay) + } + + if h.shouldFail(&cfg) { + h.fail(w, &cfg) return } - h.writeAPIResponse(w, r, json, known, req, spec) + h.writeAPIResponse(w, json, known, req, spec, &cfg) } +// methodLatency returns the realistic time a method blocks before responding, so +// the mock approximates the real server's behavior. CreateSIPParticipant blocks +// until the callee answers when wait_until_answered is set; TransferSIPParticipant +// always blocks until the transfer (REFER) completes. +func methodLatency(key string, req proto.Message) time.Duration { + switch key { + case "livekit.SIP/CreateSIPParticipant": + if requestBool(req, "wait_until_answered") { + return sipAnswerLatency + } + case "livekit.SIP/TransferSIPParticipant": + return sipAnswerLatency + } + return 0 +} + +// sipAnswerLatency is how long a SIP call takes to be answered/transferred in the +// mock — long enough to exercise client-side timeouts around these calls. +const sipAnswerLatency = 11 * time.Second + // writeAPIResponse serves a populated, type-correct response for a known API -// method. The response is the reflection-populated default unless the request -// carries an X-Lk-Mock-Response header (protojson), which overrides it -// entirely. Content type (protobuf vs JSON) mirrors the request. -func (h *mockHandler) writeAPIResponse(w http.ResponseWriter, r *http.Request, json, known bool, req proto.Message, spec apiSpec) { +// method. The response is the reflection-populated default unless the mock +// config carries a `response` (protojson), which overrides it entirely. Content +// type (protobuf vs JSON) mirrors the request. +func (h *mockHandler) writeAPIResponse(w http.ResponseWriter, json, known bool, req proto.Message, spec apiSpec, cfg *mockConfig) { w.Header().Set(headerRegion, strconv.Itoa(h.regionIndex)) if !known { @@ -164,8 +198,8 @@ func (h *mockHandler) writeAPIResponse(w http.ResponseWriter, r *http.Request, j } resp := spec.newResp() - if override := r.Header.Get(headerResponse); override != "" { - if err := protojson.Unmarshal([]byte(override), resp); err != nil { + if len(cfg.Response) > 0 { + if err := protojson.Unmarshal(cfg.Response, resp); err != nil { // Malformed override: fall back to the populated default. resp = spec.newResp() populateMessage(resp.ProtoReflect(), req.ProtoReflect(), 1) diff --git a/cmd/test-server/main.go b/cmd/test-server/main.go index 07449539d..5d54885b6 100644 --- a/cmd/test-server/main.go +++ b/cmd/test-server/main.go @@ -22,6 +22,7 @@ import ( "net/http" "os" "os/signal" + "slices" "strconv" "strings" "syscall" @@ -31,23 +32,6 @@ import ( "github.com/livekit/protocol/utils/protojson" ) -// X-Lk-Mock-* request headers control the mock's behavior; see the README. -const ( - headerFailRegions = "X-Lk-Mock-Fail-Regions" - headerFailMode = "X-Lk-Mock-Fail-Mode" - headerFailStatus = "X-Lk-Mock-Fail-Status" - headerFailTwirpCode = "X-Lk-Mock-Fail-Twirp-Code" - headerDelayMs = "X-Lk-Mock-Delay-Ms" - headerRegionsStatus = "X-Lk-Mock-Regions-Status" - headerResponse = "X-Lk-Mock-Response" - // headerSkipAuth disables permission enforcement for a request. - headerSkipAuth = "X-Lk-Mock-Skip-Auth" - // headerRegion is set on responses to the index of the region that served it. - headerRegion = "X-Lk-Mock-Region" -) - -const defaultDelayMs = 30_000 - func main() { portsFlag := flagValue("--ports", "LK_TEST_SERVER_PORTS", "9999,10000,10001,10002") advertiseHost := flagValue("--advertise-host", "LK_TEST_SERVER_ADVERTISE_HOST", "http://127.0.0.1") @@ -116,8 +100,9 @@ func (h *mockHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } func (h *mockHandler) handleRegions(w http.ResponseWriter, r *http.Request) { - if status := parseStatus(r.Header.Get(headerRegionsStatus), 0); status != 0 && status != http.StatusOK { - w.WriteHeader(status) + cfg := parseMockConfig(r) + if cfg.RegionsStatus != 0 && cfg.RegionsStatus != http.StatusOK { + w.WriteHeader(cfg.RegionsStatus) return } body, err := protojson.Marshal(h.regions) @@ -135,21 +120,12 @@ func (h *mockHandler) handleTwirp(w http.ResponseWriter, r *http.Request) { h.serveAPI(w, r) } -func (h *mockHandler) shouldFail(r *http.Request) bool { - for _, part := range strings.Split(r.Header.Get(headerFailRegions), ",") { - part = strings.TrimSpace(part) - if part == "" { - continue - } - if idx, err := strconv.Atoi(part); err == nil && idx == h.regionIndex { - return true - } - } - return false +func (h *mockHandler) shouldFail(cfg *mockConfig) bool { + return slices.Contains(cfg.FailRegions, h.regionIndex) } -func (h *mockHandler) fail(w http.ResponseWriter, r *http.Request) { - switch strings.ToLower(r.Header.Get(headerFailMode)) { +func (h *mockHandler) fail(w http.ResponseWriter, cfg *mockConfig) { + switch strings.ToLower(cfg.FailMode) { case "drop": if hj, ok := w.(http.Hijacker); ok { if conn, _, err := hj.Hijack(); err == nil { @@ -158,20 +134,21 @@ func (h *mockHandler) fail(w http.ResponseWriter, r *http.Request) { } } w.WriteHeader(http.StatusServiceUnavailable) + return case "delay": - delay := defaultDelayMs - if ms, err := strconv.Atoi(r.Header.Get(headerDelayMs)); err == nil && ms >= 0 { - delay = ms - } - time.Sleep(time.Duration(delay) * time.Millisecond) - writeTwirpError(w, r, parseStatus(r.Header.Get(headerFailStatus), http.StatusServiceUnavailable)) - default: - writeTwirpError(w, r, parseStatus(r.Header.Get(headerFailStatus), http.StatusServiceUnavailable)) + // Deprecated legacy mode: sleep, then status-fail. New clients should set + // DelayMs (which delays every response) instead. + time.Sleep(time.Duration(cfg.legacyDelayMs) * time.Millisecond) } + status := cfg.FailStatus + if status < 100 || status > 599 { + status = http.StatusServiceUnavailable + } + writeTwirpError(w, cfg, status) } -func writeTwirpError(w http.ResponseWriter, r *http.Request, status int) { - code := r.Header.Get(headerFailTwirpCode) +func writeTwirpError(w http.ResponseWriter, cfg *mockConfig, status int) { + code := cfg.FailTwirpCode if code == "" { code = twirpCodeForStatus(status) } @@ -205,16 +182,6 @@ func twirpCodeForStatus(status int) string { } } -func parseStatus(s string, def int) int { - if s == "" { - return def - } - if v, err := strconv.Atoi(strings.TrimSpace(s)); err == nil && v >= 100 && v <= 599 { - return v - } - return def -} - func parsePorts(s string) ([]int, error) { var ports []int for _, part := range strings.Split(s, ",") { From 0bc73fd0daf5b312bdc06706fd391efd6018931e Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sat, 4 Jul 2026 21:47:02 +0200 Subject: [PATCH 14/15] added mocking function for SIP dialing methods (#4642) --- cmd/test-server/README.md | 3 +++ cmd/test-server/config.go | 14 ++++++++++++++ cmd/test-server/handlers.go | 28 ++++++++++++++++++++++++++++ cmd/test-server/main.go | 20 ++++++++++++++++++++ 4 files changed, 65 insertions(+) diff --git a/cmd/test-server/README.md b/cmd/test-server/README.md index 02991a934..2f13f00da 100644 --- a/cmd/test-server/README.md +++ b/cmd/test-server/README.md @@ -66,6 +66,7 @@ field — for normal behavior. Every field is optional: | `regionsStatus` | `200` | override the status of `GET /settings/regions`. | | `response` | — | the response message for the called method (a JSON object, protojson-shaped); replaces the populated default, giving full control over the returned payload. | | `skipAuth` | `false` | `true` disables permission enforcement for the request (use for tests that aren't about authz, e.g. failover tests with a placeholder token). | +| `sipStatus` | — | fail a SIP dial method (`CreateSIPParticipant`/`TransferSIPParticipant`) with a SIP status, e.g. `{"code":486,"status":"Busy Here"}` (`status` optional). The Twirp error code and `sip_status_code`/`sip_status`/`error_details` metadata are derived from it exactly as the real server does. Composes with `delayMs` to simulate "ring, then fail". | Example: `X-Lk-Mock: {"skipAuth":true,"failRegions":[0],"failStatus":400}` @@ -129,6 +130,8 @@ about permissions. | Timeout test | `{"delayMs":30000}` | | Region discovery unreachable | `{"regionsStatus":500}` | | Custom response payload | `{"response":{"sid":"RM_x","name":"my-room"}}` | +| SIP busy signal | `{"sipStatus":{"code":486,"status":"Busy Here"}}` | +| SIP carrier decline | `{"sipStatus":{"code":603}}` | Note: SDK region failover normally only engages for `*.livekit.cloud` hosts. Since tests point at `127.0.0.1`, set the SDK's failover-enable option to its diff --git a/cmd/test-server/config.go b/cmd/test-server/config.go index 2e1aefe40..bca4ee251 100644 --- a/cmd/test-server/config.go +++ b/cmd/test-server/config.go @@ -80,12 +80,26 @@ type mockConfig struct { // SkipAuth disables permission enforcement for this request (for tests that // aren't about authz, e.g. failover tests with a placeholder token). SkipAuth bool `json:"skipAuth,omitempty"` + // SIPStatus, when set on a SIP dial method (CreateSIPParticipant / + // TransferSIPParticipant), fails the call with this SIP status. The Twirp + // error code and metadata (sip_status_code, sip_status, error_details) are + // derived from it exactly as the real server does, so the SDK sees an + // identical error. Composes with DelayMs to simulate "ring, then fail". + SIPStatus *sipStatusConfig `json:"sipStatus,omitempty"` // legacyDelayMs is the sleep used by the deprecated "delay" fail mode. It is // populated only from the legacy X-Lk-Mock-Delay-Ms header, never from JSON. legacyDelayMs int } +// sipStatusConfig is a SIP response to inject; see mockConfig.SIPStatus. +type sipStatusConfig struct { + // Code is the SIP response code, e.g. 486 (Busy Here) or 603 (Decline). + Code int `json:"code"` + // Status is the SIP reason phrase; defaults to the code's canonical name. + Status string `json:"status,omitempty"` +} + // parseMockConfig builds the request's config. Deprecated individual X-Lk-Mock-* // headers form the base; the unified X-Lk-Mock JSON header (if present) is // overlaid on top, so its fields win per-field while absent fields keep the diff --git a/cmd/test-server/handlers.go b/cmd/test-server/handlers.go index 7878b7856..9d14fd2b3 100644 --- a/cmd/test-server/handlers.go +++ b/cmd/test-server/handlers.go @@ -27,6 +27,7 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/utils/protojson" + "github.com/livekit/protocol/utils/xtwirp" ) // apiSpec captures the request and response message types for one Twirp method, @@ -155,6 +156,13 @@ func (h *mockHandler) serveAPI(w http.ResponseWriter, r *http.Request) { time.Sleep(delay) } + // A SIP dial that fails carries a SIP status; the Twirp code and metadata are + // derived from it exactly as the real server does. + if cfg.SIPStatus != nil && isSIPDialMethod(key) { + h.failSIP(w, &cfg) + return + } + if h.shouldFail(&cfg) { h.fail(w, &cfg) return @@ -183,6 +191,26 @@ func methodLatency(key string, req proto.Message) time.Duration { // mock — long enough to exercise client-side timeouts around these calls. const sipAnswerLatency = 11 * time.Second +// isSIPDialMethod reports whether key places a call that can fail with a SIP status. +func isSIPDialMethod(key string) bool { + switch key { + case "livekit.SIP/CreateSIPParticipant", "livekit.SIP/TransferSIPParticipant": + return true + } + return false +} + +// failSIP fails the request with the configured SIP status, mirroring the real +// server: the status maps to a Twirp error code and attaches sip_status_code, +// sip_status, and error_details metadata via xtwirp. +func (h *mockHandler) failSIP(w http.ResponseWriter, cfg *mockConfig) { + st := &livekit.SIPStatus{ + Code: livekit.SIPStatusCode(cfg.SIPStatus.Code), + Status: cfg.SIPStatus.Status, + } + writeTwirpErr(w, xtwirp.ToError(st)) +} + // writeAPIResponse serves a populated, type-correct response for a known API // method. The response is the reflection-populated default unless the mock // config carries a `response` (protojson), which overrides it entirely. Content diff --git a/cmd/test-server/main.go b/cmd/test-server/main.go index 5d54885b6..7850a4c7c 100644 --- a/cmd/test-server/main.go +++ b/cmd/test-server/main.go @@ -17,6 +17,7 @@ package main import ( + "encoding/json" "errors" "fmt" "net/http" @@ -28,6 +29,8 @@ import ( "syscall" "time" + "github.com/twitchtv/twirp" + "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/utils/protojson" ) @@ -163,6 +166,23 @@ func writeTwirpErrorCode(w http.ResponseWriter, status int, code, msg string) { _, _ = fmt.Fprintf(w, `{"code":%q,"msg":%q}`, code, msg) } +// writeTwirpErr writes a full Twirp JSON error — code, message, and metadata — +// using the HTTP status Twirp derives from the error code. +func writeTwirpErr(w http.ResponseWriter, terr twirp.Error) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set(headerRegion, "") + w.WriteHeader(twirp.ServerHTTPStatusFromErrorCode(terr.Code())) + _ = json.NewEncoder(w).Encode(struct { + Code string `json:"code"` + Msg string `json:"msg"` + Meta map[string]string `json:"meta,omitempty"` + }{ + Code: string(terr.Code()), + Msg: terr.Msg(), + Meta: terr.MetaMap(), + }) +} + func twirpCodeForStatus(status int) string { switch { case status == http.StatusBadRequest: From bf777e65132e393aa286bccf0463d6fd8c6d31be Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 5 Jul 2026 16:08:10 +0530 Subject: [PATCH 15/15] Make IsConnectionCanceled available at LocalParticipant interface. (#4643) Can be used to keep track of pariticpants failing connection in a room by checking this when room closes the participant. --- pkg/rtc/participant.go | 14 ++-- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 72 +++++++++++++++++++ 3 files changed, 82 insertions(+), 5 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 915a7022e..780deea9c 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1427,19 +1427,23 @@ func (p *ParticipantImpl) IsReconnect() bool { } func (p *ParticipantImpl) maybeRecordRTCanceled(closeReason types.ParticipantCloseReason) { - if p.State() >= livekit.ParticipantInfo_ACTIVE { + if p.HasConnected() { return } - if closeReason == types.ParticipantCloseReasonClientRequestLeave || + if p.IsConnectionCanceled(closeReason) { + prometheus.IncrementParticipantRtcCanceled(1) + } +} + +func (p *ParticipantImpl) IsConnectionCanceled(closeReason types.ParticipantCloseReason) bool { + return closeReason == types.ParticipantCloseReasonClientRequestLeave || closeReason == types.ParticipantCloseReasonDuplicateIdentity || closeReason == types.ParticipantCloseReasonRoomClosed || closeReason == types.ParticipantCloseReasonMigrationRequested || closeReason == types.ParticipantCloseReasonMigrationComplete || // client closing signal connection too quickly, there is a time check to handle clients timing out and leaving without sending a leave message - (time.Since(p.params.SessionStartTime) < 3*time.Second && closeReason == types.ParticipantCloseReasonSignalSourceClose) { - prometheus.IncrementParticipantRtcCanceled(1) - } + (time.Since(p.params.SessionStartTime) < 3*time.Second && closeReason == types.ParticipantCloseReasonSignalSourceClose) } func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseReason, isExpectedToResume bool) error { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 5d0c702ed..c47802ce5 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -414,6 +414,7 @@ type LocalParticipant interface { IsReady() bool ActiveAt() time.Time Disconnected() <-chan struct{} + IsConnectionCanceled(closeReason ParticipantCloseReason) bool IsIdle() bool SubscriberAsPrimary() bool GetClientInfo() *livekit.ClientInfo diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index d4198c2de..b7461f9ff 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -841,6 +841,17 @@ type FakeLocalParticipant struct { isClosedReturnsOnCall map[int]struct { result1 bool } + IsConnectionCanceledStub func(types.ParticipantCloseReason) bool + isConnectionCanceledMutex sync.RWMutex + isConnectionCanceledArgsForCall []struct { + arg1 types.ParticipantCloseReason + } + isConnectionCanceledReturns struct { + result1 bool + } + isConnectionCanceledReturnsOnCall map[int]struct { + result1 bool + } IsDependentStub func() bool isDependentMutex sync.RWMutex isDependentArgsForCall []struct { @@ -5946,6 +5957,67 @@ func (fake *FakeLocalParticipant) IsClosedReturnsOnCall(i int, result1 bool) { }{result1} } +func (fake *FakeLocalParticipant) IsConnectionCanceled(arg1 types.ParticipantCloseReason) bool { + fake.isConnectionCanceledMutex.Lock() + ret, specificReturn := fake.isConnectionCanceledReturnsOnCall[len(fake.isConnectionCanceledArgsForCall)] + fake.isConnectionCanceledArgsForCall = append(fake.isConnectionCanceledArgsForCall, struct { + arg1 types.ParticipantCloseReason + }{arg1}) + stub := fake.IsConnectionCanceledStub + fakeReturns := fake.isConnectionCanceledReturns + fake.recordInvocation("IsConnectionCanceled", []interface{}{arg1}) + fake.isConnectionCanceledMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) IsConnectionCanceledCallCount() int { + fake.isConnectionCanceledMutex.RLock() + defer fake.isConnectionCanceledMutex.RUnlock() + return len(fake.isConnectionCanceledArgsForCall) +} + +func (fake *FakeLocalParticipant) IsConnectionCanceledCalls(stub func(types.ParticipantCloseReason) bool) { + fake.isConnectionCanceledMutex.Lock() + defer fake.isConnectionCanceledMutex.Unlock() + fake.IsConnectionCanceledStub = stub +} + +func (fake *FakeLocalParticipant) IsConnectionCanceledArgsForCall(i int) types.ParticipantCloseReason { + fake.isConnectionCanceledMutex.RLock() + defer fake.isConnectionCanceledMutex.RUnlock() + argsForCall := fake.isConnectionCanceledArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) IsConnectionCanceledReturns(result1 bool) { + fake.isConnectionCanceledMutex.Lock() + defer fake.isConnectionCanceledMutex.Unlock() + fake.IsConnectionCanceledStub = nil + fake.isConnectionCanceledReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) IsConnectionCanceledReturnsOnCall(i int, result1 bool) { + fake.isConnectionCanceledMutex.Lock() + defer fake.isConnectionCanceledMutex.Unlock() + fake.IsConnectionCanceledStub = nil + if fake.isConnectionCanceledReturnsOnCall == nil { + fake.isConnectionCanceledReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.isConnectionCanceledReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) IsDependent() bool { fake.isDependentMutex.Lock() ret, specificReturn := fake.isDependentReturnsOnCall[len(fake.isDependentArgsForCall)]