rpc: enforce max clock offset across nodes in a cluster

Commits suicide if offset is different than any contacted node. In order to
change the max clock offset, the cluster must be stopped and restarted.

rpc/context.go

@@ -285,7 +285,10 @@ func (ctx *Context) IsConnHealthy(remoteAddr string) bool {
 }
 
 func (ctx *Context) runHeartbeat(cc *grpc.ClientConn, remoteAddr string) error {
-	request := PingRequest{Addr: ctx.Addr}
+	request := PingRequest{
+		Addr:           ctx.Addr,
+		MaxOffsetNanos: ctx.localClock.MaxOffset().Nanoseconds(),
+	}
 	heartbeatClient := NewHeartbeatClient(cc)
 
 	var heartbeatTimer timeutil.Timer

rpc/heartbeat.go

@@ -62,6 +62,13 @@ type HeartbeatService struct {
 // The requester should also estimate its offset from this server along
 // with the requester's address.
 func (hs *HeartbeatService) Ping(ctx context.Context, args *PingRequest) (*PingResponse, error) {
+	// Enforce that clock max offsets are identical between nodes.
+	// Commit suicide in the event that this is ever untrue.
+	// This check is ignored if either offset is set to 0 (for unittests).
+	mo, amo := hs.clock.MaxOffset(), time.Duration(args.MaxOffsetNanos)
+	if mo != 0 && amo != 0 && mo != amo {
+		panic(fmt.Sprintf("clock max offset mismatch node=%s vs %s=%s", mo, args.Addr, amo))
+	}
 	serverOffset := args.Offset
 	// The server offset should be the opposite of the client offset.
 	serverOffset.Offset = -serverOffset.Offset

rpc/heartbeat.proto

@@ -49,6 +49,8 @@ message PingRequest {
   optional RemoteOffset offset = 2 [(gogoproto.nullable) = false];
   // The address of the client.
   optional string addr = 3 [(gogoproto.nullable) = false];
+  // The configured maximum clock offset (in nanoseconds) on the server.
+  optional int64 max_offset_nanos = 4 [(gogoproto.nullable) = false];
 }
 
 // A PingResponse contains the echoed ping request string.

rpc/heartbeat_test.go

@@ -17,6 +17,8 @@
 package rpc
 
 import (
+	"fmt"
+	"regexp"
 	"testing"
 	"time"
 
@@ -103,3 +105,31 @@ func TestManualHeartbeat(t *testing.T) {
 			manualResponse.ServerTime, regularResponse.ServerTime)
 	}
 }
+
+func TestClockOffsetMismatch(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer func() {
+		if r := recover(); r != nil {
+			fmt.Println(r)
+			if match, _ := regexp.MatchString("clock max offset mismatch", r.(string)); !match {
+				t.Errorf("expected clock mismatch error")
+			}
+		}
+	}()
+
+	clock := hlc.NewClock(hlc.UnixNano)
+	clock.SetMaxOffset(250 * time.Millisecond)
+	hs := &HeartbeatService{
+		clock:              clock,
+		remoteClockMonitor: newRemoteClockMonitor(context.TODO(), clock, time.Hour),
+	}
+
+	request := &PingRequest{
+		Ping:           "testManual",
+		Addr:           "test",
+		MaxOffsetNanos: (500 * time.Millisecond).Nanoseconds(),
+	}
+	ctx := context.Background()
+	_, _ = hs.Ping(ctx, request)
+	t.Fatal("should not reach")
+}