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 99a4c465..a4095cf0 100644 --- a/connection.go +++ b/connection.go @@ -15,21 +15,48 @@ package quickfix -import "io" +import ( + "net" + "time" +) -func writeLoop(connection io.Writer, 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 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()) + if writeTimeout > 0 { + 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 081b3c11..78f005d4 100644 --- a/connection_internal_test.go +++ b/connection_internal_test.go @@ -17,12 +17,48 @@ package quickfix import ( "bytes" + "net" + "os" "strings" "testing" + "time" ) -func TestWriteLoop(t *testing.T) { - writer := bytes.NewBufferString("") +// 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 + deadlineErr error + 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(_ []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 TestWriteLoopWithoutTimeout(t *testing.T) { + conn := &mockConn{} msgOut := make(chan []byte) go func() { @@ -31,12 +67,145 @@ func TestWriteLoop(t *testing.T) { msgOut <- []byte("test msg 3") close(msgOut) }() - writeLoop(writer, msgOut, nullLog{}) + writeLoop(conn, msgOut, 0, 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 != 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) + } +} + +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 +// 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, time.Second, 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()) + } +} + +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") } } 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