From c0973773b2d915cd77fa6d24b14a66d6e454cf6d Mon Sep 17 00:00:00 2001 From: Amaan Sayed Date: Wed, 22 Jul 2026 10:43:26 -0500 Subject: [PATCH 1/3] Add write deadline and fatal-write teardown to writeLoop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A socket write to a peer that has stopped reading blocks forever: writeLoop called connection.Write with no deadline, and on error it logged and kept looping. When both sides of a session answer each other's resend requests simultaneously (two-way sequence gap after an unclean disconnect), each side's session goroutine blocks pushing its replay into messageOut while its peer does the same — mutual TCP backpressure with no timeout anywhere. This is the deadlock class of quickfixgo/quickfix#169; the responding side was left unbounded, and it froze a production-path session for 29.5 hours on 2026-07-07/08 (peak6-labs/kalshi incident). writeLoop now takes the net.Conn (both call sites already pass one), arms a 30s write deadline before every write, and treats any write error as fatal to the connection: 1. Close the conn — readLoop's blocked Read fails and it closes msgIn. 2. Drain messageOut — a session goroutine blocked in a sendBytes channel send (e.g. mid-resendMessages) unblocks, returns to its event loop, sees the closed msgIn, and runs onDisconnect, which closes messageOut and ends the drain. The close-before-drain order is load-bearing: the drain terminates only via that disconnect path. Callers need no changes — the initiator wrapper and acceptor handleConnection already Close on their own paths (second close is logged and harmless), so a wedged session now becomes a clean disconnect + standard reconnect within seconds. Note: TestSessionFactorySuite/TestConfigureSocketConnectAddress fails on the unmodified base commit (d520d754) as well — pre-existing, untouched. Co-Authored-By: Claude Fable 5 --- connection.go | 33 +++++++++++++- connection_internal_test.go | 91 +++++++++++++++++++++++++++++++++++-- 2 files changed, 118 insertions(+), 6 deletions(-) diff --git a/connection.go b/connection.go index 99a4c465..94357a89 100644 --- a/connection.go +++ b/connection.go @@ -15,17 +15,46 @@ package quickfix -import "io" +import ( + "net" + "time" +) -func writeLoop(connection io.Writer, messageOut chan []byte, log Log) { +// writeTimeout bounds each socket write. A peer that stops reading — e.g. +// both sides of a session answering each other's resend requests at once, +// neither draining inbound (mutual TCP backpressure) — would otherwise block +// Write forever with no error and no timeout. Package-level so tests can +// shorten it. +var writeTimeout = 30 * time.Second + +func writeLoop(connection net.Conn, messageOut chan []byte, log Log) { for { msg, ok := <-messageOut if !ok { return } + if err := connection.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + log.OnEvent(err.Error()) + } if _, err := connection.Write(msg); err != nil { log.OnEvent(err.Error()) + // A failed write is fatal to the connection. Tear it down in an + // order that unwedges every dependent goroutine: + // 1. Close the conn: readLoop's blocked Read fails and it + // closes msgIn. + // 2. Drain messageOut: a session goroutine blocked in a + // sendBytes channel send (e.g. mid-resendMessages) unblocks, + // returns to its event loop, sees the closed msgIn, and runs + // onDisconnect — which closes messageOut and ends the drain. + // Closing before draining is load-bearing: the drain terminates + // only via that disconnect path. + if closeErr := connection.Close(); closeErr != nil { + log.OnEvent(closeErr.Error()) + } + for range messageOut { //nolint:revive // discard until closed + } + return } } } diff --git a/connection_internal_test.go b/connection_internal_test.go index 081b3c11..ff8e707f 100644 --- a/connection_internal_test.go +++ b/connection_internal_test.go @@ -17,12 +17,44 @@ package quickfix import ( "bytes" + "net" + "os" "strings" "testing" + "time" ) +// mockConn is a minimal net.Conn for exercising writeLoop. Write returns the +// error queued in writeErrs for that call (nil = success, message appended to +// buf). All methods are called only from the writeLoop goroutine; assertions +// read state after writeLoop returns. +type mockConn struct { + buf bytes.Buffer + writeErrs []error + writeCalls int + deadlines int + closeCalls int +} + +func (c *mockConn) Write(p []byte) (int, error) { + call := c.writeCalls + c.writeCalls++ + if call < len(c.writeErrs) && c.writeErrs[call] != nil { + return 0, c.writeErrs[call] + } + return c.buf.Write(p) +} + +func (c *mockConn) Close() error { c.closeCalls++; return nil } +func (c *mockConn) Read(p []byte) (int, error) { return 0, os.ErrClosed } +func (c *mockConn) LocalAddr() net.Addr { return nil } +func (c *mockConn) RemoteAddr() net.Addr { return nil } +func (c *mockConn) SetDeadline(t time.Time) error { return nil } +func (c *mockConn) SetReadDeadline(t time.Time) error { return nil } +func (c *mockConn) SetWriteDeadline(t time.Time) error { c.deadlines++; return nil } + func TestWriteLoop(t *testing.T) { - writer := bytes.NewBufferString("") + conn := &mockConn{} msgOut := make(chan []byte) go func() { @@ -31,12 +63,63 @@ func TestWriteLoop(t *testing.T) { msgOut <- []byte("test msg 3") close(msgOut) }() - writeLoop(writer, msgOut, nullLog{}) + writeLoop(conn, msgOut, nullLog{}) expected := "test msg 1 test msg 2 test msg 3" - if writer.String() != expected { - t.Errorf("expected %v got %v", expected, writer.String()) + if conn.buf.String() != expected { + t.Errorf("expected %v got %v", expected, conn.buf.String()) + } + if conn.deadlines != 3 { + t.Errorf("expected a write deadline set before each of 3 writes, got %v", conn.deadlines) + } + if conn.closeCalls != 0 { + t.Errorf("clean channel close must not close the connection, got %v Close calls", conn.closeCalls) + } +} + +// TestWriteLoopWriteErrorTeardown covers the fatal-write path: a write that +// fails (e.g. deadline exceeded because the peer stopped reading) must close +// the connection and drain messageOut so a session goroutine blocked in a +// send can reach its disconnect path, and writeLoop must return only once +// messageOut is closed. This is the escape hatch for the mutual +// resend-backpressure deadlock (both peers writing, neither reading). +func TestWriteLoopWriteErrorTeardown(t *testing.T) { + conn := &mockConn{writeErrs: []error{os.ErrDeadlineExceeded}} + msgOut := make(chan []byte) + + producerDone := make(chan struct{}) + go func() { + defer close(producerDone) + msgOut <- []byte("failing msg") + // This send blocks until the teardown drain consumes it — exactly + // the position of a session goroutine stuck mid-resend. + msgOut <- []byte("queued msg") + close(msgOut) + }() + + loopDone := make(chan struct{}) + go func() { + defer close(loopDone) + writeLoop(conn, msgOut, nullLog{}) + }() + + select { + case <-producerDone: + case <-time.After(5 * time.Second): + t.Fatal("producer still blocked: writeLoop did not drain messageOut after a write error") + } + select { + case <-loopDone: + case <-time.After(5 * time.Second): + t.Fatal("writeLoop did not return after messageOut was drained and closed") + } + + if conn.closeCalls != 1 { + t.Errorf("expected the connection closed exactly once on write error, got %v", conn.closeCalls) + } + if conn.buf.Len() != 0 { + t.Errorf("no bytes should reach the connection after a failed write, got %q", conn.buf.String()) } } From e42af79a406595302628f9dcca6666b067a507d9 Mon Sep 17 00:00:00 2001 From: Amaan Sayed Date: Wed, 22 Jul 2026 15:19:51 -0500 Subject: [PATCH 2/3] Make socket write timeout session-scoped and opt-in --- acceptor.go | 2 +- config/configuration.go | 17 ++++++ connection.go | 46 ++++++++-------- connection_internal_test.go | 100 +++++++++++++++++++++++++++++------ initiator.go | 2 +- internal/session_settings.go | 1 + session_factory.go | 10 ++++ session_factory_test.go | 15 ++++++ 8 files changed, 149 insertions(+), 44 deletions(-) diff --git a/acceptor.go b/acceptor.go index f58ef01f..80c84de9 100644 --- a/acceptor.go +++ b/acceptor.go @@ -372,7 +372,7 @@ func (a *Acceptor) handleConnection(netConn net.Conn) { readLoop(parser, msgIn, a.globalLog) }() - writeLoop(netConn, msgOut, a.globalLog) + writeLoop(netConn, msgOut, session.SocketWriteTimeout, a.globalLog) } func (a *Acceptor) dynamicSessionsLoop() { diff --git a/config/configuration.go b/config/configuration.go index 1425cde7..45197751 100644 --- a/config/configuration.go +++ b/config/configuration.go @@ -476,6 +476,23 @@ const ( // Valid Values: // - Any positive integer MaxLatency string = "MaxLatency" + + // SocketWriteTimeout bounds each write to the FIX session's network connection. + // A write that exceeds the timeout fails and disconnects the session. This is + // useful when a peer stops reading and TCP backpressure would otherwise block + // the session indefinitely. Applies to both initiators and acceptors. + // + // Example Values: + // - SocketWriteTimeout=30s + // - SocketWriteTimeout=500ms + // + // Required: No + // + // Default: 0 (no timeout) + // + // Valid Values: + // - A positive go time.Duration + SocketWriteTimeout string = "SocketWriteTimeout" ) const ( diff --git a/connection.go b/connection.go index 94357a89..d0697e64 100644 --- a/connection.go +++ b/connection.go @@ -20,45 +20,41 @@ import ( "time" ) -// writeTimeout bounds each socket write. A peer that stops reading — e.g. -// both sides of a session answering each other's resend requests at once, -// neither draining inbound (mutual TCP backpressure) — would otherwise block -// Write forever with no error and no timeout. Package-level so tests can -// shorten it. -var writeTimeout = 30 * time.Second - -func writeLoop(connection net.Conn, messageOut chan []byte, log Log) { +func writeLoop(connection net.Conn, messageOut chan []byte, writeTimeout time.Duration, log Log) { for { msg, ok := <-messageOut if !ok { return } - if err := connection.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { - log.OnEvent(err.Error()) + if writeTimeout > 0 { + if err := connection.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + log.OnEvent(err.Error()) + closeAndDrainConnection(connection, messageOut, log) + return + } } if _, err := connection.Write(msg); err != nil { log.OnEvent(err.Error()) - // A failed write is fatal to the connection. Tear it down in an - // order that unwedges every dependent goroutine: - // 1. Close the conn: readLoop's blocked Read fails and it - // closes msgIn. - // 2. Drain messageOut: a session goroutine blocked in a - // sendBytes channel send (e.g. mid-resendMessages) unblocks, - // returns to its event loop, sees the closed msgIn, and runs - // onDisconnect — which closes messageOut and ends the drain. - // Closing before draining is load-bearing: the drain terminates - // only via that disconnect path. - if closeErr := connection.Close(); closeErr != nil { - log.OnEvent(closeErr.Error()) - } - for range messageOut { //nolint:revive // discard until closed - } + closeAndDrainConnection(connection, messageOut, log) return } } } +// closeAndDrainConnection tears down a failed writer in an order that +// unwedges the session and reader goroutines. Closing the connection makes the +// reader fail and eventually disconnect the session. Draining messageOut lets +// a session blocked mid-resend finish sending, observe that disconnect, and +// close messageOut. Closing before draining is therefore load-bearing. +func closeAndDrainConnection(connection net.Conn, messageOut chan []byte, log Log) { + if err := connection.Close(); err != nil { + log.OnEvent(err.Error()) + } + for range messageOut { //nolint:revive // discard until session disconnects + } +} + func readLoop(parser *parser, msgIn chan fixIn, log Log) { defer close(msgIn) diff --git a/connection_internal_test.go b/connection_internal_test.go index ff8e707f..68fda1c7 100644 --- a/connection_internal_test.go +++ b/connection_internal_test.go @@ -29,11 +29,12 @@ import ( // buf). All methods are called only from the writeLoop goroutine; assertions // read state after writeLoop returns. type mockConn struct { - buf bytes.Buffer - writeErrs []error - writeCalls int - deadlines int - closeCalls int + buf bytes.Buffer + writeErrs []error + writeCalls int + deadlineErr error + deadlines int + closeCalls int } func (c *mockConn) Write(p []byte) (int, error) { @@ -45,15 +46,18 @@ func (c *mockConn) Write(p []byte) (int, error) { return c.buf.Write(p) } -func (c *mockConn) Close() error { c.closeCalls++; return nil } -func (c *mockConn) Read(p []byte) (int, error) { return 0, os.ErrClosed } -func (c *mockConn) LocalAddr() net.Addr { return nil } -func (c *mockConn) RemoteAddr() net.Addr { return nil } -func (c *mockConn) SetDeadline(t time.Time) error { return nil } -func (c *mockConn) SetReadDeadline(t time.Time) error { return nil } -func (c *mockConn) SetWriteDeadline(t time.Time) error { c.deadlines++; return nil } +func (c *mockConn) Close() error { c.closeCalls++; return nil } +func (c *mockConn) Read(_ []byte) (int, error) { return 0, os.ErrClosed } +func (c *mockConn) LocalAddr() net.Addr { return nil } +func (c *mockConn) RemoteAddr() net.Addr { return nil } +func (c *mockConn) SetDeadline(_ time.Time) error { return nil } +func (c *mockConn) SetReadDeadline(_ time.Time) error { return nil } +func (c *mockConn) SetWriteDeadline(_ time.Time) error { + c.deadlines++ + return c.deadlineErr +} -func TestWriteLoop(t *testing.T) { +func TestWriteLoopWithoutTimeout(t *testing.T) { conn := &mockConn{} msgOut := make(chan []byte) @@ -63,21 +67,36 @@ func TestWriteLoop(t *testing.T) { msgOut <- []byte("test msg 3") close(msgOut) }() - writeLoop(conn, msgOut, nullLog{}) + writeLoop(conn, msgOut, 0, nullLog{}) expected := "test msg 1 test msg 2 test msg 3" if conn.buf.String() != expected { t.Errorf("expected %v got %v", expected, conn.buf.String()) } - if conn.deadlines != 3 { - t.Errorf("expected a write deadline set before each of 3 writes, got %v", conn.deadlines) + if conn.deadlines != 0 { + t.Errorf("disabled timeout must not set a deadline, got %v calls", conn.deadlines) } if conn.closeCalls != 0 { t.Errorf("clean channel close must not close the connection, got %v Close calls", conn.closeCalls) } } +func TestWriteLoopSetsDeadlineBeforeEveryWrite(t *testing.T) { + conn := &mockConn{} + msgOut := make(chan []byte, 3) + msgOut <- []byte("one") + msgOut <- []byte("two") + msgOut <- []byte("three") + close(msgOut) + + writeLoop(conn, msgOut, time.Second, nullLog{}) + + if conn.deadlines != 3 { + t.Errorf("expected a write deadline before each write, got %v calls", conn.deadlines) + } +} + // TestWriteLoopWriteErrorTeardown covers the fatal-write path: a write that // fails (e.g. deadline exceeded because the peer stopped reading) must close // the connection and drain messageOut so a session goroutine blocked in a @@ -101,7 +120,7 @@ func TestWriteLoopWriteErrorTeardown(t *testing.T) { loopDone := make(chan struct{}) go func() { defer close(loopDone) - writeLoop(conn, msgOut, nullLog{}) + writeLoop(conn, msgOut, time.Second, nullLog{}) }() select { @@ -123,6 +142,53 @@ func TestWriteLoopWriteErrorTeardown(t *testing.T) { } } +func TestWriteLoopDeadlineErrorTeardown(t *testing.T) { + conn := &mockConn{deadlineErr: os.ErrInvalid} + msgOut := make(chan []byte, 1) + msgOut <- []byte("must not be written") + close(msgOut) + + writeLoop(conn, msgOut, time.Second, nullLog{}) + + if conn.writeCalls != 0 { + t.Errorf("deadline failure must prevent an unbounded write, got %v writes", conn.writeCalls) + } + if conn.closeCalls != 1 { + t.Errorf("deadline failure must close the connection once, got %v calls", conn.closeCalls) + } +} + +func TestWriteLoopBlockedWriteTimesOutAndDrainsProducer(t *testing.T) { + conn, peer := net.Pipe() + defer peer.Close() + + msgOut := make(chan []byte) + producerDone := make(chan struct{}) + go func() { + defer close(producerDone) + msgOut <- []byte("blocked write") + msgOut <- []byte("queued behind blocked write") + close(msgOut) + }() + + loopDone := make(chan struct{}) + go func() { + defer close(loopDone) + writeLoop(conn, msgOut, 100*time.Millisecond, nullLog{}) + }() + + select { + case <-producerDone: + case <-time.After(5 * time.Second): + t.Fatal("producer remained blocked after the socket write deadline") + } + select { + case <-loopDone: + case <-time.After(5 * time.Second): + t.Fatal("writeLoop did not return after draining the closed output channel") + } +} + func TestReadLoop(t *testing.T) { msgIn := make(chan fixIn) stream := "hello8=FIX.4.09=5blah10=103garbage8=FIX.4.09=4foo10=103" diff --git a/initiator.go b/initiator.go index 525ff5b6..0419d3fc 100644 --- a/initiator.go +++ b/initiator.go @@ -245,7 +245,7 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di go readLoop(newParser(bufio.NewReader(netConn)), msgIn, session.log) disconnected = make(chan interface{}) go func() { - writeLoop(netConn, msgOut, session.log) + writeLoop(netConn, msgOut, session.SocketWriteTimeout, session.log) if err := netConn.Close(); err != nil { session.log.OnEvent(err.Error()) } diff --git a/internal/session_settings.go b/internal/session_settings.go index ef7413eb..fe124ded 100644 --- a/internal/session_settings.go +++ b/internal/session_settings.go @@ -17,6 +17,7 @@ type SessionSettings struct { EnableNextExpectedMsgSeqNum bool SkipCheckLatency bool MaxLatency time.Duration + SocketWriteTimeout time.Duration DisableMessagePersist bool TimeZone *time.Location ResetSeqTime time.Time diff --git a/session_factory.go b/session_factory.go index 0ef53ca4..78014d3f 100644 --- a/session_factory.go +++ b/session_factory.go @@ -245,6 +245,16 @@ func (f sessionFactory) newSession( } } + if settings.HasSetting(config.SocketWriteTimeout) { + if s.SocketWriteTimeout, err = settings.DurationSetting(config.SocketWriteTimeout); err != nil { + return + } + if s.SocketWriteTimeout <= 0 { + err = errors.New("SocketWriteTimeout must be greater than zero") + return + } + } + if settings.HasSetting(config.StartTime) || settings.HasSetting(config.EndTime) { var startTimeStr, endTimeStr string if startTimeStr, err = settings.Setting(config.StartTime); err != nil { diff --git a/session_factory_test.go b/session_factory_test.go index 349e7580..dac34cfe 100644 --- a/session_factory_test.go +++ b/session_factory_test.go @@ -62,6 +62,7 @@ func (s *SessionFactorySuite) TestDefaults() { s.Equal("", session.DefaultApplVerID) s.False(session.InitiateLogon) s.Equal(0, session.ResendRequestChunkSize) + s.Zero(session.SocketWriteTimeout) s.False(session.EnableLastMsgSeqNumProcessed) s.False(session.SkipCheckLatency) s.Equal(Millis, session.timestampPrecision) @@ -70,6 +71,20 @@ func (s *SessionFactorySuite) TestDefaults() { s.False(session.HeartBtIntOverride) } +func (s *SessionFactorySuite) TestSocketWriteTimeout() { + s.SessionSettings.Set(config.SocketWriteTimeout, "250ms") + session, err := s.newSession(s.SessionID, s.MessageStoreFactory, s.SessionSettings, s.LogFactory, s.App) + s.NoError(err) + s.Equal(250*time.Millisecond, session.SocketWriteTimeout) + + for _, value := range []string{"not-a-duration", "0s", "-1s"} { + s.SetupTest() + s.SessionSettings.Set(config.SocketWriteTimeout, value) + _, err = s.newSession(s.SessionID, s.MessageStoreFactory, s.SessionSettings, s.LogFactory, s.App) + s.Error(err, "SocketWriteTimeout=%q must be rejected", value) + } +} + func (s *SessionFactorySuite) TestResetOnLogon() { var tests = []struct { setting string From 945c5381c36deff595197db44ed8e887787a6d88 Mon Sep 17 00:00:00 2001 From: Amaan Sayed Date: Wed, 22 Jul 2026 16:07:14 -0500 Subject: [PATCH 3/3] Preserve write behavior when timeout is disabled --- connection.go | 6 ++++-- connection_internal_test.go | 24 ++++++++++++++++++++++-- 2 files changed, 26 insertions(+), 4 deletions(-) diff --git a/connection.go b/connection.go index d0697e64..a4095cf0 100644 --- a/connection.go +++ b/connection.go @@ -36,8 +36,10 @@ func writeLoop(connection net.Conn, messageOut chan []byte, writeTimeout time.Du } if _, err := connection.Write(msg); err != nil { log.OnEvent(err.Error()) - closeAndDrainConnection(connection, messageOut, log) - return + if writeTimeout > 0 { + closeAndDrainConnection(connection, messageOut, log) + return + } } } } diff --git a/connection_internal_test.go b/connection_internal_test.go index 68fda1c7..78f005d4 100644 --- a/connection_internal_test.go +++ b/connection_internal_test.go @@ -97,8 +97,28 @@ func TestWriteLoopSetsDeadlineBeforeEveryWrite(t *testing.T) { } } -// TestWriteLoopWriteErrorTeardown covers the fatal-write path: a write that -// fails (e.g. deadline exceeded because the peer stopped reading) must close +func TestWriteLoopWithoutTimeoutPreservesWriteErrorBehavior(t *testing.T) { + conn := &mockConn{writeErrs: []error{os.ErrClosed}} + msgOut := make(chan []byte, 2) + msgOut <- []byte("failing msg") + msgOut <- []byte("next msg") + close(msgOut) + + writeLoop(conn, msgOut, 0, nullLog{}) + + if conn.closeCalls != 0 { + t.Errorf("disabled timeout must preserve the existing non-fatal write-error path, got %v Close calls", conn.closeCalls) + } + if conn.writeCalls != 2 { + t.Errorf("disabled timeout must continue after a write error, got %v writes", conn.writeCalls) + } + if conn.buf.String() != "next msg" { + t.Errorf("expected the next message to be written, got %q", conn.buf.String()) + } +} + +// TestWriteLoopWriteErrorTeardown covers the opted-in fatal-write path: a write +// that fails (e.g. deadline exceeded because the peer stopped reading) must close // the connection and drain messageOut so a session goroutine blocked in a // send can reach its disconnect path, and writeLoop must return only once // messageOut is closed. This is the escape hatch for the mutual