fix broken connections in case of high latency

When listening and accepting an incoming connection request, the
response might be received by the peer with some delay due to latency.
This causes the peer to send a second connection request, that is not
detected as duplicate because the first connection request has already
been removed from the map that is used to check for duplicates
(connReqs), so it is treated as a brand new connection request,
breaking the first connection.

This patch fixes the issue by replacing the map that is used to check
for duplicates (connReqs) with a map that keeps track of both
connection requests and accepted connections by their peer ID
(connByPeers), allowing to discard handshakes with an existing peer ID.

Author: aler9

conn_request.go

@@ -264,24 +264,24 @@ func newConnRequest(ln *listener, p packet.Packet) *connRequest {
 		}
 
 		ln.lock.Lock()
-		_, exists := ln.connReqs[cif.SRTSocketId]
-		if !exists {
-			ln.connReqs[cif.SRTSocketId] = req
-		}
-		ln.lock.Unlock()
 
-		// We received a duplicate request: reject silently
+		// Duplicate request: reject silently
+		_, exists := ln.connsByPeer[cif.SRTSocketId]
 		if exists {
+			ln.lock.Unlock()
 			return nil
 		}
 
+		// Already fill connsByPeer for this peer socket ID
+		ln.connsByPeer[cif.SRTSocketId] = nil
+
 		// Already reserve a socketId for this connection
-		ln.lock.Lock()
 		socketId, err := req.generateSocketId()
 		if err == nil {
 			ln.conns[socketId] = nil
 			req.socketId = socketId
 		}
+
 		ln.lock.Unlock()
 
 		// We couldn't create a socketId: reject silently
