RapidAI · znsoftm · Sep 14, 2026
diff --git a/README.md b/README.md
@@ -323,7 +323,7 @@ curl -sS -H "Authorization: Bearer $TOKEN" http://PUBLIC_IP:18780/settings/confi
 # equivalently ?token= on the query string
 ```
 
-If no relay is connected, the proxy returns `503 {"error":"search backend offline","code":"offline"}`. Tunnel protocol: TCP, `AUTH <SEARCH_TOKEN>` then length-prefixed JSON request/response frames (bodies base64). Latest tunnel connection wins. Hub tokens are not used in that handshake.
+If no relay is connected, the proxy returns `503 {"error":"search backend offline","code":"offline"}`. Tunnel protocol: TCP, `AUTH <SEARCH_TOKEN>` then length-prefixed JSON request/response frames (bodies base64). Latest tunnel connection wins. Hub tokens are not used in that handshake. Both ends ping every 12s, treat a missing pong within 45s as dead, refresh a 60s read deadline on every frame (including ping/pong), enable 15s TCP keepalive, and the relay reconnects from 500ms up to 10s.
 
 Binaries: `go build -o search-proxy ./cmd/proxy` and `go build -o search-relay ./cmd/relay`. Copy them to the VPS; they are static-ish Go binaries (same module).
 

diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go
@@ -139,12 +139,10 @@ func (h *hub) handleTunnel(c net.Conn) {
 		return
 	}
 	_ = c.SetDeadline(time.Time{})
-	if tc, ok := c.(*net.TCPConn); ok {
-		_ = tc.SetKeepAlive(true)
-		_ = tc.SetKeepAlivePeriod(30 * time.Second)
-	}
+	tunnel.EnableTCPKeepAlive(c)
 
 	s := newSession(c, br)
+	s.startKeepalive()
 	h.mu.Lock()
 	old := h.sess
 	h.sess = s
@@ -153,9 +151,10 @@ func (h *hub) handleTunnel(c net.Conn) {
 		log.Printf("replacing previous tunnel from %s", old.remote)
 		old.close(errReplaced)
 	}
-	log.Printf("tunnel connected from %s", c.RemoteAddr())
+	log.Printf("tunnel connected from %s (ping=%s pong_wait=%s read_idle=%s tcp_keepalive=%s)",
+		c.RemoteAddr(), tunnel.PingInterval, tunnel.PongTimeout, tunnel.ReadIdleTimeout, tunnel.TCPKeepAlivePeriod)
 	s.readLoop()
-	log.Printf("tunnel disconnected from %s", c.RemoteAddr())
+	log.Printf("tunnel disconnected from %s: %v", c.RemoteAddr(), s.closeReason())
 	h.mu.Lock()
 	if h.sess == s {
 		h.sess = nil
@@ -397,18 +396,39 @@ type session struct {
 	mu     sync.Mutex
 	pend   map[string]chan tunnel.Frame
 	closed bool
+	onPong func()
+	stopKA func()
+	reason error
 }
 
 func newSession(c net.Conn, br *bufio.Reader) *session {
-	return &session{conn: c, br: br, remote: c.RemoteAddr().String(), pend: map[string]chan tunnel.Frame{}}
+	return &session{
+		conn:   c,
+		br:     br,
+		remote: c.RemoteAddr().String(),
+		pend:   map[string]chan tunnel.Frame{},
+		onPong: func() {},
+		stopKA: func() {},
+	}
+}
+
+func (s *session) startKeepalive() {
+	s.startKeepaliveCfg(tunnel.DefaultKeepalive())
+}
+
+func (s *session) startKeepaliveCfg(cfg tunnel.KeepaliveConfig) {
+	s.onPong, s.stopKA = tunnel.StartKeepalive(context.Background(), cfg, s.write, func(err error) {
+		log.Printf("tunnel keepalive dead from %s: %v", s.remote, err)
+		s.close(err)
+	})
 }
 
 func (s *session) write(f tunnel.Frame) error {
 	s.wmu.Lock()
 	defer s.wmu.Unlock()
-	_ = s.conn.SetWriteDeadline(time.Now().Add(30 * time.Second))
+	_ = tunnel.SetWriteIdle(s.conn)
 	err := tunnel.WriteFrame(s.conn, f)
-	_ = s.conn.SetWriteDeadline(time.Time{})
+	tunnel.ClearWriteDeadline(s.conn)
 	return err
 }
 
@@ -437,19 +457,28 @@ func (s *session) roundTrip(ctx context.Context, f tunnel.Frame) (tunnel.Frame,
 	}
 }
 
+func (s *session) closeReason() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.reason != nil {
+		return s.reason
+	}
+	return io.EOF
+}
+
 func (s *session) readLoop() {
 	defer s.close(io.EOF)
 	for {
-		_ = s.conn.SetReadDeadline(time.Now().Add(90 * time.Second))
 		var f tunnel.Frame
-		if err := tunnel.ReadFrame(s.br, &f); err != nil {
+		if err := tunnel.ReadFrameRefreshing(s.conn, s.br, &f); err != nil {
+			s.close(tunnel.ClassifyReadError(err))
 			return
 		}
 		switch f.Type {
-		case tunnel.TypePong, tunnel.TypePing:
-			if f.Type == tunnel.TypePing {
-				_ = s.write(tunnel.Frame{Type: tunnel.TypePong, ID: f.ID})
-			}
+		case tunnel.TypePong:
+			s.onPong()
+		case tunnel.TypePing:
+			_ = s.write(tunnel.Frame{Type: tunnel.TypePong, ID: f.ID})
 		case tunnel.TypeResp, tunnel.TypeRespHead, tunnel.TypeRespChunk, tunnel.TypeRespEnd:
 			s.mu.Lock()
 			ch := s.pend[f.ID]
@@ -475,14 +504,21 @@ func (s *session) close(err error) {
 		return
 	}
 	s.closed = true
+	if err != nil {
+		s.reason = err
+	}
 	for id, ch := range s.pend {
 		select {
 		case ch <- tunnel.Frame{Type: "resp", ID: id, Status: 503, Error: "offline"}:
 		default:
 		}
 		delete(s.pend, id)
 	}
+	stopKA := s.stopKA
 	s.mu.Unlock()
+	if stopKA != nil {
+		stopKA()
+	}
 	_ = s.conn.Close()
 }
 

diff --git a/cmd/proxy/proxy_test.go b/cmd/proxy/proxy_test.go
@@ -1,12 +1,18 @@
 package main
 
 import (
+	"bufio"
 	"encoding/json"
+	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
+	"sync"
 	"testing"
+	"time"
 
 	"search-service/internal/proxyauth"
+	"search-service/internal/tunnel"
 )
 
 func testProxy(t *testing.T) *hub {
@@ -71,3 +77,82 @@ func TestProxyPapersSkipsBearerSoLoginCanRender(t *testing.T) {
 		}
 	}
 }
+
+func TestSessionEmitsAndAnswersPing(t *testing.T) {
+	client, server := net.Pipe()
+	defer client.Close()
+	defer server.Close()
+
+	s := newSession(server, bufio.NewReader(server))
+	s.startKeepaliveCfg(tunnel.KeepaliveConfig{
+		Interval: 30 * time.Millisecond,
+		PongWait: 400 * time.Millisecond,
+	})
+	done := make(chan struct{})
+	go func() {
+		s.readLoop()
+		close(done)
+	}()
+
+	var mu sync.Mutex
+	sawPong := false
+	sawPing := false
+	got := make(chan struct{}, 4)
+	go func() {
+		for {
+			_ = client.SetReadDeadline(time.Now().Add(2 * time.Second))
+			var f tunnel.Frame
+			if err := tunnel.ReadFrame(client, &f); err != nil {
+				return
+			}
+			switch f.Type {
+			case tunnel.TypePong:
+				if f.ID == "peer" {
+					mu.Lock()
+					sawPong = true
+					mu.Unlock()
+					got <- struct{}{}
+				}
+			case tunnel.TypePing:
+				_ = tunnel.WriteFrame(client, tunnel.Frame{Type: tunnel.TypePong, ID: f.ID})
+				mu.Lock()
+				first := !sawPing
+				sawPing = true
+				mu.Unlock()
+				if first {
+					got <- struct{}{}
+				}
+			}
+		}
+	}()
+
+	select {
+	case <-got:
+	case <-time.After(2 * time.Second):
+		t.Fatal("proxy should emit its own ping")
+	}
+	if err := tunnel.WriteFrame(client, tunnel.Frame{Type: tunnel.TypePing, ID: "peer"}); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-got:
+	case <-time.After(2 * time.Second):
+		t.Fatal("proxy should answer peer ping")
+	}
+	mu.Lock()
+	okPing, okPong := sawPing, sawPong
+	mu.Unlock()
+	if !okPing {
+		t.Fatal("proxy should emit its own ping")
+	}
+	if !okPong {
+		t.Fatal("proxy should answer peer ping")
+	}
+	s.close(io.EOF)
+	_ = client.Close()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("readLoop did not exit")
+	}
+}
diff --git a/cmd/relay/main.go b/cmd/relay/main.go
@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"io"
 	"log"
-	"net"
 	"net/http"
 	"net/url"
 	"os"
@@ -45,31 +44,28 @@ func main() {
 		},
 	}
 
-	log.Printf("search-relay backend=%s tunnel=%s", backend, tun)
-	backoff := time.Second
+	log.Printf("search-relay backend=%s tunnel=%s keepalive ping=%s pong_wait=%s read_idle=%s tcp_keepalive=%s",
+		backend, tun, tunnel.PingInterval, tunnel.PongTimeout, tunnel.ReadIdleTimeout, tunnel.TCPKeepAlivePeriod)
+	backoff := tunnel.ReconnectMin
 	for ctx.Err() == nil {
 		connectedAt := time.Now()
 		err := runOnce(ctx, tun, token, backend, client)
 		if ctx.Err() != nil {
 			break
 		}
-		if time.Since(connectedAt) > 10*time.Second {
-			backoff = time.Second
+		alive := time.Since(connectedAt)
+		if alive > tunnel.HealthyResetAfter {
+			backoff = tunnel.ReconnectMin
 		}
-		log.Printf("tunnel dropped: %v; reconnect in %s", err, backoff)
+		log.Printf("tunnel reconnect: reason=%v alive=%s backoff=%s", err, alive.Truncate(time.Millisecond), backoff)
 		t := time.NewTimer(backoff)
 		select {
 		case <-ctx.Done():
 			t.Stop()
 			return
 		case <-t.C:
 		}
-		if backoff < 30*time.Second {
-			backoff *= 2
-			if backoff > 30*time.Second {
-				backoff = 30 * time.Second
-			}
-		}
+		backoff = tunnel.NextReconnectBackoff(backoff)
 	}
 }
 
@@ -106,12 +102,13 @@ func parseTunnelAddr(s string) (string, error) {
 }
 
 func runOnce(ctx context.Context, addr, token, backend string, client *http.Client) error {
-	d := net.Dialer{Timeout: 15 * time.Second, KeepAlive: 30 * time.Second}
+	d := tunnel.Dialer(15 * time.Second)
 	c, err := d.DialContext(ctx, "tcp", addr)
 	if err != nil {
 		return err
 	}
 	defer c.Close()
+	tunnel.EnableTCPKeepAlive(c)
 
 	_ = c.SetDeadline(time.Now().Add(15 * time.Second))
 	if _, err := c.Write([]byte("AUTH " + token + "\n")); err != nil {
@@ -132,61 +129,62 @@ func runOnce(ctx context.Context, addr, token, backend string, client *http.Clie
 	write := func(f tunnel.Frame) error {
 		wmu.Lock()
 		defer wmu.Unlock()
-		_ = c.SetWriteDeadline(time.Now().Add(30 * time.Second))
+		_ = tunnel.SetWriteIdle(c)
 		err := tunnel.WriteFrame(c, f)
-		_ = c.SetWriteDeadline(time.Time{})
+		tunnel.ClearWriteDeadline(c)
 		return err
 	}
 
-	stopPing := make(chan struct{})
-	go func() {
-		t := time.NewTicker(25 * time.Second)
-		defer t.Stop()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-stopPing:
-				return
-			case <-t.C:
-				if err := write(tunnel.Frame{Type: "ping"}); err != nil {
-					_ = c.Close()
-					return
-				}
-			}
+	var closeOnce sync.Once
+	closeConn := func() { closeOnce.Do(func() { _ = c.Close() }) }
+	defer closeConn()
+
+	errCh := make(chan error, 1)
+	die := func(err error) {
+		select {
+		case errCh <- err:
+		default:
 		}
-	}()
-	defer func() { close(stopPing); _ = c.Close() }()
+		closeConn()
+	}
+
+	onPong, stopKA := tunnel.StartKeepalive(ctx, tunnel.DefaultKeepalive(), write, die)
+	defer stopKA()
 
 	for {
-		select {
-		case <-ctx.Done():
+		if ctx.Err() != nil {
 			return ctx.Err()
-		default:
 		}
-		_ = c.SetReadDeadline(time.Now().Add(90 * time.Second))
 		var f tunnel.Frame
-		if err := tunnel.ReadFrame(br, &f); err != nil {
-			return err
+		if err := tunnel.ReadFrameRefreshing(c, br, &f); err != nil {
+			select {
+			case kerr := <-errCh:
+				return tunnel.ClassifyReadError(kerr)
+			default:
+			}
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			return tunnel.ClassifyReadError(err)
 		}
 		switch f.Type {
-		case "ping":
-			_ = write(tunnel.Frame{Type: "pong", ID: f.ID})
-		case "pong":
-			// keepalive
+		case tunnel.TypePing:
+			_ = write(tunnel.Frame{Type: tunnel.TypePong, ID: f.ID})
+		case tunnel.TypePong:
+			onPong()
 		case tunnel.TypeReq:
 			go func(f tunnel.Frame) {
 				if tunnel.PathNeedsStream(f.Path) {
 					if err := streamBackend(ctx, client, backend, f, write); err != nil {
 						log.Printf("stream download id=%s: %v", f.ID, err)
-						_ = c.Close()
+						die(fmt.Errorf("stream write: %w", err))
 					}
 					return
 				}
 				resp := doBackend(ctx, client, backend, f)
 				if err := write(resp); err != nil {
 					log.Printf("write resp id=%s: %v", f.ID, err)
-					_ = c.Close()
+					die(fmt.Errorf("resp write: %w", err))
 				}
 			}(f)
 		}