aboutsummaryrefslogtreecommitdiff
path: root/hotline/client_conn.go
diff options
context:
space:
mode:
authorJeff Halter <868228+jhalter@users.noreply.github.com>2026-06-12 08:11:55 -0700
committerJeff Halter <868228+jhalter@users.noreply.github.com>2026-06-12 08:11:55 -0700
commit7ebc802d0a269218f05b3b51eb10ac66eacb4d1f (patch)
tree0c700edcde0fcc034605dfcf98b81ddb1cf174b6 /hotline/client_conn.go
parentd2791fcbadcf332dc05e5bebc7350eca50617263 (diff)
Replace shared outbox with per-client send queues
The outbox channel spawned one goroutine per outbound transaction, so concurrent sends to the same client could interleave bytes within the transaction framing, per-client message ordering was not guaranteed, and a slow client accumulated unbounded goroutines. Each ClientConn now has a bounded send queue drained by a single writer goroutine, which serializes writes and preserves enqueue order. Send never blocks: if a client's queue overflows, its connection is closed and the read loop performs the usual disconnect cleanup. Server.Send routes transactions to the target client's queue, replacing processOutbox and sendTransaction. Handler signatures are unchanged. Disconnect now removes the client from the manager before notifying peers so no new transactions are routed to a departing client, then idempotently closes its send queue. New tests cover write ordering, framing integrity under concurrent senders, the slow-client disconnect policy, and a Send/Disconnect race exercise (run with -race).
Diffstat (limited to 'hotline/client_conn.go')
-rw-r--r--hotline/client_conn.go81
1 files changed, 78 insertions, 3 deletions
diff --git a/hotline/client_conn.go b/hotline/client_conn.go
index c17ef35..363376e 100644
--- a/hotline/client_conn.go
+++ b/hotline/client_conn.go
@@ -21,6 +21,10 @@ var clientConnSortFunc = func(a, b *ClientConn) int {
)
}
+// sendQueueDepth is the number of transactions that can be queued for delivery to a client before
+// the client is considered too slow and is disconnected.
+const sendQueueDepth = 64
+
// ClientConn represents a client connected to a Server
type ClientConn struct {
Connection io.ReadWriteCloser
@@ -43,6 +47,74 @@ type ClientConn struct {
Logger *slog.Logger
mu sync.RWMutex
+
+ sendCh chan Transaction
+ sendInit sync.Once
+ sendMu sync.Mutex // guards sendClosed and close(sendCh)
+ sendClosed bool
+}
+
+func (cc *ClientConn) initSendQueue() {
+ cc.sendInit.Do(func() { cc.sendCh = make(chan Transaction, sendQueueDepth) })
+}
+
+// Send enqueues t for delivery to this client by its writer goroutine, preserving enqueue order.
+// It never blocks: if the queue is full, the client is considered too slow and its connection is
+// closed, which unblocks the client's read loop and triggers the usual Disconnect cleanup.
+func (cc *ClientConn) Send(t Transaction) {
+ cc.initSendQueue()
+
+ cc.sendMu.Lock()
+ defer cc.sendMu.Unlock()
+
+ if cc.sendClosed {
+ return
+ }
+
+ select {
+ case cc.sendCh <- t:
+ default:
+ cc.sendClosed = true
+ close(cc.sendCh)
+
+ if cc.Logger != nil {
+ cc.Logger.Warn("Send queue full; disconnecting slow client")
+ }
+ if cc.Connection != nil {
+ _ = cc.Connection.Close()
+ }
+ }
+}
+
+// closeSendQueue idempotently closes the send queue, stopping the client's writer goroutine after
+// it drains any remaining queued transactions.
+func (cc *ClientConn) closeSendQueue() {
+ cc.initSendQueue()
+
+ cc.sendMu.Lock()
+ defer cc.sendMu.Unlock()
+
+ if !cc.sendClosed {
+ cc.sendClosed = true
+ close(cc.sendCh)
+ }
+}
+
+// writeLoop is the single writer to cc.Connection. Serializing all writes through one goroutine
+// prevents concurrent sends from interleaving bytes within the connection's transaction framing.
+// It runs until the send queue is closed or a write fails.
+func (cc *ClientConn) writeLoop() {
+ cc.initSendQueue()
+
+ for t := range cc.sendCh {
+ if _, err := io.Copy(cc.Connection, &t); err != nil {
+ if cc.Logger != nil {
+ cc.Logger.Debug("error writing transaction to client", "err", err)
+ }
+ _ = cc.Connection.Close()
+ return
+ }
+ }
}
func (cc *ClientConn) TextDecoder() *encoding.Decoder { return cc.Server.TextDecoder }
@@ -109,7 +181,7 @@ func (cftm *ClientFileTransferMgr) Delete(ftType FileTransferType, id FileTransf
func (cc *ClientConn) SendAll(t [2]byte, fields ...Field) {
for _, c := range cc.Server.ClientMgr.List() {
- cc.Server.outbox <- NewTransaction(t, c.ID, fields...)
+ c.Send(NewTransaction(t, c.ID, fields...))
}
}
@@ -124,7 +196,7 @@ func (cc *ClientConn) handleTransaction(transaction Transaction) {
}
for _, t := range handler(cc, &transaction) {
- cc.Server.outbox <- t
+ cc.Server.Send(t)
}
}
@@ -169,12 +241,15 @@ func (cc *ClientConn) Authorize(access int) bool {
// Disconnect notifies other clients that a client has disconnected and closes the connection.
func (cc *ClientConn) Disconnect() {
+ // Remove the client from the manager first so no new transactions are routed to it.
cc.Server.ClientMgr.Delete(cc.ID)
for _, t := range cc.NotifyOthers(NewTransaction(TranNotifyDeleteUser, [2]byte{}, NewField(FieldUserID, cc.ID[:]))) {
- cc.Server.outbox <- t
+ cc.Server.Send(t)
}
+ cc.closeSendQueue()
+
if err := cc.Connection.Close(); err != nil {
cc.Server.Logger.Debug("error closing client connection", "remoteAddr", cc.RemoteAddr)
}