fix(initiator): bound writes on stop and close the connection

config/configuration.go

@@ -587,7 +587,7 @@ const (
 	//
 	// Required: No
 	//
-	// Default: 0 (no timeout)
+	// Default: 0 (the handshake is bounded by LogonTimeout)
 	//
 	// Valid Values:
 	//  - A valid go time.Duration

connection.go

@@ -15,7 +15,60 @@
 
 package quickfix
 
-import "io"
+import (
+	"crypto/tls"
+	"errors"
+	"io"
+	"net"
+	"sync/atomic"
+	"time"
+)
+
+// stopWriter is the initiator's connection as its writer sees it. Once the
+// initiator stops, every write, including one already in progress, must
+// finish within timeout. A write that does not leaves the stream with a
+// partial message on it and the counterparty not reading, so it closes the
+// connection: the writer then fails at once, and the session can stop.
+type stopWriter struct {
+	conn     net.Conn
+	timeout  time.Duration
+	stopping atomic.Bool
+}
+
+// stop bounds the write in progress and every later one by timeout.
+func (w *stopWriter) stop() {
+	w.stopping.Store(true)
+	_ = w.conn.SetWriteDeadline(time.Now().Add(w.timeout))
+}
+
+func (w *stopWriter) Write(b []byte) (int, error) {
+	if w.stopping.Load() {
+		_ = w.conn.SetWriteDeadline(time.Now().Add(w.timeout))
+	}
+	n, err := w.conn.Write(b)
+	if err != nil && w.stopping.Load() {
+		_ = transport(w.conn).Close()
+	}
+	return n, err
+}
+
+// transport is the connection under TLS, if any. Closing it ends a blocked
+// write or close notification at once; closing a tls.Conn already being
+// closed only reports net.ErrClosed while that close notification waits.
+func transport(conn net.Conn) net.Conn {
+	if tlsConn, ok := conn.(*tls.Conn); ok {
+		return tlsConn.NetConn()
+	}
+	return conn
+}
+
+// closeTransport closes the connection's transport. It is closed once this
+// returns, whoever closed it first.
+func closeTransport(conn net.Conn, log Log) {
+	if err := transport(conn).Close(); err != nil && !errors.Is(err, net.ErrClosed) {
+		log.OnEvent(err.Error())
+	}
+}
 
 func writeLoop(connection io.Writer, messageOut chan []byte, log Log) {
 	for {

dialer.go

@@ -25,21 +25,30 @@ import (
 	"github.com/quickfixgo/quickfix/config"
 )
 
+// loadSocketTimeout reads SocketTimeout: a duration, else whole seconds. It
+// is zero when the setting is absent.
+func loadSocketTimeout(settings *SessionSettings) (time.Duration, error) {
+	if !settings.HasSetting(config.SocketTimeout) {
+		return 0, nil
+	}
+	timeout, err := settings.DurationSetting(config.SocketTimeout)
+	if err == nil {
+		return timeout, nil
+	}
+	timeoutInt, err := settings.IntSetting(config.SocketTimeout)
+	if err != nil {
+		return 0, err
+	}
+	return time.Duration(timeoutInt) * time.Second, nil
+}
+
 func loadDialerConfig(settings *SessionSettings) (dialer proxy.ContextDialer, err error) {
 	stdDialer := &net.Dialer{}
-	if settings.HasSetting(config.SocketTimeout) {
-		timeout, err := settings.DurationSetting(config.SocketTimeout)
-		if err != nil {
-			timeoutInt, err := settings.IntSetting(config.SocketTimeout)
-			if err != nil {
-				return stdDialer, err
-			}
-
-			stdDialer.Timeout = time.Duration(timeoutInt) * time.Second
-		} else {
-			stdDialer.Timeout = timeout
-		}
+	timeout, err := loadSocketTimeout(settings)
+	if err != nil {
+		return stdDialer, err
 	}
+	stdDialer.Timeout = timeout
 	dialer = stdDialer
 
 	if !settings.HasSetting(config.ProxyType) {

initiator.go

@@ -156,6 +156,14 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 		wg.Wait()
 	}()
 
+	// SocketTimeout bounds the TLS handshake. Without it LogonTimeout does,
+	// so a counterparty that accepts the connection and never answers cannot
+	// hold a connection attempt open.
+	handshakeTimeout := session.LogonTimeout
+	if timeout, err := loadSocketTimeout(i.sessionSettings[session.sessionID]); err == nil && timeout > 0 {
+		handshakeTimeout = timeout
+	}
+
 	connectionAttempt := 0
 
 	for {
@@ -179,6 +187,7 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 		var disconnected chan interface{}
 		var msgIn chan fixIn
 		var msgOut chan []byte
+		var writer *stopWriter
 
 		address := session.SocketConnectAddress[connectionAttempt%len(session.SocketConnectAddress)]
 		session.log.OnEventf("Connecting to: %v", address)
@@ -197,9 +206,14 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 				}
 				tlsConfig.ServerName = serverName
 			}
+			// The handshake ends with the dial context, so Stop cancels it.
 			tlsConn := tls.Client(netConn, tlsConfig)
-			if err = tlsConn.Handshake(); err != nil {
+			handshakeCtx, cancelHandshake := context.WithTimeout(ctx, handshakeTimeout)
+			err = tlsConn.HandshakeContext(handshakeCtx)
+			cancelHandshake()
+			if err != nil {
 				session.log.OnEventf("Failed handshake: %v", err)
+				closeTransport(netConn, session.log)
 				goto reconnect
 			}
 			netConn = tlsConn
@@ -209,13 +223,15 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 		msgOut = make(chan []byte)
 		if err := session.connect(msgIn, msgOut); err != nil {
 			session.log.OnEventf("Failed to initiate: %v", err)
+			closeTransport(netConn, session.log)
 			goto reconnect
 		}
 
 		go readLoop(newParser(bufio.NewReader(netConn)), msgIn, session.log)
 		disconnected = make(chan interface{})
+		writer = &stopWriter{conn: netConn, timeout: session.LogoutTimeout}
 		go func() {
-			writeLoop(netConn, msgOut, session.log)
+			writeLoop(writer, msgOut, session.log)
 			if err := netConn.Close(); err != nil {
 				session.log.OnEvent(err.Error())
 			}
@@ -229,6 +245,26 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 		select {
 		case <-disconnected:
 		case <-i.stopChan:
+			// A write has no deadline of its own, and the session loop
+			// waits for the writer to take each message: a counterparty
+			// that stops reading would hold a write, the session loop
+			// behind it, and so this stop, with the connection open. From
+			// here every write gets LogoutTimeout. A write still not taken
+			// then closes the connection, and the session can stop.
+			writer.stop()
+			session.stop()
+			wg.Wait()
+			// The writer may still be on its last message (a Logout, or
+			// the answer to the counterparty's) or closing the connection.
+			// Give it LogoutTimeout, then close the transport under it, so
+			// Stop never returns while the connection is open.
+			timer := time.NewTimer(session.LogoutTimeout)
+			select {
+			case <-disconnected:
+			case <-timer.C:
+			}
+			timer.Stop()
+			closeTransport(netConn, session.log)
 			return
 		}