fix: a quiet cluster was treated as a dead one and reconnected
Reported by an operator whose node sends little: nothing for two minutes, then a reconnect — losing the login, the filters, and any spot that arrived during the gap. The read had a 120 s deadline and ANY error ended the session, a timeout included. That reads a silence as a verdict on the link. It is not: a cluster on a dead band legitimately says nothing for minutes. Only a real error ends a session now. A genuinely dead peer is still caught, and by the mechanism meant for it — TCP keepalive probes an idle connection and its failure surfaces as an error on Read, not as a timeout. The dialer sets it explicitly rather than inheriting a default, since it is now what the design depends on. One trap the advice this came from did not mention: ReadString returns the bytes it HAD read along with the timeout. Discarding them would lose the first half of any spot line straddling a deadline, so the partial is carried over. Tested: a listener that goes silent past the deadline and then sends a spot — the spot arrives on the ORIGINAL connection, and the listener never accepts a second one. The tick is a var so that test runs in four seconds instead of sixty; a slow test is a test that gets skipped.
This commit is contained in:
@@ -11,6 +11,7 @@ package cluster
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"regexp"
|
||||
@@ -346,12 +347,29 @@ func (s *session) run() {
|
||||
}
|
||||
}
|
||||
|
||||
// idleTick is how often the read is interrupted so the loop can notice a stop
|
||||
// request. It is NOT a liveness test — see the read loop.
|
||||
//
|
||||
// A var, not a const, so the test that proves a silent node survives can shorten
|
||||
// it: at 30 s that test spends a minute doing nothing, and a slow test is a test
|
||||
// that gets skipped.
|
||||
var idleTick = 30 * time.Second
|
||||
|
||||
// quietNotice is the silence after which the log says so, once.
|
||||
const quietNotice = 10 * time.Minute
|
||||
|
||||
// runOnce dials, optionally logs in, sends init commands, parses spots.
|
||||
// Returns the moment we marked the link "connected" (zero if dial failed)
|
||||
// and the error that ended the session (nil if stopCh).
|
||||
func (s *session) runOnce() (time.Time, error) {
|
||||
addr := net.JoinHostPort(s.cfg.Host, fmt.Sprintf("%d", s.cfg.Port)) // IPv6-safe
|
||||
conn, err := net.DialTimeout("tcp", addr, 10*time.Second)
|
||||
// KeepAlive is set explicitly rather than left to the package default: it is
|
||||
// what actually detects a dead peer here, so it should be visible in the
|
||||
// code that depends on it. The OS probes an idle connection and a genuine
|
||||
// failure surfaces as an error on Read — which is the only thing that ends
|
||||
// a session below.
|
||||
d := net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}
|
||||
conn, err := d.Dial("tcp", addr)
|
||||
if err != nil {
|
||||
return time.Time{}, fmt.Errorf("dial %s: %w", addr, err)
|
||||
}
|
||||
@@ -423,6 +441,8 @@ func (s *session) runOnce() (time.Time, error) {
|
||||
}
|
||||
|
||||
var connectedAt time.Time
|
||||
var quiet time.Duration // how long the node has said nothing
|
||||
var pending string // a line cut in half by a read deadline
|
||||
rd := bufio.NewReader(conn)
|
||||
for {
|
||||
select {
|
||||
@@ -431,11 +451,37 @@ func (s *session) runOnce() (time.Time, error) {
|
||||
default:
|
||||
}
|
||||
|
||||
_ = conn.SetReadDeadline(time.Now().Add(120 * time.Second))
|
||||
line, err := rd.ReadString('\n')
|
||||
// The deadline exists to keep this loop responsive to stopCh, NOT to
|
||||
// judge the link. A quiet node is a quiet node: some clusters send
|
||||
// nothing for minutes on a dead band, and tearing the socket down over
|
||||
// it produced a reconnect every two minutes — which loses the login,
|
||||
// the filters, and any spot that arrived during the gap.
|
||||
//
|
||||
// Only a REAL error ends the session. A dead peer still gets caught:
|
||||
// TCP keepalive probes an idle connection and its failure arrives here
|
||||
// as an error, not as a timeout.
|
||||
_ = conn.SetReadDeadline(time.Now().Add(idleTick))
|
||||
chunk, err := rd.ReadString('\n')
|
||||
if err != nil {
|
||||
var ne net.Error
|
||||
if errors.As(err, &ne) && ne.Timeout() {
|
||||
// ReadString hands back what it HAD read along with the timeout.
|
||||
// Keep it: a spot line straddling the deadline would otherwise lose
|
||||
// its first half and arrive as nonsense, or vanish entirely.
|
||||
pending += chunk
|
||||
quiet += idleTick
|
||||
// Said once at the first long silence, so a genuinely mute node is
|
||||
// visible without a line every tick.
|
||||
if quiet == quietNotice {
|
||||
applog.Printf("cluster[%s] no traffic for %s — still connected", s.cfg.Name, quiet)
|
||||
}
|
||||
continue
|
||||
}
|
||||
return connectedAt, fmt.Errorf("read: %w", err)
|
||||
}
|
||||
quiet = 0
|
||||
line := pending + chunk
|
||||
pending = ""
|
||||
line = strings.TrimRight(line, "\r\n")
|
||||
// Strip terminal BELLs (\a) and other control bytes that some nodes
|
||||
// (e.g. F5LEN) append to spot lines — "DX de …0935Z KN04\a\a" — which
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"net"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A quiet cluster must not be treated as a broken one.
|
||||
//
|
||||
// Reported by an operator whose node sends little: nothing for two minutes and
|
||||
// OpsLog reconnected — losing the login, the filters, and any spot that landed
|
||||
// during the gap. The read deadline was being read as a verdict on the link
|
||||
// rather than as a way to stay responsive to a stop request.
|
||||
//
|
||||
// The test holds the connection open, silent for longer than one deadline, then
|
||||
// sends a spot. A session that survives receives it; one that reconnects does
|
||||
// not, because this listener never accepts twice.
|
||||
func TestSilenceDoesNotEndTheSession(t *testing.T) {
|
||||
// Shorten the tick so the test exercises the real code path in seconds.
|
||||
defer func(orig time.Duration) { idleTick = orig }(idleTick)
|
||||
idleTick = time.Second
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer ln.Close()
|
||||
|
||||
accepted := make(chan net.Conn, 2)
|
||||
go func() {
|
||||
for {
|
||||
c, err := ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
accepted <- c
|
||||
}
|
||||
}()
|
||||
|
||||
host, portStr, _ := net.SplitHostPort(ln.Addr().String())
|
||||
port, err := strconv.Atoi(portStr)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
spots := make(chan Spot, 4)
|
||||
s := &session{
|
||||
cfg: ServerConfig{Name: "quiet", Host: host, Port: port},
|
||||
onSpot: func(sp Spot) { spots <- sp },
|
||||
onLine: func(Line) {},
|
||||
onStatus: func() {},
|
||||
stopCh: make(chan struct{}),
|
||||
}
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() { _, err := s.runOnce(); done <- err }()
|
||||
|
||||
conn := <-accepted
|
||||
defer conn.Close()
|
||||
|
||||
// Silent for longer than one deadline: the loop must ride through it.
|
||||
time.Sleep(3 * idleTick)
|
||||
|
||||
select {
|
||||
case c := <-accepted:
|
||||
c.Close()
|
||||
t.Fatal("the session reconnected during the silence")
|
||||
case err := <-done:
|
||||
t.Fatalf("the session ended during the silence: %v", err)
|
||||
default:
|
||||
}
|
||||
|
||||
// Now a spot: it proves the ORIGINAL connection is still the live one.
|
||||
if _, err := conn.Write([]byte("DX de F4BPO: 14074.0 OY1CT FT8 1234Z\r\n")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case sp := <-spots:
|
||||
if sp.DXCall != "OY1CT" {
|
||||
t.Errorf("spot from the wrong station: %+v", sp)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the spot sent after the silence never arrived")
|
||||
}
|
||||
|
||||
close(s.stopCh)
|
||||
<-done
|
||||
}
|
||||
Reference in New Issue
Block a user