Repository: nats-io/nats-server
Pull request #1692 (2020-11-03 20:23:02)

[CHANGED] Gateway connections now always send PINGs
Connections normally suppress sending PINGs if there was some
activity. We now force GATEWAY connections to send PINGs at the
configured interval or 15 seconds, whichever is the smallest.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/server/client.go b/server/client.go
index fc3dcac8..015f1562 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3952,18 +3952,30 @@ func (c *client) processPingTimer() {
 
 	c.Debugf("%s Ping Timer", c.typeString())
 
+	var sendPing bool
+
 	// If we have had activity within the PingInterval then
 	// there is no need to send a ping. This can be client data
 	// or if we received a ping from the other side.
 	pingInterval := c.srv.getOpts().PingInterval
+	if c.kind == GATEWAY {
+		pingInterval = adjustPingIntervalForGateway(pingInterval)
+		sendPing = true
+	}
 	now := time.Now()
 	needRTT := c.rtt == 0 || now.Sub(c.rttStart) > DEFAULT_RTT_MEASUREMENT_INTERVAL
 
-	if delta := now.Sub(c.last); delta < pingInterval && !needRTT {
-		c.Debugf("Delaying PING due to client activity %v ago", delta.Round(time.Second))
-	} else if delta := now.Sub(c.ping.last); delta < pingInterval && !needRTT {
-		c.Debugf("Delaying PING due to remote ping %v ago", delta.Round(time.Second))
-	} else {
+	// Do not delay PINGs for GATEWAY connections.
+	if c.kind != GATEWAY {
+		if delta := now.Sub(c.last); delta < pingInterval && !needRTT {
+			c.Debugf("Delaying PING due to client activity %v ago", delta.Round(time.Second))
+		} else if delta := now.Sub(c.ping.last); delta < pingInterval && !needRTT {
+			c.Debugf("Delaying PING due to remote ping %v ago", delta.Round(time.Second))
+		} else {
+			sendPing = true
+		}
+	}
+	if sendPing {
 		// Check for violation
 		if c.ping.out+1 > c.srv.getOpts().MaxPingsOut {
 			c.Debugf("Stale Client Connection - Closing")
@@ -3981,12 +3993,24 @@ func (c *client) processPingTimer() {
 	c.mu.Unlock()
 }
 
+// Returns the smallest value between the given `d` and `gatewayMaxPingInterval` durations.
+// Invoked for connections known to be of GATEWAY type.
+func adjustPingIntervalForGateway(d time.Duration) time.Duration {
+	if d > gatewayMaxPingInterval {
+		return gatewayMaxPingInterval
+	}
+	return d
+}
+
 // Lock should be held
 func (c *client) setPingTimer() {
 	if c.srv == nil {
 		return
 	}
 	d := c.srv.getOpts().PingInterval
+	if c.kind == GATEWAY {
+		d = adjustPingIntervalForGateway(d)
+	}
 	c.ping.tmr = time.AfterFunc(d, c.processPingTimer)
 }
 
diff --git a/server/const.go b/server/const.go
index 4769e2d6..243fcc1a 100644
--- a/server/const.go
+++ b/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.28"
+	VERSION = "2.2.0-beta.29"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/gateway.go b/server/gateway.go
index 6b0aff05..5f7a2903 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -49,6 +49,10 @@ const (
 	gwClusterOffset  = gwReplyPrefixLen
 	gwServerOffset   = gwClusterOffset + gwHashLen + 1
 	gwSubjectOffset  = gwServerOffset + gwHashLen + 1
+
+	// Gateway connections send PINGs regardless of traffic. The interval is
+	// either Options.PingInterval or this value, whichever is the smallest.
+	gwMaxPingInterval = 15 * time.Second
 )
 
 var (
@@ -56,6 +60,7 @@ var (
 	gatewayReconnectDelay        = defaultGatewayReconnectDelay
 	gatewayMaxRUnsubBeforeSwitch = defaultGatewayMaxRUnsubBeforeSwitch
 	gatewaySolicitDelay          = int64(defaultSolicitGatewaysDelay)
+	gatewayMaxPingInterval       = gwMaxPingInterval
 )
 
 // Warning when user configures gateway TLS insecure
diff --git a/server/server.go b/server/server.go
index ef3a0eb0..59abcb83 100644
--- a/server/server.go
+++ b/server/server.go
@@ -3255,6 +3255,9 @@ func (s *Server) setFirstPingTimer(c *client) {
 			if d > firstPingInterval {
 				d = firstPingInterval
 			}
+			if c.kind == GATEWAY {
+				d = adjustPingIntervalForGateway(d)
+			}
 		} else if d > firstClientPingInterval {
 			d = firstClientPingInterval
 		}


Test patch:
diff --git a/server/gateway_test.go b/server/gateway_test.go
index d4de012f..44d4adc6 100644
--- a/server/gateway_test.go
+++ b/server/gateway_test.go
@@ -6201,3 +6201,50 @@ func TestGatewayUpdateURLsFromRemoteCluster(t *testing.T) {
 	expected[fmt.Sprintf("127.0.0.1:%d", ob3.Gateway.Port)] = "B3"
 	checkURLs(expected)
 }
+
+type capturePingConn struct {
+	net.Conn
+	ch chan struct{}
+}
+
+func (c *capturePingConn) Write(b []byte) (int, error) {
+	if bytes.Contains(b, []byte(pingProto)) {
+		select {
+		case c.ch <- struct{}{}:
+		default:
+		}
+	}
+	return c.Conn.Write(b)
+}
+
+func TestGatewayPings(t *testing.T) {
+	gatewayMaxPingInterval = 50 * time.Millisecond
+	defer func() { gatewayMaxPingInterval = gwMaxPingInterval }()
+
+	ob := testDefaultOptionsForGateway("B")
+	sb := RunServer(ob)
+	defer sb.Shutdown()
+
+	oa := testGatewayOptionsFromToWithServers(t, "A", "B", sb)
+	sa := RunServer(oa)
+	defer sa.Shutdown()
+
+	waitForInboundGateways(t, sa, 1, 2*time.Second)
+	waitForOutboundGateways(t, sa, 1, 2*time.Second)
+	waitForInboundGateways(t, sb, 1, 2*time.Second)
+	waitForOutboundGateways(t, sb, 1, 2*time.Second)
+
+	c := sa.getOutboundGatewayConnection("B")
+	ch := make(chan struct{}, 1)
+	c.mu.Lock()
+	c.nc = &capturePingConn{c.nc, ch}
+	c.mu.Unlock()
+
+	for i := 0; i < 5; i++ {
+		select {
+		case <-ch:
+		case <-time.After(250 * time.Millisecond):
+			t.Fatalf("Did not send PING")
+		}
+	}
+}