feat(agent): add timeout on write calls

When sending a message via agent, the write connection does not have
a default timeout, leading to blocking calls if the connections is
unhealthy or simply a bad route to it. Thus, adding a write timeout
to the writes so we fail but don't block. If deadline is exceeded then
we log an error but don't close a connection, since it might fail due
to bad route but not a broken pipe per-se. The new configuration can
be found under: `pitaya.buffer.agent.writetimeout`, with 10s by deafult.

The timeout is set using `SetWriteDeadline` function from go net pkg:
https://pkg.go.dev/net#Conn

agent/agent.go

@@ -26,11 +26,13 @@ import (
 	e "errors"
 	"fmt"
 	"net"
+	"os"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/topfreegames/pitaya/v2/config"
 	"github.com/topfreegames/pitaya/v2/conn/codec"
 	"github.com/topfreegames/pitaya/v2/conn/message"
 	"github.com/topfreegames/pitaya/v2/conn/packet"
@@ -76,6 +78,7 @@ type (
 		decoder            codec.PacketDecoder // binary decoder
 		encoder            codec.PacketEncoder // binary encoder
 		heartbeatTimeout   time.Duration
+		writeTimeout       time.Duration
 		lastAt             int64 // last heartbeat unix time stamp
 		messageEncoder     message.Encoder
 		messagesBufferSize int // size of the pending messages buffer
@@ -130,6 +133,7 @@ type (
 		decoder            codec.PacketDecoder // binary decoder
 		encoder            codec.PacketEncoder // binary encoder
 		heartbeatTimeout   time.Duration
+		writeTimeout       time.Duration
 		messageEncoder     message.Encoder
 		messagesBufferSize int // size of the pending messages buffer
 		metricsReporters   []metrics.Reporter
@@ -144,6 +148,7 @@ func NewAgentFactory(
 	encoder codec.PacketEncoder,
 	serializer serialize.Serializer,
 	heartbeatTimeout time.Duration,
+	writeTimeout time.Duration,
 	messageEncoder message.Encoder,
 	messagesBufferSize int,
 	sessionPool session.SessionPool,
@@ -154,6 +159,7 @@ func NewAgentFactory(
 		decoder:            decoder,
 		encoder:            encoder,
 		heartbeatTimeout:   heartbeatTimeout,
+		writeTimeout:       writeTimeout,
 		messageEncoder:     messageEncoder,
 		messagesBufferSize: messagesBufferSize,
 		sessionPool:        sessionPool,
@@ -164,7 +170,7 @@ func NewAgentFactory(
 
 // CreateAgent returns a new agent
 func (f *agentFactoryImpl) CreateAgent(conn net.Conn) Agent {
-	return newAgent(conn, f.decoder, f.encoder, f.serializer, f.heartbeatTimeout, f.messagesBufferSize, f.appDieChan, f.messageEncoder, f.metricsReporters, f.sessionPool)
+	return newAgent(conn, f.decoder, f.encoder, f.serializer, f.heartbeatTimeout, f.writeTimeout, f.messagesBufferSize, f.appDieChan, f.messageEncoder, f.metricsReporters, f.sessionPool)
 }
 
 // NewAgent create new agent instance
@@ -174,6 +180,7 @@ func newAgent(
 	packetEncoder codec.PacketEncoder,
 	serializer serialize.Serializer,
 	heartbeatTime time.Duration,
+	writeTimeout time.Duration,
 	messagesBufferSize int,
 	dieChan chan bool,
 	messageEncoder message.Encoder,
@@ -188,6 +195,10 @@ func newAgent(
 		herdEncode(heartbeatTime, packetEncoder, messageEncoder.IsCompressionEnabled(), serializerName)
 	})
 
+	if writeTimeout <= 0 {
+		writeTimeout = config.DefaultWriteTimeout
+	}
+
 	a := &agentImpl{
 		appDieChan:         dieChan,
 		chDie:              make(chan struct{}),
@@ -199,6 +210,7 @@ func newAgent(
 		decoder:            packetDecoder,
 		encoder:            packetEncoder,
 		heartbeatTimeout:   heartbeatTime,
+		writeTimeout:       writeTimeout,
 		lastAt:             time.Now().Unix(),
 		serializer:         serializer,
 		state:              constants.StatusStart,
@@ -503,19 +515,23 @@ func (a *agentImpl) write() {
 			ctx, err, data := pWrite.ctx, pWrite.err, pWrite.data
 
 			writeErr := a.writeToConnection(ctx, data)
-			if writeErr != nil {
-				err = errors.NewError(writeErr, errors.ErrClosedRequest)
-
-				logger.Log.Errorf("Failed to write in conn: %s (ctx=%v), agent will close", writeErr.Error(), ctx)
-			}
 
 			tracing.FinishSpan(ctx, nil)
-			metrics.ReportTimingFromCtx(ctx, a.metricsReporters, handlerType, err)
 
-			// close agent if low-level conn broke during write
 			if writeErr != nil {
-				return
+				if e.Is(writeErr, os.ErrDeadlineExceeded) {
+					// Log the timeout error but continue processing
+					logger.Log.Warnf("Context deadline exceeded for write in conn %s: %s (ctx=%v)", writeErr.Error(), ctx)
+				} else {
+					err = errors.NewError(writeErr, errors.ErrClosedRequest)
+					logger.Log.Errorf("Failed to write in conn: %s (ctx=%v), agent will close", writeErr.Error(), ctx)
+					metrics.ReportTimingFromCtx(ctx, a.metricsReporters, handlerType, err)
+					// close agent if low-level conn broke during write
+					return
+				}
 			}
+
+			metrics.ReportTimingFromCtx(ctx, a.metricsReporters, handlerType, err)
 		case <-a.chStopWrite:
 			return
 		}
@@ -524,17 +540,15 @@ func (a *agentImpl) write() {
 
 func (a *agentImpl) writeToConnection(ctx context.Context, data []byte) error {
 	span := createConnectionSpan(ctx, a.conn, "conn write")
-
-	_, writeErr := a.conn.Write(data)
-
 	defer span.Finish()
 
+	a.conn.SetWriteDeadline(time.Now().Add(a.writeTimeout))
+	_, writeErr := a.conn.Write(data)
 	if writeErr != nil {
 		tracing.LogError(span, writeErr.Error())
 		return writeErr
 	}
-
-	return nil
+	return writeErr
 }
 
 func createConnectionSpan(ctx context.Context, conn net.Conn, op string) opentracing.Span {
@@ -549,7 +563,7 @@ func createConnectionSpan(ctx context.Context, conn net.Conn, op string) opentra
 
 	tags := opentracing.Tags{
 		"span.kind": "connection",
-		"addr": remoteAddress,
+		"addr":      remoteAddress,
 	}
 
 	var parent opentracing.SpanContext