Relay: a stream leg accepted as the connect wait ends is used, not left open

CHANGELOG.md

@@ -29,6 +29,7 @@ All notable changes to this project are documented here. The format follows
 - **Docs and tooling:** CI compiles the C client (`client/psnr.c`, `client/examples/minimal.c`) with MSVC on Windows and does not run it there; the build instructions in `docs/integrating.md` give the MSVC command.
 
 ### Fixed
+- **Server:** a relay stream whose accepting side arrived just as the connecting side's 10 second wait ended no longer leaves the accepting connection open. The accepting leg is now used if it had already claimed the stream, and the connection is closed otherwise; before, the accepting player could be told the stream was ready while the connecting player was told it was not.
 - **Server:** the in-memory score store now keeps boards for at most 1000 titles, next to its existing 1000 boards per title and 10000 players per board. A RECORD_SCORE that would start a board for a 1001st title gets ERROR 6 and is not kept; titles that already have boards, and players already on a board, are unaffected. Before, the number of titles was not limited. The limit is fixed, not a flag. See `docs/api.md`, Leaderboards.
 - **Server:** `-max-conns-per-addr` now counts an IPv6 address as its /64 (an IPv4-mapped address as the IPv4 address), as the UDP limit and the admin failure delay already did. Before, it counted each exact IPv6 address, so connections from different addresses of one network each had a full allowance. The address a player is known by (rooms, peers, logs) is unchanged. Players behind one IPv6 /64, such as one LAN, now share the limit, as players behind one IPv4 address always did; raise `-max-conns-per-addr` if that is too tight.
 - **Server:** the `-http` status page now closes a connection that does not finish its request header within 5 seconds, its request within 10 seconds, does not read its answer within 10 seconds, or sits idle for 30 seconds. Before, the status server had no such limits.

server/relay.go

@@ -118,6 +118,7 @@ func (s *Server) streamLeg(conn net.Conn, t byte, p []byte) (handedOff bool) {
 			return fail(errNotFound)
 		}
 		if ready() != nil {
+			st.accept <- nil
 			return false
 		}
 		st.accept <- conn
