add write timeout, in case peer stalled

chotki.go

@@ -84,8 +84,8 @@ type Options struct {
 	ReadMinBufferSizeToProcess int
 	TcpReadBufferSize          int
 	TcpWriteBufferSize         int
-
-	TlsConfig *tls.Config
+	WriteTimeout               time.Duration
+	TlsConfig                  *tls.Config
 }
 
 func (o *Options) SetDefaults() {
@@ -119,6 +119,10 @@ func (o *Options) SetDefaults() {
 		o.ReadAccumTimeLimit = 5 * time.Second
 	}
 
+	if o.WriteTimeout == 0 {
+		o.WriteTimeout = 5 * time.Minute
+	}
+
 	o.Merger = &pebble.Merger{
 		Name: "CRDT",
 		Merge: func(key, value []byte) (pebble.ValueMerger, error) {
@@ -262,6 +266,7 @@ func Open(dirname string, opts Options) (*Chotki, error) {
 			BufferMinToProcess: cho.opts.ReadMinBufferSizeToProcess,
 		},
 		&protocol.TcpBufferSizeOpt{Read: cho.opts.TcpReadBufferSize, Write: cho.opts.TcpWriteBufferSize},
+		&protocol.NetWriteTimeoutOpt{Timeout: cho.opts.WriteTimeout},
 	)
 
 	if !exists {

protocol/net.go

@@ -57,6 +57,7 @@ type Net struct {
 	readBufferTcpSize  int
 	writeBufferTcpSize int
 	readAccumTimeLimit time.Duration
+	writeTimeout       time.Duration
 	bufferMaxSize      int
 	bufferMinToProcess int
 }
@@ -65,6 +66,14 @@ type NetOpt interface {
 	Apply(*Net)
 }
 
+type NetWriteTimeoutOpt struct {
+	Timeout time.Duration
+}
+
+func (opt *NetWriteTimeoutOpt) Apply(n *Net) {
+	n.writeTimeout = opt.Timeout
+}
+
 type NetTlsConfigOpt struct {
 	Config *tls.Config
 }
@@ -315,6 +324,7 @@ func (n *Net) keepPeer(name string, conn net.Conn) {
 	peer := &Peer{
 		inout:               n.onInstall(name),
 		conn:                conn,
+		writeTimeout:        n.writeTimeout,
 		readAccumtTimeLimit: n.readAccumTimeLimit,
 		bufferMaxSize:       n.bufferMaxSize,
 		bufferMinToProcess:  n.bufferMinToProcess,

protocol/net_test.go

@@ -80,15 +80,15 @@ func TestTCPDepot_Connect(t *testing.T) {
 	lCon := utils.NewFDQueue[Records](16, time.Millisecond, 0)
 	l := NewNet(log, func(_ string) FeedDrainCloserTraced {
 		return &TracedQueue[Records, []byte]{lCon}
-	}, func(_ string, t Traced) { lCon.Close() }, &NetTlsConfigOpt{tlsConfig("a.chotki.local")})
+	}, func(_ string, t Traced) { lCon.Close() }, &NetTlsConfigOpt{tlsConfig("a.chotki.local")}, &NetWriteTimeoutOpt{Timeout: 1 * time.Minute})
 
 	err := l.Listen(loop)
 	assert.Nil(t, err)
 
 	cCon := utils.NewFDQueue[Records](16, time.Millisecond, 0)
 	c := NewNet(log, func(_ string) FeedDrainCloserTraced {
 		return &TracedQueue[Records, []byte]{cCon}
-	}, func(_ string, t Traced) { cCon.Close() }, &NetTlsConfigOpt{tlsConfig("b.chotki.local")})
+	}, func(_ string, t Traced) { cCon.Close() }, &NetTlsConfigOpt{tlsConfig("b.chotki.local")}, &NetWriteTimeoutOpt{Timeout: 1 * time.Minute})
 
 	err = c.Connect(loop)
 	assert.Nil(t, err)

protocol/peer.go

@@ -26,6 +26,7 @@ type Peer struct {
 	readAccumtTimeLimit time.Duration
 	bufferMaxSize       int
 	bufferMinToProcess  int
+	writeTimeout        time.Duration
 }
 
 func (p *Peer) getReadTimeLimit() time.Duration {
@@ -148,6 +149,9 @@ func (p *Peer) keepWrite(ctx context.Context) error {
 		p.writeBatchSize.Add(float64(batchSize))
 
 		b := net.Buffers(recs)
+		if p.writeTimeout != 0 {
+			p.conn.SetWriteDeadline(time.Now().Add(p.writeTimeout))
+		}
 		for len(b) > 0 && err == nil {
 			if _, err = b.WriteTo(p.conn); err != nil {
 				return err