Repository: anycable/anycable

Issue:
ws_max_pending_size option documented as kB instead of bytes
<!--
  This template is for bug reports. If you are reporting a bug, please continue on.
  If you are here for another reason, please, go to discussions: https://github.com/anycable/anycable/discussions
-->

### Tell us about your environment

**AnyCable version:** 1.6.0

### What did you do?

Run AnyCable with `ws_max_pending_size` parameter ([documented as `(in kB)`](https://github.com/anycable/anycable/blob/ff7a1e2b4fa00aae325fad8135e1a713e6a21627/cli/options.go#L1161))

```
anycable-go --ws_max_pending_size=50000
```

### What did you expect to happen?

Disconnects when queue exceeds ~50MB, otherwise messages are sent as normal.

### What actually happened?

Disconnects when queue exceeds ~50kB, with message: `{"type":"disconnect","reason":"too_slow","reconnect":true}`

<!--
  Please, provide verbose logs for both anycable-go and RPC server (`anycable-go --debug` and `ANYCABLE_DEBUG=1 bin/anycable` respectively)
-->

Discussion:

Solution patch:
diff --git a/cli/options.go b/cli/options.go
index 4222c62..2dee8d6 100644
--- a/cli/options.go
+++ b/cli/options.go
@@ -1158,7 +1158,7 @@ func wsCLIFlags(c *config.Config) []cli.Flag {
 
 		&cli.Uint64Flag{
 			Name:        "ws_max_pending_size",
-			Usage:       "Maximum size of the write queue for session before it's considered slow and disconnected (in kB)",
+			Usage:       "Maximum size (in bytes) of the write queue for a session before it's considered slow and disconnected (0 = unlimited)",
 			Value:       c.WS.MaxPendingSize,
 			Destination: &c.WS.MaxPendingSize,
 		},



Test patch:
diff --git a/node/session_test.go b/node/session_test.go
index f66dfe8..faa6a5b 100644
--- a/node/session_test.go
+++ b/node/session_test.go
@@ -1,7 +1,6 @@
 package node
 
 import (
-	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -223,37 +222,69 @@ func TestMarkDisconnectable(t *testing.T) {
 }
 
 func TestSend__maxPendingSize(t *testing.T) {
-	node := NewMockNode()
-	session := NewMockSession("123", node)
-	session.maxPendingSize = 1
+	sendTestMessages := func(session *Session) {
+		// Lock writing to populate the queue
+		session.wmu.Lock()
 
-	session.Send(&common.Reply{Type: "message", Message: "hello"})
+		// This message is picked up by the writer routine, so it's not part of the queue
+		session.Send(&common.Reply{Type: "message", Message: "test A"})
+		time.Sleep(100 * time.Millisecond)
 
-	msg, err := session.conn.Read()
-	assert.Nil(t, err)
+		// This message stays in the queue, size is 37 bytes (31 boilerplate + 6 content)
+		session.Send(&common.Reply{Type: "message", Message: "test B"})
 
-	assert.Equal(t, `{"type":"message","message":"hello"}`, string(msg))
+		// This message tests the maxPendingSize check
+		session.Send(&common.Reply{Type: "message", Message: "test C"})
+		session.wmu.Unlock()
 
-	// lock writing
-	session.wmu.Lock()
+		// Pop initial non-queued message
+		msg, err := session.conn.Read()
+		assert.Nil(t, err)
+		assert.Equal(t, `{"type":"message","message":"test A"}`, string(msg))
+	}
 
-	session.Send(&common.Reply{Type: "message", Message: strings.Repeat("karavana", 64)})
-	// Make sure the message is picked up the the writer routine
-	time.Sleep(100 * time.Millisecond)
+	t.Run("Disconnect when queue size at least maxPendingSize", func(t *testing.T) {
+		node := NewMockNode()
+		session := NewMockSession("123", node)
+		session.maxPendingSize = 37
 
-	// this one should be dropped
-	session.Send(&common.Reply{Type: "message", Message: strings.Repeat("marivana", 64)})
+		sendTestMessages(session)
 
-	// The queue has already > 1kb of data — we must close the session
-	session.Send(&common.Reply{Type: "message", Message: strings.Repeat("karambas", 64)})
-	session.wmu.Unlock()
+		// Receive slow disconnect notice instead of queued messages
+		msg, err := session.conn.Read()
+		assert.Nil(t, err)
+		assert.Equal(t, `{"type":"disconnect","reason":"too_slow","reconnect":true}`, string(msg))
+	})
 
-	// Read the second message
-	msg, err = session.conn.Read()
-	assert.Nil(t, err)
-	assert.Contains(t, string(msg), "karavana")
-	// must receive only slow disconnect notice
-	msg, err = session.conn.Read()
-	assert.Nil(t, err)
-	assert.Equal(t, `{"type":"disconnect","reason":"too_slow","reconnect":true}`, string(msg))
+	t.Run("Allow when queue size less than maxPendingSize", func(t *testing.T) {
+		node := NewMockNode()
+		session := NewMockSession("123", node)
+		session.maxPendingSize = 38
+
+		sendTestMessages(session)
+
+		// Receive queued messages
+		msg, err := session.conn.Read()
+		assert.Nil(t, err)
+		assert.Equal(t, `{"type":"message","message":"test B"}`, string(msg))
+		msg, err = session.conn.Read()
+		assert.Nil(t, err)
+		assert.Equal(t, `{"type":"message","message":"test C"}`, string(msg))
+	})
+
+	t.Run("Allow when maxPendingSize is 0 (unlimited)", func(t *testing.T) {
+		node := NewMockNode()
+		session := NewMockSession("123", node)
+		session.maxPendingSize = 0
+
+		sendTestMessages(session)
+
+		// Receive queued messages
+		msg, err := session.conn.Read()
+		assert.Nil(t, err)
+		assert.Equal(t, `{"type":"message","message":"test B"}`, string(msg))
+		msg, err = session.conn.Read()
+		assert.Nil(t, err)
+		assert.Equal(t, `{"type":"message","message":"test C"}`, string(msg))
+	})
 }