diff --git a/pkg/sip/media_pipeline.go b/pkg/sip/media_pipeline.go index 23b4c632..e73a18a2 100644 --- a/pkg/sip/media_pipeline.go +++ b/pkg/sip/media_pipeline.go @@ -148,16 +148,19 @@ func (p *MediaPort) dtmfHandler(h *rtp.Header, payload []byte) error { if fnc == nil { return nil } - // RFC 4733 requires all packets of a given digit to share identical timestamps. - // The marker bit could be used instead, but it is prone to occasional loss. - if h.Timestamp == p.lastDTMFTimestamp.Load() { - return nil - } ev, err := dtmf.Decode(payload) if err != nil { return nil } - p.lastDTMFTimestamp.Store(h.Timestamp) + // RFC 4733 requires all packets of a given digit to share identical timestamps. + // Some SIP devices or carriers may reuse the timestamp of the previous digit + // for the next one, so we combine timestamp and event code for deduplication. + // The marker bit could be used instead, but it is prone to occasional loss. + eventID := uint64(h.Timestamp)<<8 | uint64(ev.Code) + if eventID == p.lastDTMFEvent.Load() { + return nil + } + p.lastDTMFEvent.Store(eventID) fnc(ev) return nil } diff --git a/pkg/sip/media_port.go b/pkg/sip/media_port.go index f854de96..f9a4d0f7 100644 --- a/pkg/sip/media_port.go +++ b/pkg/sip/media_port.go @@ -410,7 +410,7 @@ func NewMediaPortWith(tid traceid.ID, log logger.Logger, mon *stats.CallMonitor, audioIn: msdk.NewSwitchWriter(inSampleRate), stats: opts.Stats, } - p.lastDTMFTimestamp.Store(math.MaxUint32) + p.lastDTMFEvent.Store(math.MaxUint64) if p.opts.IgnorePreanswerData { p.port.startDiscarding() } @@ -458,7 +458,7 @@ type MediaPort struct { audioIn *msdk.SwitchWriter // SIP RTP -> LK PCM audioInHandler rtp.Handler // for debug only dtmfIn atomic.Pointer[func(ev dtmf.Event)] - lastDTMFTimestamp atomic.Uint32 // rtp timestamp of last DTMF packet seen + lastDTMFEvent atomic.Uint64 // composite (timestamp, event code) of last DTMF packet seen } func (p *MediaPort) DisableOut() { diff --git a/pkg/sip/media_port_test.go b/pkg/sip/media_port_test.go index 64b693b4..bc122581 100644 --- a/pkg/sip/media_port_test.go +++ b/pkg/sip/media_port_test.go @@ -872,7 +872,7 @@ func TestMediaPortDTMF(t *testing.T) { for _, lossPackets := range lossCases { t.Run(fmt.Sprintf("digits=%s/loss=%s", digits, lossPackets), func(t *testing.T) { p := &MediaPort{} - p.lastDTMFTimestamp.Store(math.MaxUint32) + p.lastDTMFEvent.Store(math.MaxUint64) got := "" p.HandleDTMF(func(ev dtmf.Event) { t.Logf("received DTMF event: %+v", ev) @@ -895,6 +895,40 @@ func TestMediaPortDTMF(t *testing.T) { } } +func TestMediaPortDTMFSameTimestamp(t *testing.T) { + // Some SIP carriers reuse the RTP timestamp across different DTMF digits. + // Verify that each distinct event code is still reported even when timestamps are shared. + p := &MediaPort{} + p.lastDTMFEvent.Store(math.MaxUint64) + var codes []byte + p.HandleDTMF(func(ev dtmf.Event) { + codes = append(codes, ev.Code) + }) + + // Three different digits, all with the same timestamp (non-standard but seen in production). + sameTS := uint32(100000) + // Use digit characters so we go through dtmf.Write which produces proper DTMF packets. + digits := "123" + + for i := range digits { + var buf rtp.Buffer + w := rtp.NewSeqWriter(&buf).NewStream(101, dtmf.SampleRate) + err := dtmf.Write(context.Background(), nil, w, sameTS, digits[i:i+1]) + require.NoError(t, err) + require.NotEmpty(t, buf) + // Take the first packet of each digit and send to handler. + // All packets share the same timestamp, simulating the bug scenario. + pkt := buf[0] + h := pkt.Header + require.NoError(t, p.dtmfHandler(&h, pkt.Payload)) + } + + require.Equal(t, 3, len(codes)) + require.NotEqual(t, codes[0], codes[1]) + require.NotEqual(t, codes[1], codes[2]) + require.NotEqual(t, codes[0], codes[2]) +} + // Test util for incrementing prometheus counter metrics. func gatherCounter(t testing.TB, name string, labels map[string]string) float64 { t.Helper()