Halfpipe Timeout Handling (#152)

* reset conn timeouts on a write in either direction

* lint fixes

Author: jmwample

application/lib/proxies.go

@@ -105,6 +105,19 @@ func halfPipe(src net.Conn, dst net.Conn,
 				}
 				break
 			}
+
+			// refresh stall timeout - set both because it only happens on write
+			// so if connection is sending traffic unidirectionally we prevent
+			// the receiving side from timing out.
+			err := src.SetDeadline(time.Now().Add(proxyStallTimeout))
+			if err != nil {
+				logger.Errorln("error setting deadline for src conn: ", tag)
+			}
+			err = dst.SetDeadline(time.Now().Add(proxyStallTimeout))
+			if err != nil {
+				logger.Errorln("error setting deadline for dst conn: ", tag)
+			}
+
 		}
 		return totWritten, err
 

application/lib/proxies_test.go

@@ -7,13 +7,17 @@ package lib
 import (
 	"bytes"
 	"errors"
+	"fmt"
+	"io"
 	"net"
 	"os"
 	"sync"
 	"syscall"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/refraction-networking/conjure/application/log"
 )
 
@@ -114,3 +118,154 @@ func (m *mockConn) SetReadDeadline(t time.Time) error {
 func (m *mockConn) SetWriteDeadline(t time.Time) error {
 	return errNotExist
 }
+
+func TestHalfpipeDeadlineEcho(t *testing.T) {
+	if os.Getenv("HALFPIPE") == "" {
+		t.Skip("Skipping slow tests involving halfpipe timeouts")
+	}
+
+	clientClient, clientStation := net.Pipe()
+	stationCovert, covertCovert := net.Pipe()
+
+	logger := log.New(os.Stdout, "", 0)
+	logger.SetLevel(log.TraceLevel)
+	wg := sync.WaitGroup{}
+	oncePrintErr := sync.Once{}
+	wg.Add(2)
+
+	go halfPipe(clientStation, stationCovert, &wg, &oncePrintErr, logger, "Up "+"XXXXXX")
+	go halfPipe(stationCovert, clientStation, &wg, &oncePrintErr, logger, "Down "+"XXXXXX")
+
+	go func() {
+		defer covertCovert.Close()
+		_, _ = io.Copy(covertCovert, covertCovert)
+	}()
+
+	start := time.Now()
+	for i := 0; i < 10; i++ {
+
+		_, err := clientClient.Write([]byte(fmt.Sprintf("%d", i)))
+		if err != nil {
+			t.Fatalf("received '%v' at client", err)
+		}
+
+		b := make([]byte, 10)
+		n, err := clientClient.Read(b)
+		if errors.Is(err, io.EOF) {
+			t.Fatalf("received EOF at client")
+		} else if e, ok := err.(net.Error); ok && e.Timeout() {
+			t.Fatalf("received Timeout at client")
+		} else if err != nil {
+			t.Fatalf("received '%v' at client", err)
+		}
+
+		t.Logf("%s, %d - %s", time.Since(start), n, string(b))
+
+		time.Sleep(4 * time.Second)
+	}
+
+	clientClient.Close()
+	wg.Wait()
+}
+
+func TestHalfpipeDeadlineUpload(t *testing.T) {
+	if os.Getenv("HALFPIPE") == "" {
+		t.Skip("Skipping slow tests involving halfpipe timeouts")
+	}
+
+	clientClient, clientStation := net.Pipe()
+	stationCovert, covertCovert := net.Pipe()
+
+	logger := log.New(os.Stdout, "", 0)
+	logger.SetLevel(log.TraceLevel)
+	wg := sync.WaitGroup{}
+	oncePrintErr := sync.Once{}
+	wg.Add(2)
+
+	go halfPipe(clientStation, stationCovert, &wg, &oncePrintErr, logger, "Up "+"XXXXXX")
+	go halfPipe(stationCovert, clientStation, &wg, &oncePrintErr, logger, "Down "+"XXXXXX")
+
+	go func() {
+		defer covertCovert.Close()
+		_, _ = io.Copy(io.Discard, covertCovert)
+	}()
+
+	start := time.Now()
+	for i := 0; i < 10; i++ {
+
+		n, err := clientClient.Write([]byte(fmt.Sprintf("%d", i)))
+		if errors.Is(err, io.EOF) {
+			t.Fatalf("received EOF at client")
+		} else if e, ok := err.(net.Error); ok && e.Timeout() {
+			t.Fatalf("received Timeout at client")
+		} else if err != nil {
+			t.Fatalf("received '%v' at client", err)
+		}
+
+		t.Logf("%s, %d %d", time.Since(start), n, i)
+
+		time.Sleep(4 * time.Second)
+	}
+
+	clientClient.Close()
+	covertCovert.Close()
+	wg.Wait()
+}
+
+// Test that we actually timeout after one side (client) stalls too long.
+func TestHalfpipeDeadlineActual(t *testing.T) {
+	if os.Getenv("HALFPIPE") == "" {
+		t.Skip("Skipping slow tests involving halfpipe timeouts")
+	}
+
+	clientClient, clientStation := net.Pipe()
+	stationCovert, covertCovert := net.Pipe()
+
+	logger := log.New(os.Stdout, "", 0)
+	logger.SetLevel(log.TraceLevel)
+	wg := sync.WaitGroup{}
+	oncePrintErr := sync.Once{}
+	wg.Add(2)
+
+	go halfPipe(clientStation, stationCovert, &wg, &oncePrintErr, logger, "Up "+"XXXXXX")
+	go halfPipe(stationCovert, clientStation, &wg, &oncePrintErr, logger, "Down "+"XXXXXX")
+
+	var serverErr error
+	go func() {
+		defer covertCovert.Close()
+		for {
+			b := make([]byte, 10)
+			_, serverErr = covertCovert.Read(b)
+			if serverErr != nil {
+				return
+			}
+		}
+	}()
+
+	start := time.Now()
+	for i := 0; i < 3; i++ {
+
+		n, err := clientClient.Write([]byte(fmt.Sprintf("%d", i)))
+		if errors.Is(err, io.EOF) {
+			t.Fatalf("received EOF at client")
+		} else if e, ok := err.(net.Error); ok && e.Timeout() {
+			t.Fatalf("received Timeout at client")
+		} else if err != nil {
+			t.Fatalf("received '%v' at client", err)
+		}
+
+		t.Logf("%s, %d %d", time.Since(start), n, i)
+
+		time.Sleep(4 * time.Second)
+	}
+
+	// sleep 27 + 4 = 31 > proxyStallTimeout
+	time.Sleep(27 * time.Second)
+
+	// covertStation will Timeout and send an EOF to covertCovert
+	require.ErrorIs(t, io.EOF, serverErr)
+
+	clientClient.Close()
+	covertCovert.Close()
+	wg.Wait()
+}