quic-go backend: fix dangling api calls

Project-Faster · May 14, 2024 · 1e1bcb0 · 1e1bcb0
1 parent 4f23b6f
commit 1e1bcb0
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 19 deletions.
diff --git a/backend/backend_quicgo.go b/backend/backend_quicgo.go
@@ -106,6 +106,8 @@ type qgoConnectionAdapter struct {
 	context    context.Context
 	listener   quic.Listener
 	connection quic.Connection
+
+	streams []quic.Stream
 }
 
 func (c *qgoConnectionAdapter) LocalAddr() net.Addr {
@@ -128,9 +130,29 @@ func (c *qgoConnectionAdapter) RemoteAddr() net.Addr {
 	panic(shared.ErrInvalidBackendOperation)
 }
 
+func (c *qgoConnectionAdapter) AcceptConnection(ctx context.Context) (QuicBackendConnection, error) {
+	if c.listener != nil {
+		conn, err := c.listener.Accept(ctx)
+		if err != nil {
+			return nil, err
+		}
+		cNew := &qgoConnectionAdapter{
+			context:    ctx,
+			listener:   c.listener,
+			connection: conn,
+			streams:    make([]quic.Stream, 0, 32),
+		}
+		return cNew, nil
+	}
+	panic(shared.ErrInvalidBackendOperation)
+}
+
 func (c *qgoConnectionAdapter) AcceptStream(ctx context.Context) (QuicBackendStream, error) {
 	if c.connection != nil {
 		stream, err := c.connection.AcceptStream(ctx)
+		if stream != nil {
+			c.streams = append(c.streams, stream)
+		}
 		return &qgoStreamAdapter{
 			Stream: stream,
 		}, err
@@ -148,28 +170,18 @@ func (c *qgoConnectionAdapter) OpenStream(ctx context.Context) (QuicBackendStrea
 	panic(shared.ErrInvalidBackendOperation)
 }
 
-func (c *qgoConnectionAdapter) AcceptConnection(ctx context.Context) (QuicBackendConnection, error) {
-	if c.listener != nil {
-		conn, err := c.listener.Accept(ctx)
-		if err != nil {
-			return nil, err
-		}
-		cNew := &qgoConnectionAdapter{
-			context:    ctx,
-			listener:   c.listener,
-			connection: conn,
-		}
-		return cNew, nil
-	}
-	panic(shared.ErrInvalidBackendOperation)
-}
-
 func (c *qgoConnectionAdapter) Close(code int, message string) error {
 	defer func() {
 		c.connection = nil
 		c.listener = nil
+		c.streams = nil
 	}()
 	if c.connection != nil {
+		for _, st := range c.streams {
+			st.CancelRead(quic.StreamErrorCode(0))
+			st.CancelWrite(quic.StreamErrorCode(0))
+			_ = st.Close()
+		}
 		return c.connection.CloseWithError(quic.ApplicationErrorCode(code), message)
 	}
 	if c.listener != nil {
@@ -194,10 +206,12 @@ type qgoStreamAdapter struct {
 }
 
 func (stream *qgoStreamAdapter) AbortRead(code uint64) {
+	stream.CancelRead(quic.StreamErrorCode(code))
 	stream.closedRead = true
 }
 
 func (stream *qgoStreamAdapter) AbortWrite(code uint64) {
+	stream.CancelWrite(quic.StreamErrorCode(code))
 	stream.closedWrite = true
 }
 

diff --git a/workers/server/server_network.go b/workers/server/server_network.go
@@ -96,9 +96,8 @@ func listenQuicConn(quicSession backend.QuicBackendConnection) {
 	for {
 		stream, err := quicSession.AcceptStream(context.Background())
 		if err != nil {
-			if err.Error() != "NO_ERROR: No recent network activity" {
-				logger.Error("Unrecoverable error while accepting QUIC stream: %s\n", err)
-			}
+			logger.Error("Unrecoverable error while accepting QUIC stream: %s\n", err)
+			quicSession.Close(0, "")
 			return
 		}
 		go func(st backend.QuicBackendStream) {