From f11b107e3ae06aac936e104a66ec1946828f2a1c Mon Sep 17 00:00:00 2001
From: Vishesh Gupta <vish.gupta@alpaca.markets>
Date: Thu, 16 Jul 2026 12:52:15 -0400
Subject: [PATCH] Stop writeLoop flood after TCP dead-socket write errors

writeLoop previously logged and kept writing after the first broken pipe,
producing thousands of events per disconnect under load. On write failure,
log once, close the connection to trigger disconnect, and drain messageOut
without further writes so the session send path does not deadlock.
---
 connection.go      | 13 +++++++
 connection_test.go | 97 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 110 insertions(+)
 create mode 100644 connection_test.go

diff --git a/connection.go b/connection.go
index 99a4c465e..ac6627052 100644
--- a/connection.go
+++ b/connection.go
@@ -18,14 +18,27 @@ package quickfix
 import "io"
 
 func writeLoop(connection io.Writer, messageOut chan []byte, log Log) {
+	var writeFailed bool
 	for {
 		msg, ok := <-messageOut
 		if !ok {
 			return
 		}
 
+		// After the first write failure the peer socket is dead. Keep draining
+		// messageOut (without writing or re-logging) so the session goroutine
+		// is not blocked on send, while closing the connection so readLoop
+		// fails and triggers a clean disconnect that closes messageOut.
+		if writeFailed {
+			continue
+		}
+
 		if _, err := connection.Write(msg); err != nil {
 			log.OnEvent(err.Error())
+			writeFailed = true
+			if closer, ok := connection.(io.Closer); ok {
+				_ = closer.Close()
+			}
 		}
 	}
 }
diff --git a/connection_test.go b/connection_test.go
new file mode 100644
index 000000000..6ab6ce5a7
--- /dev/null
+++ b/connection_test.go
@@ -0,0 +1,97 @@
+package quickfix
+
+import (
+	"errors"
+	"io"
+	"sync"
+	"testing"
+	"time"
+)
+
+type stubLog struct {
+	mu     sync.Mutex
+	events []string
+}
+
+func (l *stubLog) OnIncoming([]byte)           {}
+func (l *stubLog) OnOutgoing([]byte)           {}
+func (l *stubLog) OnEventf(string, ...interface{}) {}
+func (l *stubLog) OnEvent(s string) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.events = append(l.events, s)
+}
+
+func (l *stubLog) eventCount() int {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return len(l.events)
+}
+
+type failWriter struct {
+	mu     sync.Mutex
+	writes int
+	closed bool
+}
+
+func (w *failWriter) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.writes++
+	return 0, errors.New("write tcp: broken pipe")
+}
+
+func (w *failWriter) Close() error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.closed = true
+	return nil
+}
+
+func (w *failWriter) writeCount() int {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.writes
+}
+
+func (w *failWriter) wasClosed() bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.closed
+}
+
+var _ io.WriteCloser = (*failWriter)(nil)
+
+func TestWriteLoopStopsWritingAfterBrokenPipe(t *testing.T) {
+	msgOut := make(chan []byte)
+	log := &stubLog{}
+	w := &failWriter{}
+
+	done := make(chan struct{})
+	go func() {
+		writeLoop(w, msgOut, log)
+		close(done)
+	}()
+
+	// Flood the write loop with messages after the socket is dead.
+	for i := 0; i < 50; i++ {
+		msgOut <- []byte("8=FIX.4.2|35=8|")
+	}
+	close(msgOut)
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("writeLoop did not exit after messageOut was closed")
+	}
+
+	if got := w.writeCount(); got != 1 {
+		t.Fatalf("expected exactly 1 Write attempt after broken pipe, got %d", got)
+	}
+	if got := log.eventCount(); got != 1 {
+		t.Fatalf("expected exactly 1 broken-pipe log event, got %d", got)
+	}
+	if !w.wasClosed() {
+		t.Fatal("expected connection to be closed after write failure")
+	}
+}