Merge pull request #2793 from OffchainLabs/feed-handshake

remove gobwas/ws handshake extensions race condition workaround

broadcastclient/broadcastclient.go

@@ -130,9 +130,10 @@ type BroadcastClient struct {
 
 	chainId uint64
 
-	// Protects conn and shuttingDown
-	connMutex sync.Mutex
-	conn      net.Conn
+	// Protects conn, shuttingDown and compression
+	connMutex   sync.Mutex
+	conn        net.Conn
+	compression bool
 
 	retryCount atomic.Int64
 
@@ -299,7 +300,7 @@ func (bc *BroadcastClient) connect(ctx context.Context, nextSeqNum arbutil.Messa
 		return nil, nil
 	}
 
-	conn, br, _, err := timeoutDialer.Dial(ctx, bc.websocketUrl)
+	conn, br, hs, err := timeoutDialer.Dial(ctx, bc.websocketUrl)
 	if errors.Is(err, ErrIncorrectFeedServerVersion) || errors.Is(err, ErrIncorrectChainId) {
 		return nil, err
 	}
@@ -325,6 +326,24 @@ func (bc *BroadcastClient) connect(ctx context.Context, nextSeqNum arbutil.Messa
 		return nil, ErrMissingFeedServerVersion
 	}
 
+	compressionNegotiated := false
+	for _, ext := range hs.Extensions {
+		if ext.Equal(deflateExt) {
+			compressionNegotiated = true
+			break
+		}
+	}
+	if !compressionNegotiated && config.EnableCompression {
+		log.Warn("Compression was not negotiated when connecting to feed server.")
+	}
+	if compressionNegotiated && !config.EnableCompression {
+		err := conn.Close()
+		if err != nil {
+			return nil, fmt.Errorf("error closing connection when negotiated disabled extension: %w", err)
+		}
+		return nil, errors.New("error dialing feed server: negotiated compression ws extension, but it is disabled")
+	}
+
 	var earlyFrameData io.Reader
 	if br != nil {
 		// Depending on how long the client takes to read the response, there may be
@@ -339,6 +358,7 @@ func (bc *BroadcastClient) connect(ctx context.Context, nextSeqNum arbutil.Messa
 
 	bc.connMutex.Lock()
 	bc.conn = conn
+	bc.compression = compressionNegotiated
 	bc.connMutex.Unlock()
 	log.Info("Feed connected", "feedServerVersion", feedServerVersion, "chainId", chainId, "requestedSeqNum", nextSeqNum)
 
@@ -362,7 +382,7 @@ func (bc *BroadcastClient) startBackgroundReader(earlyFrameData io.Reader) {
 			var op ws.OpCode
 			var err error
 			config := bc.config()
-			msg, op, err = wsbroadcastserver.ReadData(ctx, bc.conn, earlyFrameData, config.Timeout, ws.StateClientSide, config.EnableCompression, flateReader)
+			msg, op, err = wsbroadcastserver.ReadData(ctx, bc.conn, earlyFrameData, config.Timeout, ws.StateClientSide, bc.compression, flateReader)
 			if err != nil {
 				if bc.isShuttingDown() {
 					return

broadcastclient/broadcastclient_test.go

@@ -30,43 +30,30 @@ import (
 	"github.com/offchainlabs/nitro/wsbroadcastserver"
 )
 
-func TestReceiveMessagesWithoutCompression(t *testing.T) {
+func TestReceiveMessages(t *testing.T) {
 	t.Parallel()
-	testReceiveMessages(t, false, false, false, false)
-}
-
-func TestReceiveMessagesWithCompression(t *testing.T) {
-	t.Parallel()
-	testReceiveMessages(t, true, true, false, false)
-}
-
-func TestReceiveMessagesWithServerOptionalCompression(t *testing.T) {
-	t.Parallel()
-	testReceiveMessages(t, true, true, false, false)
-}
-
-func TestReceiveMessagesWithServerOnlyCompression(t *testing.T) {
-	t.Parallel()
-	testReceiveMessages(t, false, true, false, false)
-}
-
-func TestReceiveMessagesWithClientOnlyCompression(t *testing.T) {
-	t.Parallel()
-	testReceiveMessages(t, true, false, false, false)
-}
-
-func TestReceiveMessagesWithRequiredCompression(t *testing.T) {
-	t.Parallel()
-	testReceiveMessages(t, true, true, true, false)
-}
-
-func TestReceiveMessagesWithRequiredCompressionButClientDisabled(t *testing.T) {
-	t.Parallel()
-	testReceiveMessages(t, false, true, true, true)
+	t.Run("withoutCompression", func(t *testing.T) {
+		testReceiveMessages(t, false, false, false, false)
+	})
+	t.Run("withServerOptionalCompression", func(t *testing.T) {
+		testReceiveMessages(t, true, true, false, false)
+	})
+	t.Run("withServerOnlyCompression", func(t *testing.T) {
+		testReceiveMessages(t, false, true, false, false)
+	})
+	t.Run("withClientOnlyCompression", func(t *testing.T) {
+		testReceiveMessages(t, true, false, false, false)
+	})
+	t.Run("withRequiredCompression", func(t *testing.T) {
+		testReceiveMessages(t, true, true, true, false)
+	})
+	t.Run("withRequiredCompressionButClientDisabled", func(t *testing.T) {
+		testReceiveMessages(t, false, true, true, true)
+	})
 }
 
 func testReceiveMessages(t *testing.T, clientCompression bool, serverCompression bool, serverRequire bool, expectNoMessagesReceived bool) {
-	t.Helper()
+	t.Parallel()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 

wsbroadcastserver/utils.go

@@ -137,7 +137,7 @@ func ReadData(ctx context.Context, conn net.Conn, earlyFrameData io.Reader, time
 		var data []byte
 		if msg.IsCompressed() {
 			if !compression {
-				return nil, 0, errors.New("Received compressed frame even though compression is disabled")
+				return nil, 0, errors.New("Received compressed frame even though compression extension wasn't negotiated")
 			}
 			flateReader.Reset(&reader)
 			data, err = io.ReadAll(flateReader)