From 3c4790fc739f34bba086beb19b5f2a924f06df62 Mon Sep 17 00:00:00 2001 From: Debasis Das Date: Thu, 17 Sep 2026 15:22:29 -0700 Subject: [PATCH 1/3] dialog_client: clone ACK per 2xx retransmission (fix PrependHeader panic) WriteAck cloned the ACK once and re-sent that same *sip.Request on every 200 OK retransmission. Each resend runs WriteRequest -> ClientRequestAddVia -> Request.PrependHeader, which mutates the request's header slice in place. Each retransmitted 2xx is delivered on its own transaction-layer goroutine (passUpRetransmission releases the FSM lock before invoking the hook), so two close retransmissions race on the one shared request and panic in PrependHeader: panic: runtime error: index out of range [N] with length N sip.(*headers).PrependHeader ClientRequestAddVia DialogClientSession.WriteAck.func1 ClientTx.passUpRetransmission The panic is raised in the transaction-layer goroutine, so callers cannot recover it and the process crashes. Reusing the clone also appends a duplicate Via on every resend. Fix: snapshot the ACK once and send a fresh clone per retransmission. Concurrent Clone() calls are concurrent reads of the immutable snapshot (safe); each resend mutates its own copy. Adds TestDialogClientACKRetransmissionOverlap, which overlaps the retransmission callbacks and asserts one Via per ACK. It fails under -race on the shared request and passes with the fix; the existing sequential test does not catch it. --- dialog_client.go | 10 ++++++-- dialog_client_test.go | 57 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 2 deletions(-) diff --git a/dialog_client.go b/dialog_client.go index da03441f..b47efb12 100644 --- a/dialog_client.go +++ b/dialog_client.go @@ -402,14 +402,20 @@ func (s *DialogClientSession) WriteAck(ctx context.Context, ack *sip.Request) er // request is passed to the transport layer directly for transmission, // rather than a client transaction. This is because the UAC core // handles retransmissions of the ACK, not the transaction layer. - retransmissionAck := ack.Clone() // We need to clone for RACE safety + // Snapshot once and never mutate it. Each retransmission sends a FRESH clone: + // WriteRequest -> ClientRequestAddVia mutates the request's header list in place + // (PrependHeader), which is not safe to share across retransmission handlers. + // Each retransmitted 2xx is delivered on its own transaction-layer goroutine, so + // two close retransmissions racing on one shared request panic in PrependHeader + // ("index out of range"); reusing it also appends a duplicate Via per resend. + ackTemplate := ack.Clone() s.inviteTx.OnRetransmission(func(r *sip.Response) { // Detect retransmission if r.StatusCode != 200 { return } - if err := s.WriteRequest(retransmissionAck); err != nil { + if err := s.WriteRequest(ackTemplate.Clone()); err != nil { s.endWithCause(fmt.Errorf("ACK retransmission failed: %w", err)) } }) diff --git a/dialog_client_test.go b/dialog_client_test.go index 6fc6d52b..4e010c1f 100644 --- a/dialog_client_test.go +++ b/dialog_client_test.go @@ -3,6 +3,7 @@ package sipgo import ( "context" "log/slog" + "sync" "sync/atomic" "testing" "time" @@ -305,6 +306,62 @@ func TestDialogClientACKRetransmission(t *testing.T) { assert.EqualValues(t, 3, atomic.LoadInt32(&acks)) } +// TestDialogClientACKRetransmissionOverlap fires several 200 OK retransmissions at +// once so their ACK-resend callbacks run concurrently. If WriteAck reuses one shared +// ACK across retransmissions, the concurrent WriteRequest -> ClientRequestAddVia -> +// Request.PrependHeader races (panics with "index out of range" under -race) and can +// emit an ACK carrying more than one Via. Cloning the ACK per retransmission keeps +// each resend independent: no shared mutation, and exactly one Via per ACK. +// +// The sequential TestDialogClientACKRetransmission does not catch this: it sends +// replies one at a time and only counts ACKs, so it passes with the shared request. +func TestDialogClientACKRetransmissionOverlap(t *testing.T) { + const retransmits = 8 + var mu sync.Mutex + var viaCounts []int + answered := make(chan struct{}) + + client := testClientResponder(t, func(req *sip.Request, w *siptest.ClientTxResponder) { + if req.IsAck() { + mu.Lock() + viaCounts = append(viaCounts, len(req.GetHeaders("Via"))) + mu.Unlock() + return + } + res := sip.NewResponseFromRequest(req, 200, "OK", nil) + w.Receive(res) // initial answer -> WaitAnswer returns, caller ACKs + <-answered // wait until OnRetransmission is registered (after Ack) + // Fire the retransmissions simultaneously so the ACK-resend callbacks overlap. + var wg sync.WaitGroup + start := make(chan struct{}) + for i := 0; i < retransmits; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + w.Receive(res) + }() + } + close(start) + wg.Wait() + }) + + dua := DialogUA{Client: client} + d, err := dua.Invite(context.TODO(), sip.Uri{User: "test", Host: "localhost"}, nil) + require.NoError(t, err) + require.NoError(t, d.WaitAnswer(context.TODO(), AnswerOptions{})) + require.NoError(t, d.Ack(context.TODO())) + close(answered) // release the overlapping retransmissions + time.Sleep(4 * sip.T1) + + mu.Lock() + defer mu.Unlock() + require.GreaterOrEqual(t, len(viaCounts), 2, "want the initial ACK plus retransmission ACKs") + for i, n := range viaCounts { + assert.Equalf(t, 1, n, "ACK #%d has %d Via headers, want exactly 1", i, n) + } +} + func BenchmarkDialogDo(b *testing.B) { ua, _ := NewUA() cli, _ := NewClient(ua) From 9d5f4daa64ad2d8fa35b85e6a5487cfd977162d5 Mon Sep 17 00:00:00 2001 From: Debasis Das Date: Wed, 30 Sep 2026 16:54:07 -0700 Subject: [PATCH 2/3] Preserve dialog routing when replaying ACKs --- dialog_ack_route_test.go | 146 +++++++++++++++++++++++++++++++++++++++ dialog_ack_tcp_test.go | 103 +++++++++++++++++++++++++++ dialog_client.go | 16 ++--- 3 files changed, 257 insertions(+), 8 deletions(-) create mode 100644 dialog_ack_route_test.go create mode 100644 dialog_ack_tcp_test.go diff --git a/dialog_ack_route_test.go b/dialog_ack_route_test.go new file mode 100644 index 00000000..a309f89d --- /dev/null +++ b/dialog_ack_route_test.go @@ -0,0 +1,146 @@ +package sipgo + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/emiago/sipgo/sip" + "github.com/stretchr/testify/require" +) + +type ackReplayTransaction struct { + sip.ClientTransaction + replay sip.FnTxResponse +} + +func (tx *ackReplayTransaction) OnRetransmission(f sip.FnTxResponse) bool { + tx.replay = f + return true +} + +type ackRequestWriter func(*sip.Request) error + +func (w ackRequestWriter) Request(_ context.Context, req *sip.Request) (sip.ClientTransaction, error) { + return nil, w(req) +} + +func TestDialogClientACKRetransmissionRouting(t *testing.T) { + for _, tc := range []struct { + name string + transport string + routes []string + destination string + rewrite bool + want string + }{ + {name: "direct", transport: "TCP", want: "192.0.2.20:5060"}, + {name: "loose TCP", transport: "TCP", routes: []string{"", ""}, want: "192.0.2.1:5060"}, + {name: "loose UDP", transport: "UDP", routes: []string{"", ""}, want: "192.0.2.1:5060"}, + {name: "strict", transport: "TCP", routes: []string{"", ""}, want: "192.0.2.1:5060"}, + {name: "explicit destination", transport: "TCP", routes: []string{""}, destination: "198.51.100.50:5090", want: "198.51.100.50:5090"}, + {name: "rewritten contact", transport: "TCP", rewrite: true, want: "192.0.2.30:5090"}, + } { + t.Run(tc.name, func(t *testing.T) { + ua, err := NewUA() + require.NoError(t, err) + t.Cleanup(func() { _ = ua.Close() }) + client, err := NewClient(ua) + require.NoError(t, err) + var mu sync.Mutex + var sent []*sip.Request + client.TxRequester = ackRequestWriter(func(req *sip.Request) error { + mu.Lock() + defer mu.Unlock() + sent = append(sent, req) + return nil + }) + invite := sip.NewRequest(sip.INVITE, sip.Uri{User: "user", Host: "192.0.2.20", Port: 5060}) + invite.SetTransport(tc.transport) + require.NoError(t, clientRequestBuildReq(client, invite)) + res := sip.NewResponseFromRequest(invite, 200, "OK", nil) + res.AppendHeader(sip.NewHeader("Contact", "")) + res.SetSource("192.0.2.30:5090") + for _, route := range tc.routes { + res.AppendHeader(sip.NewHeader("Record-Route", route)) + } + tx := &ackReplayTransaction{} + d := &DialogClientSession{ + UA: &DialogUA{Client: client, RewriteContact: tc.rewrite}, + Dialog: Dialog{InviteRequest: invite, InviteResponse: res}, + inviteTx: tx, + } + d.Init() + d.lastCSeqNo.Store(invite.CSeq().SeqNo) + ack := newAckRequestUAC(invite, res, []byte("answer")) + ack.SetDestination(tc.destination) + ack.AppendHeader(sip.NewHeader("X-Test", "preserved")) + require.NoError(t, d.WriteAck(context.Background(), ack)) + require.Len(t, sent, 1) + require.Equal(t, tc.want, sent[0].Destination()) + d.lastCSeqNo.Add(1) + + const retries = 8 + var wg sync.WaitGroup + start := make(chan struct{}) + for i := 0; i < retries; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + tx.replay(res) + }() + } + close(start) + wg.Wait() + require.Len(t, sent, retries+1) + for _, req := range sent { + require.Equal(t, tc.want, req.Destination()) + require.Equal(t, sent[0].Recipient.String(), req.Recipient.String()) + require.Equal(t, invite.CSeq().SeqNo, req.CSeq().SeqNo) + require.Equal(t, sip.ACK, req.CSeq().MethodName) + require.Len(t, req.GetHeaders("Via"), 1) + require.Equal(t, sent[0].Via().Value(), req.Via().Value()) + require.Len(t, req.GetHeaders("Route"), len(tc.routes)) + for i, route := range req.GetHeaders("Route") { + require.Equal(t, tc.routes[len(tc.routes)-1-i], route.Value()) + } + require.Equal(t, "preserved", req.GetHeader("X-Test").Value()) + require.Equal(t, []byte("answer"), req.Body()) + require.Equal(t, tc.transport, req.Transport()) + } + for i := 1; i < len(sent); i++ { + require.NotSame(t, sent[0], sent[i]) + } + }) + } +} + +func TestDialogClientACKRetransmissionWriteFailure(t *testing.T) { + ua, err := NewUA() + require.NoError(t, err) + t.Cleanup(func() { _ = ua.Close() }) + client, err := NewClient(ua) + require.NoError(t, err) + writes := 0 + failure := errors.New("transport failed") + client.TxRequester = ackRequestWriter(func(req *sip.Request) error { + writes++ + if writes > 1 { + return failure + } + return nil + }) + invite := sip.NewRequest(sip.INVITE, sip.Uri{User: "user", Host: "192.0.2.20"}) + require.NoError(t, clientRequestBuildReq(client, invite)) + res := sip.NewResponseFromRequest(invite, 200, "OK", nil) + tx := &ackReplayTransaction{} + d := &DialogClientSession{UA: &DialogUA{Client: client}, Dialog: Dialog{InviteRequest: invite, InviteResponse: res}, inviteTx: tx} + d.Init() + d.lastCSeqNo.Store(invite.CSeq().SeqNo) + require.NoError(t, d.Ack(context.Background())) + tx.replay(res) + require.Equal(t, sip.DialogStateEnded, d.LoadState()) + require.ErrorIs(t, context.Cause(d.Context()), failure) +} diff --git a/dialog_ack_tcp_test.go b/dialog_ack_tcp_test.go new file mode 100644 index 00000000..d5c5e455 --- /dev/null +++ b/dialog_ack_tcp_test.go @@ -0,0 +1,103 @@ +package sipgo + +import ( + "context" + "fmt" + "net" + "testing" + "time" + + "github.com/emiago/sipgo/sip" + "github.com/stretchr/testify/require" +) + +func TestDialogClientACKRetransmissionTCPProxy(t *testing.T) { + for _, auth := range []bool{false, true} { + t.Run(fmt.Sprintf("authenticate=%t", auth), func(t *testing.T) { + proxy, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = proxy.Close() }) + contact, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = contact.Close() }) + serverUA, err := NewUA() + require.NoError(t, err) + t.Cleanup(func() { _ = serverUA.Close() }) + server, err := NewServer(serverUA) + require.NoError(t, err) + answers := make(chan *sip.Response, 1) + acks := make(chan *sip.Request, 8) + server.OnInvite(func(req *sip.Request, tx sip.ServerTransaction) { + if auth && req.GetHeader("Proxy-Authorization") == nil { + res := sip.NewResponseFromRequest(req, 407, "Proxy Authentication Required", nil) + res.AppendHeader(sip.NewHeader("Proxy-Authenticate", `Digest realm="test", nonce="nonce", algorithm=MD5`)) + if err := tx.Respond(res); err != nil { + t.Errorf("challenge: %v", err) + } + return + } + res := sip.NewResponseFromRequest(req, 200, "OK", nil) + res.AppendHeader(sip.NewHeader("Contact", "")) + res.AppendHeader(sip.NewHeader("Record-Route", "")) + if err := tx.Respond(res); err != nil { + t.Errorf("answer: %v", err) + } + answers <- res + }) + server.OnAck(func(req *sip.Request, _ sip.ServerTransaction) { + if req.Recipient.Host == "carrier.invalid" { + return + } + acks <- req + }) + served := make(chan struct{}) + go func() { + defer close(served) + _ = server.ServeTCP(proxy) + }() + t.Cleanup(func() { + _ = proxy.Close() + <-served + }) + clientUA, err := NewUA() + require.NoError(t, err) + t.Cleanup(func() { _ = clientUA.Close() }) + client, err := NewClient(clientUA) + require.NoError(t, err) + dua := DialogUA{Client: client} + req := sip.NewRequest(sip.INVITE, sip.Uri{Scheme: "sip", User: "user", Host: "carrier.invalid"}) + req.SetTransport("TCP") + req.SetDestination(proxy.Addr().String()) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + d, err := dua.WriteInvite(ctx, req) + require.NoError(t, err) + t.Cleanup(func() { + d.inviteTx.Terminate() + _ = d.Close() + }) + require.NoError(t, d.WaitAnswer(ctx, AnswerOptions{Username: "user", Password: "password"})) + require.NoError(t, d.Ack(ctx)) + var answer *sip.Response + select { + case answer = <-answers: + case <-ctx.Done(): + t.Fatal("missing proxy answer") + } + for i := 0; i < 3; i++ { + if i > 0 { + require.NoError(t, server.WriteResponse(answer.Clone())) + } + select { + case ack := <-acks: + require.Equal(t, "sip:"+contact.Addr().String(), ack.Recipient.String()) + require.Equal(t, req.CSeq().SeqNo, ack.CSeq().SeqNo) + require.Len(t, ack.GetHeaders("Route"), 1) + require.Equal(t, proxy.Addr().String(), ack.Route().Address.Host+fmt.Sprintf(":%d", ack.Route().Address.Port)) + case <-ctx.Done(): + t.Fatalf("ACK %d did not reach the proxy", i+1) + } + } + }) + } +} diff --git a/dialog_client.go b/dialog_client.go index b47efb12..bc17a2f6 100644 --- a/dialog_client.go +++ b/dialog_client.go @@ -402,12 +402,12 @@ func (s *DialogClientSession) WriteAck(ctx context.Context, ack *sip.Request) er // request is passed to the transport layer directly for transmission, // rather than a client transaction. This is because the UAC core // handles retransmissions of the ACK, not the transaction layer. - // Snapshot once and never mutate it. Each retransmission sends a FRESH clone: - // WriteRequest -> ClientRequestAddVia mutates the request's header list in place - // (PrependHeader), which is not safe to share across retransmission handlers. - // Each retransmitted 2xx is delivered on its own transaction-layer goroutine, so - // two close retransmissions racing on one shared request panic in PrependHeader - // ("index out of range"); reusing it also appends a duplicate Via per resend. + s.buildReq(ack) + if err := s.requestValidate(s.UA.Client, ack); err != nil { + return err + } + // Snapshot the fully routed ACK once; Clone pins the current destination. + // Replay fresh copies without rebuilding dialog headers to avoid races and header accumulation. ackTemplate := ack.Clone() s.inviteTx.OnRetransmission(func(r *sip.Response) { // Detect retransmission @@ -415,12 +415,12 @@ func (s *DialogClientSession) WriteAck(ctx context.Context, ack *sip.Request) er return } - if err := s.WriteRequest(ackTemplate.Clone()); err != nil { + if err := s.UA.Client.WriteRequest(ackTemplate.Clone(), s.requestValidate); err != nil { s.endWithCause(fmt.Errorf("ACK retransmission failed: %w", err)) } }) - if err := s.WriteRequest(ack); err != nil { + if err := s.UA.Client.WriteRequest(ack, s.requestValidate); err != nil { // Make sure we close our error // s.Close() return err From aa3557f7a50eda1ad3ed4b5d1acacf5d11e52252 Mon Sep 17 00:00:00 2001 From: Debasis Das Date: Thu, 1 Oct 2026 16:05:34 -0700 Subject: [PATCH 3/3] Remove added ACK regression tests as requested --- dialog_ack_route_test.go | 146 --------------------------------------- dialog_ack_tcp_test.go | 103 --------------------------- dialog_client_test.go | 57 --------------- 3 files changed, 306 deletions(-) delete mode 100644 dialog_ack_route_test.go delete mode 100644 dialog_ack_tcp_test.go diff --git a/dialog_ack_route_test.go b/dialog_ack_route_test.go deleted file mode 100644 index a309f89d..00000000 --- a/dialog_ack_route_test.go +++ /dev/null @@ -1,146 +0,0 @@ -package sipgo - -import ( - "context" - "errors" - "sync" - "testing" - - "github.com/emiago/sipgo/sip" - "github.com/stretchr/testify/require" -) - -type ackReplayTransaction struct { - sip.ClientTransaction - replay sip.FnTxResponse -} - -func (tx *ackReplayTransaction) OnRetransmission(f sip.FnTxResponse) bool { - tx.replay = f - return true -} - -type ackRequestWriter func(*sip.Request) error - -func (w ackRequestWriter) Request(_ context.Context, req *sip.Request) (sip.ClientTransaction, error) { - return nil, w(req) -} - -func TestDialogClientACKRetransmissionRouting(t *testing.T) { - for _, tc := range []struct { - name string - transport string - routes []string - destination string - rewrite bool - want string - }{ - {name: "direct", transport: "TCP", want: "192.0.2.20:5060"}, - {name: "loose TCP", transport: "TCP", routes: []string{"", ""}, want: "192.0.2.1:5060"}, - {name: "loose UDP", transport: "UDP", routes: []string{"", ""}, want: "192.0.2.1:5060"}, - {name: "strict", transport: "TCP", routes: []string{"", ""}, want: "192.0.2.1:5060"}, - {name: "explicit destination", transport: "TCP", routes: []string{""}, destination: "198.51.100.50:5090", want: "198.51.100.50:5090"}, - {name: "rewritten contact", transport: "TCP", rewrite: true, want: "192.0.2.30:5090"}, - } { - t.Run(tc.name, func(t *testing.T) { - ua, err := NewUA() - require.NoError(t, err) - t.Cleanup(func() { _ = ua.Close() }) - client, err := NewClient(ua) - require.NoError(t, err) - var mu sync.Mutex - var sent []*sip.Request - client.TxRequester = ackRequestWriter(func(req *sip.Request) error { - mu.Lock() - defer mu.Unlock() - sent = append(sent, req) - return nil - }) - invite := sip.NewRequest(sip.INVITE, sip.Uri{User: "user", Host: "192.0.2.20", Port: 5060}) - invite.SetTransport(tc.transport) - require.NoError(t, clientRequestBuildReq(client, invite)) - res := sip.NewResponseFromRequest(invite, 200, "OK", nil) - res.AppendHeader(sip.NewHeader("Contact", "")) - res.SetSource("192.0.2.30:5090") - for _, route := range tc.routes { - res.AppendHeader(sip.NewHeader("Record-Route", route)) - } - tx := &ackReplayTransaction{} - d := &DialogClientSession{ - UA: &DialogUA{Client: client, RewriteContact: tc.rewrite}, - Dialog: Dialog{InviteRequest: invite, InviteResponse: res}, - inviteTx: tx, - } - d.Init() - d.lastCSeqNo.Store(invite.CSeq().SeqNo) - ack := newAckRequestUAC(invite, res, []byte("answer")) - ack.SetDestination(tc.destination) - ack.AppendHeader(sip.NewHeader("X-Test", "preserved")) - require.NoError(t, d.WriteAck(context.Background(), ack)) - require.Len(t, sent, 1) - require.Equal(t, tc.want, sent[0].Destination()) - d.lastCSeqNo.Add(1) - - const retries = 8 - var wg sync.WaitGroup - start := make(chan struct{}) - for i := 0; i < retries; i++ { - wg.Add(1) - go func() { - defer wg.Done() - <-start - tx.replay(res) - }() - } - close(start) - wg.Wait() - require.Len(t, sent, retries+1) - for _, req := range sent { - require.Equal(t, tc.want, req.Destination()) - require.Equal(t, sent[0].Recipient.String(), req.Recipient.String()) - require.Equal(t, invite.CSeq().SeqNo, req.CSeq().SeqNo) - require.Equal(t, sip.ACK, req.CSeq().MethodName) - require.Len(t, req.GetHeaders("Via"), 1) - require.Equal(t, sent[0].Via().Value(), req.Via().Value()) - require.Len(t, req.GetHeaders("Route"), len(tc.routes)) - for i, route := range req.GetHeaders("Route") { - require.Equal(t, tc.routes[len(tc.routes)-1-i], route.Value()) - } - require.Equal(t, "preserved", req.GetHeader("X-Test").Value()) - require.Equal(t, []byte("answer"), req.Body()) - require.Equal(t, tc.transport, req.Transport()) - } - for i := 1; i < len(sent); i++ { - require.NotSame(t, sent[0], sent[i]) - } - }) - } -} - -func TestDialogClientACKRetransmissionWriteFailure(t *testing.T) { - ua, err := NewUA() - require.NoError(t, err) - t.Cleanup(func() { _ = ua.Close() }) - client, err := NewClient(ua) - require.NoError(t, err) - writes := 0 - failure := errors.New("transport failed") - client.TxRequester = ackRequestWriter(func(req *sip.Request) error { - writes++ - if writes > 1 { - return failure - } - return nil - }) - invite := sip.NewRequest(sip.INVITE, sip.Uri{User: "user", Host: "192.0.2.20"}) - require.NoError(t, clientRequestBuildReq(client, invite)) - res := sip.NewResponseFromRequest(invite, 200, "OK", nil) - tx := &ackReplayTransaction{} - d := &DialogClientSession{UA: &DialogUA{Client: client}, Dialog: Dialog{InviteRequest: invite, InviteResponse: res}, inviteTx: tx} - d.Init() - d.lastCSeqNo.Store(invite.CSeq().SeqNo) - require.NoError(t, d.Ack(context.Background())) - tx.replay(res) - require.Equal(t, sip.DialogStateEnded, d.LoadState()) - require.ErrorIs(t, context.Cause(d.Context()), failure) -} diff --git a/dialog_ack_tcp_test.go b/dialog_ack_tcp_test.go deleted file mode 100644 index d5c5e455..00000000 --- a/dialog_ack_tcp_test.go +++ /dev/null @@ -1,103 +0,0 @@ -package sipgo - -import ( - "context" - "fmt" - "net" - "testing" - "time" - - "github.com/emiago/sipgo/sip" - "github.com/stretchr/testify/require" -) - -func TestDialogClientACKRetransmissionTCPProxy(t *testing.T) { - for _, auth := range []bool{false, true} { - t.Run(fmt.Sprintf("authenticate=%t", auth), func(t *testing.T) { - proxy, err := net.Listen("tcp", "127.0.0.1:0") - require.NoError(t, err) - t.Cleanup(func() { _ = proxy.Close() }) - contact, err := net.Listen("tcp", "127.0.0.1:0") - require.NoError(t, err) - t.Cleanup(func() { _ = contact.Close() }) - serverUA, err := NewUA() - require.NoError(t, err) - t.Cleanup(func() { _ = serverUA.Close() }) - server, err := NewServer(serverUA) - require.NoError(t, err) - answers := make(chan *sip.Response, 1) - acks := make(chan *sip.Request, 8) - server.OnInvite(func(req *sip.Request, tx sip.ServerTransaction) { - if auth && req.GetHeader("Proxy-Authorization") == nil { - res := sip.NewResponseFromRequest(req, 407, "Proxy Authentication Required", nil) - res.AppendHeader(sip.NewHeader("Proxy-Authenticate", `Digest realm="test", nonce="nonce", algorithm=MD5`)) - if err := tx.Respond(res); err != nil { - t.Errorf("challenge: %v", err) - } - return - } - res := sip.NewResponseFromRequest(req, 200, "OK", nil) - res.AppendHeader(sip.NewHeader("Contact", "")) - res.AppendHeader(sip.NewHeader("Record-Route", "")) - if err := tx.Respond(res); err != nil { - t.Errorf("answer: %v", err) - } - answers <- res - }) - server.OnAck(func(req *sip.Request, _ sip.ServerTransaction) { - if req.Recipient.Host == "carrier.invalid" { - return - } - acks <- req - }) - served := make(chan struct{}) - go func() { - defer close(served) - _ = server.ServeTCP(proxy) - }() - t.Cleanup(func() { - _ = proxy.Close() - <-served - }) - clientUA, err := NewUA() - require.NoError(t, err) - t.Cleanup(func() { _ = clientUA.Close() }) - client, err := NewClient(clientUA) - require.NoError(t, err) - dua := DialogUA{Client: client} - req := sip.NewRequest(sip.INVITE, sip.Uri{Scheme: "sip", User: "user", Host: "carrier.invalid"}) - req.SetTransport("TCP") - req.SetDestination(proxy.Addr().String()) - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - d, err := dua.WriteInvite(ctx, req) - require.NoError(t, err) - t.Cleanup(func() { - d.inviteTx.Terminate() - _ = d.Close() - }) - require.NoError(t, d.WaitAnswer(ctx, AnswerOptions{Username: "user", Password: "password"})) - require.NoError(t, d.Ack(ctx)) - var answer *sip.Response - select { - case answer = <-answers: - case <-ctx.Done(): - t.Fatal("missing proxy answer") - } - for i := 0; i < 3; i++ { - if i > 0 { - require.NoError(t, server.WriteResponse(answer.Clone())) - } - select { - case ack := <-acks: - require.Equal(t, "sip:"+contact.Addr().String(), ack.Recipient.String()) - require.Equal(t, req.CSeq().SeqNo, ack.CSeq().SeqNo) - require.Len(t, ack.GetHeaders("Route"), 1) - require.Equal(t, proxy.Addr().String(), ack.Route().Address.Host+fmt.Sprintf(":%d", ack.Route().Address.Port)) - case <-ctx.Done(): - t.Fatalf("ACK %d did not reach the proxy", i+1) - } - } - }) - } -} diff --git a/dialog_client_test.go b/dialog_client_test.go index 4e010c1f..6fc6d52b 100644 --- a/dialog_client_test.go +++ b/dialog_client_test.go @@ -3,7 +3,6 @@ package sipgo import ( "context" "log/slog" - "sync" "sync/atomic" "testing" "time" @@ -306,62 +305,6 @@ func TestDialogClientACKRetransmission(t *testing.T) { assert.EqualValues(t, 3, atomic.LoadInt32(&acks)) } -// TestDialogClientACKRetransmissionOverlap fires several 200 OK retransmissions at -// once so their ACK-resend callbacks run concurrently. If WriteAck reuses one shared -// ACK across retransmissions, the concurrent WriteRequest -> ClientRequestAddVia -> -// Request.PrependHeader races (panics with "index out of range" under -race) and can -// emit an ACK carrying more than one Via. Cloning the ACK per retransmission keeps -// each resend independent: no shared mutation, and exactly one Via per ACK. -// -// The sequential TestDialogClientACKRetransmission does not catch this: it sends -// replies one at a time and only counts ACKs, so it passes with the shared request. -func TestDialogClientACKRetransmissionOverlap(t *testing.T) { - const retransmits = 8 - var mu sync.Mutex - var viaCounts []int - answered := make(chan struct{}) - - client := testClientResponder(t, func(req *sip.Request, w *siptest.ClientTxResponder) { - if req.IsAck() { - mu.Lock() - viaCounts = append(viaCounts, len(req.GetHeaders("Via"))) - mu.Unlock() - return - } - res := sip.NewResponseFromRequest(req, 200, "OK", nil) - w.Receive(res) // initial answer -> WaitAnswer returns, caller ACKs - <-answered // wait until OnRetransmission is registered (after Ack) - // Fire the retransmissions simultaneously so the ACK-resend callbacks overlap. - var wg sync.WaitGroup - start := make(chan struct{}) - for i := 0; i < retransmits; i++ { - wg.Add(1) - go func() { - defer wg.Done() - <-start - w.Receive(res) - }() - } - close(start) - wg.Wait() - }) - - dua := DialogUA{Client: client} - d, err := dua.Invite(context.TODO(), sip.Uri{User: "test", Host: "localhost"}, nil) - require.NoError(t, err) - require.NoError(t, d.WaitAnswer(context.TODO(), AnswerOptions{})) - require.NoError(t, d.Ack(context.TODO())) - close(answered) // release the overlapping retransmissions - time.Sleep(4 * sip.T1) - - mu.Lock() - defer mu.Unlock() - require.GreaterOrEqual(t, len(viaCounts), 2, "want the initial ACK plus retransmission ACKs") - for i, n := range viaCounts { - assert.Equalf(t, 1, n, "ACK #%d has %d Via headers, want exactly 1", i, n) - } -} - func BenchmarkDialogDo(b *testing.B) { ua, _ := NewUA() cli, _ := NewClient(ua)