Bound connection outbound queues

internal/network/conn.go

@@ -4,6 +4,7 @@ package network
 
 import (
 	"bufio"
+	"errors"
 	"io"
 	"net"
 	"sync"
@@ -13,7 +14,18 @@ import (
 	"regionio/internal/server"
 )
 
-const networkWriteTimeout = 30 * time.Second
+const (
+	networkWriteTimeout = 30 * time.Second
+	outboundQueueSize   = 256
+	maxOutboundBytes    = 8 * 1024 * 1024
+)
+
+var ErrOutboundFull = errors.New("network: outbound queue full")
+
+type outboundPacket struct {
+	frame []byte
+	done  chan error
+}
 
 // Conn wraps a TCP connection with buffered reads and tracks protocol state.
 type Conn struct {
@@ -27,19 +39,27 @@ type Conn struct {
 	// Profile is populated once the client identifies during login.
 	Profile server.Profile
 
-	// writeMu serializes writes so a background sender (e.g. keep-alive) and
+	outbound     chan outboundPacket
-	// the read-loop handler never interleave bytes on the wire.
+	done         chan struct{}
-	writeMu sync.Mutex
+	writerDone   chan struct{}
+	closeOnce    sync.Once
+	outboundMu   sync.Mutex
+	pendingBytes int
 }
 
 // NewConn wraps a raw TCP connection.
 func NewConn(raw net.Conn) *Conn {
-	return &Conn{
+	c := &Conn{
 		raw:                  raw,
 		br:                   bufio.NewReaderSize(raw, 4096),
 		state:                protocol.StateHandshaking,
 		compressionThreshold: -1,
+		outbound:             make(chan outboundPacket, outboundQueueSize),
+		done:                 make(chan struct{}),
+		writerDone:           make(chan struct{}),
 	}
+	go c.writeLoop()
+	return c
 }
 
 // State returns the current protocol state.
@@ -68,16 +88,18 @@ func (c *Conn) ReadPacket() (protocol.Packet, error) {
 	return pkt, err
 }
 
-// Send writes a packet with the given ID and pre-encoded body. Safe for
+// Send queues a packet and waits until the connection's sole writer has put it
-// concurrent use.
+// on the wire. Waiting preserves state-machine ordering while keeping all socket
+// writes serialized through the bounded outbound queue.
 func (c *Conn) Send(id int32, body []byte) error {
-	c.writeMu.Lock()
+	return c.sendFrame(protocol.AppendPacket(nil, c.compressionThreshold, id, body), true)
-	defer c.writeMu.Unlock()
+}
-	if err := c.raw.SetWriteDeadline(time.Now().Add(networkWriteTimeout)); err != nil {
+
-		return err
+// Enqueue adds a packet without waiting for the socket. Server broadcasts use
-	}
+// it so one slow recipient cannot stall the world or another connection. A
-	defer c.raw.SetWriteDeadline(time.Time{})
+// client that exceeds the bounded queue is disconnected.
-	return protocol.WritePacket(c.raw, c.compressionThreshold, id, body)
+func (c *Conn) Enqueue(id int32, body []byte) error {
+	return c.sendFrame(protocol.AppendPacket(nil, c.compressionThreshold, id, body), false)
 }
 
 // SendWriter writes a packet whose body was built with a protocol.Writer.
@@ -89,8 +111,73 @@ func (c *Conn) SendWriter(id int32, w *protocol.Writer) error {
 // The frame must have been built for this connection's compression threshold.
 // Safe for concurrent use.
 func (c *Conn) SendFramed(frame []byte) error {
-	c.writeMu.Lock()
+	return c.sendFrame(frame, true)
-	defer c.writeMu.Unlock()
+}
+
+func (c *Conn) sendFrame(frame []byte, wait bool) error {
+	packet := outboundPacket{frame: frame}
+	if wait {
+		packet.done = make(chan error, 1)
+	}
+	c.outboundMu.Lock()
+	if c.pendingBytes+len(frame) > maxOutboundBytes {
+		c.outboundMu.Unlock()
+		c.abort()
+		return ErrOutboundFull
+	}
+	select {
+	case <-c.done:
+		c.outboundMu.Unlock()
+		return net.ErrClosed
+	case c.outbound <- packet:
+		c.pendingBytes += len(frame)
+		c.outboundMu.Unlock()
+	default:
+		c.outboundMu.Unlock()
+		c.abort()
+		return ErrOutboundFull
+	}
+	if packet.done == nil {
+		return nil
+	}
+	select {
+	case err := <-packet.done:
+		return err
+	case <-c.done:
+		select {
+		case err := <-packet.done:
+			return err
+		default:
+			return net.ErrClosed
+		}
+	}
+}
+
+func (c *Conn) writeLoop() {
+	defer close(c.writerDone)
+	for {
+		select {
+		case packet := <-c.outbound:
+			err := c.writeFrame(packet.frame)
+			c.outboundMu.Lock()
+			c.pendingBytes -= len(packet.frame)
+			c.outboundMu.Unlock()
+			if packet.done != nil {
+				packet.done <- err
+			}
+			if err != nil {
+				c.abort()
+				c.failPending(err)
+				return
+			}
+		case <-c.done:
+			c.failPending(net.ErrClosed)
+			return
+		}
+	}
+}
+
+func (c *Conn) writeFrame(frame []byte) error {
 	if err := c.raw.SetWriteDeadline(time.Now().Add(networkWriteTimeout)); err != nil {
 		return err
 	}
@@ -108,8 +195,35 @@ func (c *Conn) SendFramed(frame []byte) error {
 	return nil
 }
 
+func (c *Conn) failPending(err error) {
+	for {
+		select {
+		case packet := <-c.outbound:
+			c.outboundMu.Lock()
+			c.pendingBytes -= len(packet.frame)
+			c.outboundMu.Unlock()
+			if packet.done != nil {
+				packet.done <- err
+			}
+		default:
+			return
+		}
+	}
+}
+
+func (c *Conn) abort() {
+	c.closeOnce.Do(func() {
+		close(c.done)
+		_ = c.raw.Close()
+	})
+}
+
 // CompressionThreshold returns the active threshold (-1 if disabled).
 func (c *Conn) CompressionThreshold() int32 { return c.compressionThreshold }
 
-// Close closes the underlying connection.
+// Close stops the writer and closes the underlying connection.
-func (c *Conn) Close() error { return c.raw.Close() }
+func (c *Conn) Close() error {
+	c.abort()
+	<-c.writerDone
+	return nil
+}

internal/network/conn_queue_test.go

@@ -0,0 +1,113 @@
+package network
+
+import (
+	"bufio"
+	"bytes"
+	"errors"
+	"io"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"regionio/internal/protocol"
+)
+
+type blockedConn struct {
+	mu      sync.Mutex
+	closed  bool
+	release chan struct{}
+	once    sync.Once
+}
+
+func newBlockedConn() *blockedConn { return &blockedConn{release: make(chan struct{})} }
+
+func (c *blockedConn) Read([]byte) (int, error) { return 0, io.EOF }
+func (c *blockedConn) Write(p []byte) (int, error) {
+	<-c.release
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return 0, net.ErrClosed
+	}
+	return len(p), nil
+}
+func (c *blockedConn) Close() error {
+	c.mu.Lock()
+	c.closed = true
+	c.mu.Unlock()
+	c.once.Do(func() { close(c.release) })
+	return nil
+}
+func (c *blockedConn) LocalAddr() net.Addr              { return testAddr("local") }
+func (c *blockedConn) RemoteAddr() net.Addr             { return testAddr("remote") }
+func (c *blockedConn) SetDeadline(time.Time) error      { return nil }
+func (c *blockedConn) SetReadDeadline(time.Time) error  { return nil }
+func (c *blockedConn) SetWriteDeadline(time.Time) error { return nil }
+
+func TestConnWriterPreservesPacketOrder(t *testing.T) {
+	raw := &recordingConn{}
+	conn := NewConn(raw)
+	defer conn.Close()
+	for id := int32(1); id <= 3; id++ {
+		if err := conn.Send(id, []byte{byte(id)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	raw.mu.Lock()
+	data := append([]byte(nil), raw.buf.Bytes()...)
+	raw.mu.Unlock()
+	reader := bufio.NewReader(bytes.NewReader(data))
+	for want := int32(1); want <= 3; want++ {
+		packet, err := protocol.ReadPacket(reader, -1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if packet.ID != want || len(packet.Data) != 1 || packet.Data[0] != byte(want) {
+			t.Fatalf("packet %d = id %d data %v", want, packet.ID, packet.Data)
+		}
+	}
+}
+
+func TestConnEnqueueDisconnectsAtByteBudget(t *testing.T) {
+	raw := newBlockedConn()
+	conn := NewConn(raw)
+	body := make([]byte, protocol.MaxPacketSize)
+	var err error
+	for i := 0; i < 8; i++ {
+		err = conn.Enqueue(1, body)
+		if err != nil {
+			break
+		}
+	}
+	if !errors.Is(err, ErrOutboundFull) {
+		t.Fatalf("enqueue error = %v, want ErrOutboundFull", err)
+	}
+	select {
+	case <-conn.done:
+	case <-time.After(time.Second):
+		t.Fatal("queue overflow did not close connection")
+	}
+	if err := conn.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestConnCloseUnblocksWaitingSend(t *testing.T) {
+	raw := newBlockedConn()
+	conn := NewConn(raw)
+	result := make(chan error, 1)
+	go func() { result <- conn.Send(1, []byte("blocked")) }()
+	time.Sleep(10 * time.Millisecond)
+	if err := conn.Close(); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case err := <-result:
+		if err == nil {
+			t.Fatal("blocked send succeeded after close")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Close did not unblock Send")
+	}
+}

internal/network/play.go

@@ -25,7 +25,7 @@ const (
 // hands chunk streaming off to the background streamer. The streamer stops when
 // h.ctx (the connection lifetime context) is cancelled.
 func (h *handler) beginPlay() error {
-	session, err := h.srv.RegisterPlayer(h.conn.Profile, h.conn.Send)
+	session, err := h.srv.RegisterPlayer(h.conn.Profile, h.conn.Enqueue)
 	if err != nil {
 		return err
 	}