@@ -350,7 +350,7 @@ func (req *connRequest) Reject(reason RejectionReason) {
 	req.ln.lock.Lock()
 	defer req.ln.lock.Unlock()
 
-	if _, hasReq := req.ln.connReqs[req.peerSocketId]; !hasReq {
+	if cr, hasReq := req.ln.connsByPeer[req.peerSocketId]; !hasReq || cr != nil {
 		return
 	}
 
@@ -367,8 +367,8 @@ func (req *connRequest) Reject(reason RejectionReason) {
 	req.ln.log("handshake:send:cif", func() string { return req.handshake.String() })
 	req.ln.send(p)
 
-	delete(req.ln.connReqs, req.peerSocketId)
 	delete(req.ln.conns, req.socketId)
+	delete(req.ln.connsByPeer, req.peerSocketId)
 }
 
 // generateSocketId generates an SRT SocketID that can be used for this connection
@@ -397,7 +397,7 @@ func (req *connRequest) Accept() (Conn, error) {
 	req.ln.lock.Lock()
 	defer req.ln.lock.Unlock()
 
-	if _, hasReq := req.ln.connReqs[req.peerSocketId]; !hasReq {
+	if cr, hasReq := req.ln.connsByPeer[req.peerSocketId]; !hasReq || cr != nil {
 		return nil, fmt.Errorf("connection already accepted")
 	}
 
@@ -472,7 +472,7 @@ func (req *connRequest) Accept() (Conn, error) {
 	req.ln.send(p)
 
 	req.ln.conns[req.socketId] = conn
-	delete(req.ln.connReqs, req.peerSocketId)
+	req.ln.connsByPeer[req.peerSocketId] = conn
 
 	return conn, nil
 }

connection.go

@@ -191,7 +191,7 @@ type srtConn struct {
 	readBuffer bytes.Buffer
 
 	onSend     func(p packet.Packet)
-	onShutdown func(socketId uint32)
+	onShutdown func(*srtConn)
 
 	tick time.Duration
 
@@ -234,7 +234,7 @@ type srtConnConfig struct {
 	crypto                      crypto.Crypto
 	keyBaseEncryption           packet.PacketEncryption
 	onSend                      func(p packet.Packet)
-	onShutdown                  func(socketId uint32)
+	onShutdown                  func(*srtConn)
 	logger                      Logger
 }
 
@@ -264,7 +264,7 @@ func newSRTConn(config srtConnConfig) *srtConn {
 	}
 
 	if c.onShutdown == nil {
-		c.onShutdown = func(socketId uint32) {}
+		c.onShutdown = func(*srtConn) {}
 	}
 
 	c.nextACKNumber = circular.New(1, packet.MAX_TIMESTAMP)
@@ -1415,7 +1415,7 @@ func (c *srtConn) close() {
 		c.log("connection:close", func() string { return "shutdown" })
 
 		go func() {
-			c.onShutdown(c.socketId)
+			c.onShutdown(c)
 		}()
 	})
 }

dial.go

@@ -513,7 +513,7 @@ func (dl *dialer) handleHandshake(p packet.Packet) {
 			crypto:                      dl.crypto,
 			keyBaseEncryption:           packet.EvenKeyEncrypted,
 			onSend:                      dl.send,
-			onShutdown:                  func(socketId uint32) { dl.Close() },
+			onShutdown:                  func(*srtConn) { dl.Close() },
 			logger:                      dl.config.Logger,
 		})
 

listen.go

@@ -121,10 +121,10 @@ type listener struct {
 
 	config Config
 
-	backlog  chan packet.Packet
-	connReqs map[uint32]*connRequest
-	conns    map[uint32]*srtConn
-	lock     sync.RWMutex
+	backlog     chan packet.Packet
+	conns       map[uint32]*srtConn
+	connsByPeer map[uint32]*srtConn
+	lock        sync.RWMutex
 
 	start time.Time
 
@@ -190,8 +190,8 @@ func Listen(network, address string, config Config) (Listener, error) {
 		return nil, fmt.Errorf("listen: no local address")
 	}
 
-	ln.connReqs = make(map[uint32]*connRequest)
 	ln.conns = make(map[uint32]*srtConn)
+	ln.connsByPeer = make(map[uint32]*srtConn)
 
 	ln.backlog = make(chan packet.Packet, 128)
 
@@ -326,9 +326,10 @@ func (ln *listener) error() error {
 	return ln.doneErr
 }
 
-func (ln *listener) handleShutdown(socketId uint32) {
+func (ln *listener) handleShutdown(c *srtConn) {
 	ln.lock.Lock()
-	delete(ln.conns, socketId)
+	delete(ln.conns, c.socketId)
+	delete(ln.connsByPeer, c.peerSocketId)
 	ln.lock.Unlock()
 }
 

listen_test.go

@@ -597,27 +597,25 @@ func TestListenDiscardRepeatedHandshakes(t *testing.T) {
 	ln, err := Listen("srt", "127.0.0.1:6003", DefaultConfig())
 	require.NoError(t, err)
 
+	singleReqReceived := make(chan struct{})
+
 	listenDone := make(chan struct{})
 	defer func() { <-listenDone }()
 
-	singleReqReceived := make(chan struct{})
+	defer ln.Close()
 
 	go func() {
 		defer close(listenDone)
 
-		var onlyRequest ConnRequest
-
 		for {
 			req, err := ln.Accept2()
 			if err != nil {
 				break
 			}
 
 			close(singleReqReceived)
-			onlyRequest = req
+			defer req.Reject(REJ_CLOSE)
 		}
-
-		onlyRequest.Reject(REJ_CLOSE)
 	}()
 
 	for i := 0; i < 4; i++ {
@@ -701,5 +699,132 @@ func TestListenDiscardRepeatedHandshakes(t *testing.T) {
 	}
 
 	<-singleReqReceived
-	ln.Close()
+}
+
+func TestListenAcceptAndDiscardRepeatedHandshakes(t *testing.T) {
+	ln, err := Listen("srt", "127.0.0.1:6003", DefaultConfig())
+	require.NoError(t, err)
+
+	singleReqAccepted := make(chan struct{})
+
+	listenDone := make(chan struct{})
+	defer func() { <-listenDone }()
+
+	defer ln.Close()
+
+	go func() {
+		defer close(listenDone)
+
+		for {
+			req, err := ln.Accept2()
+			if err != nil {
+				break
+			}
+
+			conn, err := req.Accept()
+			require.NoError(t, err)
+			defer conn.Close()
+
+			close(singleReqAccepted)
+		}
+	}()
+
+	// Client sends initial connection request
+	conn, err := net.Dial("udp", "127.0.0.1:6003")
+	require.NoError(t, err)
+	defer conn.Close()
+
+	// Send induction request
+	p := packet.NewPacket(conn.RemoteAddr())
+	p.Header().IsControlPacket = true
+	p.Header().ControlType = packet.CTRLTYPE_HANDSHAKE
+	p.Header().SubType = 0
+	p.Header().TypeSpecific = 0
+	p.Header().Timestamp = 0
+	p.Header().DestinationSocketId = 0
+	sendcif := &packet.CIFHandshake{
+		IsRequest:                   true,
+		Version:                     4,
+		EncryptionField:             0,
+		ExtensionField:              2,
+		InitialPacketSequenceNumber: circular.New(10000, packet.MAX_SEQUENCENUMBER),
+		MaxTransmissionUnitSize:     MAX_MSS_SIZE,
+		MaxFlowWindowSize:           25600,
+		HandshakeType:               packet.HSTYPE_INDUCTION,
+		SRTSocketId:                 55555,
+		SynCookie:                   0,
+	}
+	sendcif.PeerIP.FromNetAddr(conn.LocalAddr())
+	p.MarshalCIF(sendcif)
+	var buf bytes.Buffer
+	err = p.Marshal(&buf)
+	require.NoError(t, err)
+	_, err = conn.Write(buf.Bytes())
+	require.NoError(t, err)
+
+	// Read induction response
+	inbuf := make([]byte, MAX_MSS_SIZE)
+	n, err := conn.Read(inbuf)
+	require.NoError(t, err)
+	p, err = packet.NewPacketFromData(conn.RemoteAddr(), inbuf[:n])
+	require.NoError(t, err)
+	recvcif := &packet.CIFHandshake{}
+	err = p.UnmarshalCIF(recvcif)
+	require.NoError(t, err)
+
+	// Send conclusion request
+	p.Header().IsControlPacket = true
+	p.Header().ControlType = packet.CTRLTYPE_HANDSHAKE
+	p.Header().SubType = 0
+	p.Header().TypeSpecific = 0
+	p.Header().Timestamp = 0
+	p.Header().DestinationSocketId = 0
+	sendcif.Version = 5
+	sendcif.ExtensionField = recvcif.ExtensionField
+	sendcif.HandshakeType = packet.HSTYPE_CONCLUSION
+	sendcif.SynCookie = recvcif.SynCookie
+	sendcif.HasHS = true
+	sendcif.SRTHS = &packet.CIFHandshakeExtension{
+		SRTVersion: SRT_VERSION,
+		SRTFlags: packet.CIFHandshakeExtensionFlags{
+			TSBPDSND:      true,
+			TSBPDRCV:      true,
+			CRYPT:         true,
+			TLPKTDROP:     true,
+			PERIODICNAK:   true,
+			REXMITFLG:     true,
+			STREAM:        false,
+			PACKET_FILTER: false,
+		},
+		RecvTSBPDDelay: uint16(120),
+		SendTSBPDDelay: uint16(120),
+	}
+	sendcif.HasSID = true
+	sendcif.StreamId = "foobar"
+	p.MarshalCIF(sendcif)
+	buf.Reset()
+	err = p.Marshal(&buf)
+	require.NoError(t, err)
+	_, err = conn.Write(buf.Bytes())
+	require.NoError(t, err)
+
+	// read conclusion response
+	n, err = conn.Read(inbuf)
+	require.NoError(t, err)
+	p, err = packet.NewPacketFromData(conn.RemoteAddr(), inbuf[:n])
+	require.NoError(t, err)
+	recvcif = &packet.CIFHandshake{}
+	err = p.UnmarshalCIF(recvcif)
+	require.NoError(t, err)
+	require.Equal(t, packet.HSTYPE_CONCLUSION, recvcif.HandshakeType)
+	require.False(t, recvcif.IsRequest)
+
+	<-singleReqAccepted
+
+	// send conclusion request, again
+	_, err = conn.Write(buf.Bytes())
+	require.NoError(t, err)
+
+	// wait some time to make sure that close(singleReqAccepted) is not triggered
+	time.Sleep(500 * time.Millisecond)
 }