Add opt-in session-scoped FIX socket write timeout

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() {

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 (

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)
 

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")
 	}
 }
 

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())
 			}

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

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 {

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