diff --git a/pkg/sip/attrs_headers_test.go b/pkg/sip/attrs_headers_test.go new file mode 100644 index 00000000..442199c3 --- /dev/null +++ b/pkg/sip/attrs_headers_test.go @@ -0,0 +1,79 @@ +// 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 sip + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// Regression for livekit/sip#404: when the agent deletes the room before SIP +// sends BYE, LocalParticipant is gone. attributes_to_headers must still map +// from the last cached participant attributes. +func TestFillHeadersUsesCachedAttrsWhenRoomNil(t *testing.T) { + call := &inboundCall{ + attrsToHdr: map[string]string{ + "sip.custom": "X-Custom-Header", + }, + } + call.storeParticipantAttrs(map[string]string{ + "sip.custom": "value-from-cache", + "other": "ignored", + }) + call.lkRoom = nil + cc := &sipInbound{call: call} + + headers := cc.fillHeaders(nil) + require.Equal(t, map[string]string{"X-Custom-Header": "value-from-cache"}, headers) + + // No mapping configured → leave headers untouched. + call.attrsToHdr = nil + require.Nil(t, cc.fillHeaders(nil)) + + // Mapping configured but cache empty → leave headers untouched. + call.attrsToHdr = map[string]string{"sip.custom": "X-Custom-Header"} + call.attrsMu.Lock() + call.cachedAttrs = nil + call.attrsMu.Unlock() + require.Nil(t, cc.fillHeaders(nil)) +} + +func TestOutboundSetAttrsToHeadersUsesCachedAttrsWhenRoomNil(t *testing.T) { + call := &outboundCall{ + sipConf: sipOutboundConfig{ + attrsToHeaders: map[string]string{ + "sip.custom": "X-Custom-Header", + }, + }, + } + call.storeParticipantAttrs(map[string]string{ + "sip.custom": "outbound-cache", + }) + call.lkRoom = nil + + headers := call.setAttrsToHeaders(nil) + require.Equal(t, map[string]string{"X-Custom-Header": "outbound-cache"}, headers) +} + +func TestAttrsToHeaders(t *testing.T) { + attrs := map[string]string{"a": "1", "b": "2"} + mapping := map[string]string{"a": "X-A", "missing": "X-Missing"} + headers := AttrsToHeaders(attrs, mapping, map[string]string{"Keep": "yes"}) + require.Equal(t, map[string]string{ + "Keep": "yes", + "X-A": "1", + }, headers) +} diff --git a/pkg/sip/inbound.go b/pkg/sip/inbound.go index 7da5fe54..2912cb19 100644 --- a/pkg/sip/inbound.go +++ b/pkg/sip/inbound.go @@ -684,6 +684,8 @@ type inboundCall struct { callStart time.Time extraAttrs map[string]string attrsToHdr map[string]string + attrsMu sync.Mutex + cachedAttrs map[string]string // last-seen participant attrs for BYE/REFER after room teardown (#404) ctx context.Context cancel func() closeReason atomic.Pointer[ReasonHeader] @@ -1368,10 +1370,11 @@ func (c *inboundCall) close(ctx context.Context, end EndCall) { defer log.Infow("Inbound call closed") } + // Snapshot attrs before teardown. Prefer live room state, but keep the + // cache so attributes_to_headers still works when the agent deleted the + // room first (Room() is already nil). See livekit/sip#404. + c.snapshotParticipantAttrs() // Send BYE _before_ closing media/room connection. - // This ensures participant attributes are still available for - // attributes_to_headers mapping in the setHeaders callback. - // See: https://github.com/livekit/sip/issues/404 c.cc.CloseWithStatus(ctx, result, end.Headers) c.closeMedia() if callDurFn := c.callDur; callDurFn != nil { @@ -1514,6 +1517,7 @@ func (c *inboundCall) setStatus(v CallStatus) { r.LocalParticipant.SetAttributes(map[string]string{ livekit.AttrSIPCallStatus: attr, }) + c.snapshotParticipantAttrs() } func (c *inboundCall) createLiveKitParticipant(ctx context.Context, rconf RoomConfig, status CallStatus) error { @@ -1547,6 +1551,10 @@ func (c *inboundCall) createLiveKitParticipant(ctx context.Context, rconf RoomCo if err != nil { return err } + // Seed attrs cache from the join config so BYE mapping works even if the + // room is torn down before we read LocalParticipant again (#404). + c.storeParticipantAttrs(partConf.Attributes) + c.snapshotParticipantAttrs() if err := registerSignalingRPC(c.lkRoom, c.cc); err != nil { return err } @@ -1788,11 +1796,44 @@ func (c *sipInbound) fillHeaders(headers map[string]string) map[string]string { if c == nil || c.call == nil || len(c.call.attrsToHdr) == 0 { return headers } - r := c.call.lkRoom.Room() - if r == nil { + attrs := c.call.participantAttributes() + if len(attrs) == 0 { return headers } - return AttrsToHeaders(r.LocalParticipant.Attributes(), c.call.attrsToHdr, headers) + return AttrsToHeaders(attrs, c.call.attrsToHdr, headers) +} + +// snapshotParticipantAttrs caches LocalParticipant attributes while the room +// is still connected. Used so BYE/REFER can map attributes_to_headers after +// the room has already been torn down (livekit/sip#404). +func (c *inboundCall) snapshotParticipantAttrs() { + if c == nil || c.lkRoom == nil { + return + } + r := c.lkRoom.Room() + if r == nil || r.LocalParticipant == nil { + return + } + attrs := r.LocalParticipant.Attributes() // clones + c.attrsMu.Lock() + c.cachedAttrs = attrs + c.attrsMu.Unlock() +} + +func (c *inboundCall) storeParticipantAttrs(attrs map[string]string) { + if c == nil || len(attrs) == 0 { + return + } + c.attrsMu.Lock() + c.cachedAttrs = maps.Clone(attrs) + c.attrsMu.Unlock() +} + +func (c *inboundCall) participantAttributes() map[string]string { + c.snapshotParticipantAttrs() + c.attrsMu.Lock() + defer c.attrsMu.Unlock() + return maps.Clone(c.cachedAttrs) } func (c *sipInbound) Drop() { diff --git a/pkg/sip/outbound.go b/pkg/sip/outbound.go index 89d5252c..b1222635 100644 --- a/pkg/sip/outbound.go +++ b/pkg/sip/outbound.go @@ -82,11 +82,13 @@ type outboundCall struct { jitterBuf bool projectID string - mu sync.RWMutex - mon *stats.CallMonitor - lkRoom RoomInterface - lkRoomIn msdk.PCM16Writer // output to room; OPUS at 48k - sipConf sipOutboundConfig + mu sync.RWMutex + mon *stats.CallMonitor + lkRoom RoomInterface + lkRoomIn msdk.PCM16Writer // output to room; OPUS at 48k + sipConf sipOutboundConfig + attrsMu sync.Mutex + cachedAttrs map[string]string // last-seen participant attrs for BYE after room teardown (#404) } func (c *Client) newCall(ctx context.Context, tid traceid.ID, conf *config.Config, log logger.Logger, id LocalTag, room RoomConfig, sipConf sipOutboundConfig, state *CallState, projectID string) (*outboundCall, error) { @@ -171,11 +173,41 @@ func (c *outboundCall) setAttrsToHeaders(headers map[string]string) map[string]s if len(c.sipConf.attrsToHeaders) == 0 { return headers } - r := c.lkRoom.Room() - if r == nil { + attrs := c.participantAttributes() + if len(attrs) == 0 { return headers } - return AttrsToHeaders(r.LocalParticipant.Attributes(), c.sipConf.attrsToHeaders, headers) + return AttrsToHeaders(attrs, c.sipConf.attrsToHeaders, headers) +} + +func (c *outboundCall) snapshotParticipantAttrs() { + if c == nil || c.lkRoom == nil { + return + } + r := c.lkRoom.Room() + if r == nil || r.LocalParticipant == nil { + return + } + attrs := r.LocalParticipant.Attributes() // clones + c.attrsMu.Lock() + c.cachedAttrs = attrs + c.attrsMu.Unlock() +} + +func (c *outboundCall) storeParticipantAttrs(attrs map[string]string) { + if c == nil || len(attrs) == 0 { + return + } + c.attrsMu.Lock() + c.cachedAttrs = maps.Clone(attrs) + c.attrsMu.Unlock() +} + +func (c *outboundCall) participantAttributes() map[string]string { + c.snapshotParticipantAttrs() + c.attrsMu.Lock() + defer c.attrsMu.Unlock() + return maps.Clone(c.cachedAttrs) } func (c *outboundCall) ensureClosed(ctx context.Context) { @@ -375,10 +407,10 @@ func (c *outboundCall) close(ctx context.Context, end EndCall) bool { info.DisconnectReason = end.Reason }) + // Snapshot attrs before teardown so attributes_to_headers still works + // when the room was already deleted (livekit/sip#404). + c.snapshotParticipantAttrs() // Send BYE _before_ closing media/room connection. - // This ensures participant attributes are still available for - // attributes_to_headers mapping in the setHeaders callback. - // See: https://github.com/livekit/sip/issues/404 c.stopSIP(ctx, end.Term, end.Headers) if c.media != nil { c.media.Close() @@ -473,6 +505,8 @@ func (c *outboundCall) connectToRoom(ctx context.Context, lkNew RoomConfig, getR } c.lkRoom = r c.lkRoomIn = local + c.storeParticipantAttrs(attrs) + c.snapshotParticipantAttrs() if err := registerSignalingRPC(c.lkRoom, c.cc); err != nil { return err } @@ -641,6 +675,7 @@ func (c *outboundCall) setStatus(v CallStatus) { r.LocalParticipant.SetAttributes(map[string]string{ livekit.AttrSIPCallStatus: attr, }) + c.snapshotParticipantAttrs() } func (c *outboundCall) setExtraAttrs(hdrToAttr map[string]string, opts livekit.SIPHeaderOptions, cc Signaling, hdrs Headers) { @@ -649,6 +684,7 @@ func (c *outboundCall) setExtraAttrs(hdrToAttr map[string]string, opts livekit.S room := c.lkRoom.Room() if room != nil { room.LocalParticipant.SetAttributes(extra) + c.snapshotParticipantAttrs() } else { c.log.Warnw("could not set attributes on nil room", nil, "attrs", extra) }