@@ -149,18 +150,13 @@ func (s *Server) streamLeg(conn net.Conn, t byte, p []byte) (handedOff bool) {
 	s.flush([]push{{to, msgStreamOffer, w.b}})
 	s.log.Debug("relay stream", "stream", st.id, "from", from.onlineID, "to", to.onlineID, "vport", vport)
 
-	var other net.Conn
-	select {
-	case other = <-st.accept:
-	case <-s.stopped:
-		s.mu.Lock()
-		delete(s.streams, st.id)
-		s.mu.Unlock()
-		return false
-	case <-time.After(streamAcceptTimeout):
-		s.mu.Lock()
-		delete(s.streams, st.id)
-		s.mu.Unlock()
+	other := s.awaitAccept(st, streamAcceptTimeout)
+	if other == nil {
+		select {
+		case <-s.stopped:
+			return false
+		default:
+		}
 		return fail(errNotFound)
 	}
 	if ready() != nil {
@@ -173,6 +169,28 @@ func (s *Server) streamLeg(conn net.Conn, t byte, p []byte) (handedOff bool) {
 	return false
 }
 
+// awaitAccept waits up to d for the target's leg of st. It returns nil if there is none: the server is stopping, the wait ran out,
+// or the accepting side failed. An accepting side that had already taken the stream out of s.streams when the wait ran out is
+// waited for (it always sends a leg or nil), so its connection is used and not left behind.
+func (s *Server) awaitAccept(st *stream, d time.Duration) net.Conn {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case other := <-st.accept:
+		return other
+	case <-s.stopped:
+	case <-timer.C:
+	}
+	s.mu.Lock()
+	_, waiting := s.streams[st.id]
+	delete(s.streams, st.id)
+	s.mu.Unlock()
+	if waiting {
+		return nil
+	}
+	return <-st.accept
+}
+
 // relayWriteTimeout bounds one write of a relayed chunk to a leg: a peer that stops reading cannot hold the leg open.
 const relayWriteTimeout = 30 * time.Second
 

server/relay_accept_test.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"net"
+	"testing"
+	"time"
+)
+
+// A stream's accepting side takes the stream out of s.streams, answers its player and then hands over the connection. If the
+// connecting side's wait ran out in between, that connection must still be used, not left in the channel with nobody to close it.
+
+func TestAwaitAcceptUsesLegClaimedAtTimeout(t *testing.T) {
+	s := NewServer(10, false)
+	st := &stream{id: 7, accept: make(chan net.Conn, 1)}
+	// The stream is no longer in s.streams: the accepting side has claimed it.
+	ca, sa := tcpPair(t)
+	_ = ca
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		st.accept <- sa
+	}()
+	if got := s.awaitAccept(st, time.Millisecond); got != sa {
+		t.Fatalf("awaitAccept returned %v after the wait ran out, want the claimed leg", got)
+	}
+}
+
+func TestAwaitAcceptClaimedThenFailed(t *testing.T) {
+	s := NewServer(10, false)
+	st := &stream{id: 8, accept: make(chan net.Conn, 1)}
+	go func() {
+		time.Sleep(20 * time.Millisecond)
+		st.accept <- nil
+	}()
+	if got := s.awaitAccept(st, time.Millisecond); got != nil {
+		t.Fatalf("awaitAccept returned %v, want nil when the accepting side failed", got)
+	}
+}
+
+func TestAwaitAcceptTimesOutWhenUnclaimed(t *testing.T) {
+	s := NewServer(10, false)
+	st := &stream{id: 9, accept: make(chan net.Conn, 1)}
+	s.streams[st.id] = st
+	start := time.Now()
+	if got := s.awaitAccept(st, 20*time.Millisecond); got != nil {
+		t.Fatalf("awaitAccept returned %v with no accepting side", got)
+	}
+	if time.Since(start) > 2*time.Second {
+		t.Fatal("awaitAccept did not return after its wait")
+	}
+	if s.streams[st.id] != nil {
+		t.Fatal("the unclaimed stream was left in s.streams")
+	}
+}

site/data.json

@@ -1304,6 +1304,7 @@
      "**Docs and tooling:** CI compiles the C client (`client/psnr.c`, `client/examples/minimal.c`) with MSVC on Windows and does not run it there; the build instructions in `docs/integrating.md` give the MSVC command."
     ],
     "Fixed": [
+     "**Server:** a relay stream whose accepting side arrived just as the connecting side's 10 second wait ended no longer leaves the accepting connection open.",
      "**Server:** the in-memory score store now keeps boards for at most 1000 titles, next to its existing 1000 boards per title and 10000 players per board.",
      "**Server:** `-max-conns-per-addr` now counts an IPv6 address as its /64 (an IPv4-mapped address as the IPv4 address), as the UDP limit and the admin failure delay already did.",
      "**Server:** the `-http` status page now closes a connection that does not finish its request header within 5 seconds, its request within 10 seconds, does not read its answer within 10 seconds, or sits idle for 30 ...",
@@ -1334,7 +1335,7 @@
    },
    "counts": {
     "Added": 19,
-    "Fixed": 14,
+    "Fixed": 15,
     "Changed": 10
    }
   },

site/index.html

@@ -363,6 +363,7 @@ <h4>Added</h4>
 </ul>
 <h4>Fixed</h4>
 <ul>
+<li>**Server:** a relay stream whose accepting side arrived just as the connecting side&#x27;s 10 second wait ended no longer leaves the accepting connection open.</li>
 <li>**Server:** the in-memory score store now keeps boards for at most 1000 titles, next to its existing 1000 boards per title and 10000 players per board.</li>
 <li>**Server:** <code>-max-conns-per-addr</code> now counts an IPv6 address as its /64 (an IPv4-mapped address as the IPv4 address), as the UDP limit and the admin failure delay already did.</li>
 <li>**Server:** the <code>-http</code> status page now closes a connection that does not finish its request header within 5 seconds, its request within 10 seconds, does not read its answer within 10 seconds, or sits idle for 30 ...</li>