Close idle connections [#3449]

pkg/api/v3/message/handler.go

@@ -9,6 +9,7 @@ package message
 import (
 	"context"
 	"io"
+	"time"
 
 	"github.com/libp2p/go-libp2p/core/network"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/errors"
@@ -48,21 +49,48 @@ func (h *Handler) Handle(s Stream) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	// Read requests in a goroutine
+	rd := make(chan Message)
+	go func() {
+		defer close(rd)
+
+		for {
+			// Read the next request
+			req, err := s.Read()
+			switch {
+			case err == nil:
+				// Ok
+
+			case errors.Is(err, io.EOF),
+				errors.Is(err, network.ErrReset),
+				errors.Is(err, context.Canceled):
+				// Done
+				return
+
+			default:
+				slog.Error("Unable to decode request from peer", "error", err, "module", "api")
+				return
+			}
+
+			select {
+			case rd <- req:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
 	for {
-		// Read the next request
-		req, err := s.Read()
-		switch {
-		case err == nil:
-			// Ok
-
-		case errors.Is(err, io.EOF),
-			errors.Is(err, network.ErrReset),
-			errors.Is(err, context.Canceled):
-			// Done
-			return
+		var req Message
+		var ok bool
+		select {
+		case req, ok = <-rd:
+			if !ok {
+				return
+			}
 
-		default:
-			slog.Error("Unable to decode request from peer", "error", err, "module", "api")
+		case <-time.After(10 * time.Minute):
+			// If the connection is idle for more than 10 minutes, close it
 			return
 		}
 
@@ -78,7 +106,7 @@ func (h *Handler) Handle(s Stream) {
 		// Find the method
 		m, ok := h.methods[req.Type()]
 		if !ok {
-			err = s.Write(&ErrorResponse{Error: errors.NotAllowed.WithFormat("%v not supported", req.Type())})
+			err := s.Write(&ErrorResponse{Error: errors.NotAllowed.WithFormat("%v not supported", req.Type())})
 			if err != nil {
 				slog.Error("Unable to send error response to peer", "error", err, "module", "api")
 				return