Add Snell protocol. Refactor MASQUE HTTP/2, Fair Queue. Update XHTTP, OpenVPN, Sudoku, Fallback. Fixes

transport/sudoku/multiplex/session.go

@@ -18,9 +18,12 @@ const (
 )
 
 const (
-	headerSize     = 1 + 4 + 4
-	maxFrameSize   = 256 * 1024
-	maxDataPayload = 32 * 1024
+	headerSize = 1 + 4 + 4
+	// maxQueuedBytesPerStream bounds unread payload retained by a single logical stream.
+	// Backpressure is applied to the demux loop instead of dropping data.
+	maxQueuedBytesPerStream = 4 * 1024 * 1024
+	maxFrameSize            = 256 * 1024
+	maxDataPayload          = 128 * 1024
 )
 
 type acceptEvent struct {
@@ -344,6 +347,8 @@ type stream struct {
 	closeErr error
 	readBuf  []byte
 	queue    [][]byte
+	// queuedBytes includes unread bytes in readBuf and queue.
+	queuedBytes int
 
 	localAddr  net.Addr
 	remoteAddr net.Addr
@@ -362,16 +367,20 @@ func newStream(session *Session, id uint32) *stream {
 
 func (c *stream) enqueue(payload []byte) {
 	c.mu.Lock()
+	for !c.closed && c.queuedBytes+len(payload) > maxQueuedBytesPerStream {
+		c.cond.Wait()
+	}
 	if c.closed {
 		c.mu.Unlock()
 		return
 	}
+	c.queuedBytes += len(payload)
 	if len(c.readBuf) == 0 && len(c.queue) == 0 {
 		c.readBuf = payload
 	} else {
 		c.queue = append(c.queue, payload)
 	}
-	c.cond.Signal()
+	c.cond.Broadcast()
 	c.mu.Unlock()
 }
 
@@ -413,7 +422,11 @@ func (c *stream) Read(p []byte) (int, error) {
 	}
 	if len(c.readBuf) == 0 && len(c.queue) > 0 {
 		c.readBuf = c.queue[0]
+		c.queue[0] = nil
 		c.queue = c.queue[1:]
+		if len(c.queue) == 0 {
+			c.queue = nil
+		}
 	}
 	if len(c.readBuf) == 0 && c.closed {
 		if c.closeErr == nil {
@@ -424,6 +437,14 @@ func (c *stream) Read(p []byte) (int, error) {
 
 	n := copy(p, c.readBuf)
 	c.readBuf = c.readBuf[n:]
+	if len(c.readBuf) == 0 {
+		c.readBuf = nil
+	}
+	c.queuedBytes -= n
+	if c.queuedBytes < 0 {
+		c.queuedBytes = 0
+	}
+	c.cond.Broadcast()
 	return n, nil
 }
 

transport/sudoku/multiplex/session_backpressure_test.go

@@ -0,0 +1,91 @@
+package multiplex
+
+import (
+	"bytes"
+	"crypto/rand"
+	"io"
+	"net"
+	"sync"
+	"testing"
+	"time"
+)
+
+// TestSession_LargeTransferBackpressure verifies that a transfer larger than
+// maxQueuedBytesPerStream completes correctly: the demux loop applies
+// backpressure (cond.Wait) instead of dropping data, and the reader draining
+// the stream wakes the blocked loop without deadlock.
+func TestSession_LargeTransferBackpressure(t *testing.T) {
+	c1, c2 := net.Pipe()
+
+	client, err := NewClientSession(c1)
+	if err != nil {
+		t.Fatalf("client session: %v", err)
+	}
+	server, err := NewServerSession(c2)
+	if err != nil {
+		t.Fatalf("server session: %v", err)
+	}
+	defer client.Close()
+	defer server.Close()
+
+	// Payload bigger than the per-stream backpressure window (4MB).
+	const total = 12 * 1024 * 1024
+	payload := make([]byte, total)
+	if _, err := rand.Read(payload); err != nil {
+		t.Fatalf("rand: %v", err)
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	var writeErr error
+	go func() {
+		defer wg.Done()
+		stream, err := client.OpenStream([]byte("hello"))
+		if err != nil {
+			writeErr = err
+			return
+		}
+		defer stream.Close()
+		if _, err := stream.Write(payload); err != nil {
+			writeErr = err
+			return
+		}
+		_ = stream.(interface{ CloseWrite() error }).CloseWrite()
+	}()
+
+	var got []byte
+	var readErr error
+	go func() {
+		defer wg.Done()
+		stream, openPayload, err := server.AcceptStream()
+		if err != nil {
+			readErr = err
+			return
+		}
+		if string(openPayload) != "hello" {
+			readErr = io.ErrUnexpectedEOF
+			return
+		}
+		got, readErr = io.ReadAll(stream)
+	}()
+
+	done := make(chan struct{})
+	go func() { wg.Wait(); close(done) }()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("transfer deadlocked (backpressure did not release)")
+	}
+
+	if writeErr != nil {
+		t.Fatalf("write: %v", writeErr)
+	}
+	if readErr != nil {
+		t.Fatalf("read: %v", readErr)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatalf("payload mismatch: got %d bytes, want %d", len(got), len(payload))
+	}
+}