VC: Remote BN clock offset monitoring. (#4846)

* Initial commit.

* Add algorithm in comment.
Remove delays.
Fix logging statement issues.
Change update from epoch to slot.

* Obtain timestamp earlier.

* Add processing delays into algorithm.

* Fix time offset logging to produce integers instead of strings.

* Address review comments.

* Fix copyright year.
Fix updateStatus().

* Remove fields from Slot start log statement.
Fix issues when BN do not support Nimbus Extensions.
Rename metric name and type change.

* Add beacon role to disable time offset check manually.

beacon_chain/beacon_node.nim

@@ -96,6 +96,7 @@ type
       Table[ValidatorPubKey, SignedValidatorRegistrationV1]
     dutyValidatorCount*: int
       ## Number of validators that we've checked for activation
+    processingDelay*: Opt[Duration]
 
 const
   MaxEmptySlotCount* = uint64(10*60) div SECONDS_PER_SLOT

beacon_chain/nimbus_beacon_node.nim

@@ -1310,6 +1310,8 @@ proc onSlotStart(node: BeaconNode, wallTime: BeaconTime,
     finalizedEpoch = node.dag.finalizedHead.blck.slot.epoch()
     delay = wallTime - expectedSlot.start_beacon_time()
 
+  node.processingDelay = Opt.some(nanoseconds(delay.nanoseconds))
+
   info "Slot start",
     slot = shortLog(wallSlot),
     epoch = shortLog(wallSlot.epoch),

beacon_chain/nimbus_validator_client.nim

@@ -177,6 +177,8 @@ proc runVCSlotLoop(vc: ValidatorClientRef) {.async.} =
     if checkIfShouldStopAtEpoch(wallSlot, vc.config.stopAtEpoch):
       return
 
+    vc.processingDelay = Opt.some(nanoseconds(delay.nanoseconds))
+
     if len(vc.beaconNodes) > 1:
       let
         counts = vc.getNodeCounts()
@@ -197,7 +199,8 @@ proc runVCSlotLoop(vc: ValidatorClientRef) {.async.} =
         blockIn = vc.getDurationToNextBlock(wallSlot),
         validators = vc.attachedValidators[].count(),
         good_nodes = goodNodes, viable_nodes = viableNodes,
-        bad_nodes = badNodes, delay = shortLog(delay)
+        bad_nodes = badNodes,
+        delay = shortLog(delay)
     else:
       info "Slot start",
         slot = shortLog(wallSlot),

beacon_chain/rpc/rest_constants.nim

@@ -241,3 +241,6 @@ const
     "The given merkle proof index is invalid"
   FailedToObtainForkError* =
     "Failed to obtain fork information"
+  InvalidTimestampValue* =
+    "Invalid or missing timestamp value"
+

beacon_chain/rpc/rest_nimbus_api.nim

@@ -428,3 +428,27 @@ proc installNimbusApiHandlers*(router: var RestRouter, node: BeaconNode) =
         all_peers: allPeers
       )
     )
+
+  router.api(MethodPost, "/nimbus/v1/timesync") do (
+    contentBody: Option[ContentBody]) -> RestApiResponse:
+    let
+      timestamp2 = getTimestamp()
+      timestamp1 =
+        block:
+          if contentBody.isNone():
+            return RestApiResponse.jsonError(Http400, EmptyRequestBodyError)
+          let dres = decodeBody(RestNimbusTimestamp1, contentBody.get())
+          if dres.isErr():
+            return RestApiResponse.jsonError(Http400,
+                                             InvalidTimestampValue,
+                                             $dres.error())
+          dres.get().timestamp1
+    let
+      delay = node.processingDelay.valueOr: ZeroDuration
+      response = RestNimbusTimestamp2(
+        timestamp1: timestamp1,
+        timestamp2: timestamp2,
+        timestamp3: getTimestamp(),
+        delay: uint64(delay.nanoseconds)
+      )
+    return RestApiResponse.jsonResponsePlain(response)

beacon_chain/rpc/rest_utils.nim

@@ -10,15 +10,15 @@
 import std/[options, macros],
        stew/byteutils, presto,
        ../spec/[forks],
-       ../spec/eth2_apis/[rest_types, eth2_rest_serialization],
+       ../spec/eth2_apis/[rest_types, eth2_rest_serialization, rest_common],
        ../validators/validator_duties,
        ../consensus_object_pools/blockchain_dag,
        ../beacon_node,
        "."/[rest_constants, state_ttl_cache]
 
 export
   options, eth2_rest_serialization, blockchain_dag, presto, rest_types,
-  rest_constants
+  rest_constants, rest_common
 
 type
   ValidatorIndexError* {.pure.} = enum

beacon_chain/spec/eth2_apis/eth2_rest_serialization.nim

@@ -104,7 +104,8 @@ type
     capella_mev.SignedBlindedBeaconBlock |
     SignedValidatorRegistrationV1 |
     SignedVoluntaryExit |
-    Web3SignerRequest
+    Web3SignerRequest |
+    RestNimbusTimestamp1
 
   EncodeOctetTypes* =
     altair.SignedBeaconBlock |
@@ -151,7 +152,9 @@ type
     GetStateRootResponse |
     GetBlockRootResponse |
     SomeForkedLightClientObject |
-    seq[SomeForkedLightClientObject]
+    seq[SomeForkedLightClientObject] |
+    RestNimbusTimestamp1 |
+    RestNimbusTimestamp2
 
   DecodeConsensysTypes* =
     ProduceBlockResponseV2 | ProduceBlindedBlockResponse

beacon_chain/spec/eth2_apis/rest_common.nim

@@ -10,6 +10,9 @@ import
   chronos, presto/client,
   "."/[rest_types, eth2_rest_serialization]
 
+from std/times import Time, DateTime, toTime, fromUnix, now, utc, `-`,
+                      inNanoseconds
+
 export chronos, client, rest_types, eth2_rest_serialization
 
 proc raiseGenericError*(resp: RestPlainResponse) {.
@@ -52,3 +55,6 @@ proc getBodyBytesWithCap*(
     if not(isNil(reader)):
       await reader.closeWait()
     raise newHttpReadError("Could not read response")
+
+proc getTimestamp*(): uint64 =
+  uint64((toTime(now().utc) - fromUnix(0)).inNanoseconds())

beacon_chain/spec/eth2_apis/rest_nimbus_calls.nim

@@ -1,4 +1,4 @@
-# Copyright (c) 2018-2022 Status Research & Development GmbH
+# Copyright (c) 2018-2023 Status Research & Development GmbH
 # Licensed and distributed under either of
 #   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
 #   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
@@ -7,11 +7,71 @@
 {.push raises: [Defect].}
 
 import
-  chronos, presto/client,
-  "."/[rest_types, eth2_rest_serialization]
+  chronos, chronicles, presto/client,
+  "."/[rest_types, eth2_rest_serialization, rest_common]
 
 proc getValidatorsActivity*(epoch: Epoch,
                             body: seq[ValidatorIndex]
                            ): RestPlainResponse {.
      rest, endpoint: "/nimbus/v1/validator/activity/{epoch}",
      meth: MethodPost.}
+
+proc getTimesyncInifo*(body: RestNimbusTimestamp1): RestPlainResponse {.
+     rest, endpoint: "/nimbus/v1/timesync", meth: MethodPost.}
+
+proc getTimeOffset*(client: RestClientRef,
+                    delay: Duration): Future[int64] {.async.} =
+  let
+    timestamp1 = getTimestamp()
+    data = RestNimbusTimestamp1(timestamp1: timestamp1)
+    resp = await client.getTimesyncInifo(data)
+    timestamp4 = getTimestamp()
+
+  return
+    case resp.status
+    of 200:
+      if resp.contentType.isNone() or
+         isWildCard(resp.contentType.get().mediaType) or
+         resp.contentType.get().mediaType != ApplicationJsonMediaType:
+        raise newException(RestError, "Missing or incorrect Content-Type")
+
+      let stamps = decodeBytes(RestNimbusTimestamp2, resp.data,
+                               resp.contentType).valueOr:
+        raise newException(RestError, $error)
+
+      trace "Time offset data",
+            timestamp1 = timestamp1,
+            timestamp2 = stamps.timestamp2,
+            timestamp3 = stamps.timestamp3,
+            timestamp4 = timestamp4,
+            delay14 = delay.nanoseconds,
+            delay23 = stamps.delay
+
+      # t1 - time when we sent request.
+      # t2 - time when remote server received request.
+      # t3 - time when remote server sent response.
+      # t4 - time when we received response.
+      # delay14 = validator client processing delay.
+      # delay23 = beacon node processing delay.
+      #
+      # Round-trip network delay `delta` = (t4 - t1) - (t3 - t2)
+      # but with delays this will be:
+      # `delta` = (t4 - t1 + delay14) - (t3 - t2 + delay23)
+      # Estimated server time is t3 + (delta div 2)
+      # Estimated clock skew `theta` = t3 + (delta div 2) - t4
+      let
+        delay14 = delay.nanoseconds
+        delay23 = int64(stamps.delay)
+        offset = (int64(stamps.timestamp2) - int64(timestamp1) +
+                  int64(stamps.timestamp3) - int64(timestamp4) +
+                  delay14 - delay23) div 2
+      offset
+    else:
+      let error = decodeBytes(RestErrorMessage, resp.data,
+                              resp.contentType).valueOr:
+        let msg = "Incorrect response error format (" & $resp.status &
+                  ") [" & $error & "]"
+        raise (ref RestResponseError)(msg: msg, status: resp.status)
+      let msg = "Error response (" & $resp.status & ") [" & error.message & "]"
+      raise (ref RestResponseError)(
+        msg: msg, status: error.code, message: error.message)

beacon_chain/spec/eth2_apis/rest_types.nim

@@ -657,6 +657,15 @@ type
   RestRoot* = object
     root*: Eth2Digest
 
+  RestNimbusTimestamp1* = object
+    timestamp1*: uint64
+
+  RestNimbusTimestamp2* = object
+    timestamp1*: uint64
+    timestamp2*: uint64
+    timestamp3*: uint64
+    delay*: uint64
+
   # Types based on the OAPI yaml file - used in responses to requests
   GetBeaconHeadResponse* = DataEnclosedObject[Slot]
   GetAggregatedAttestationResponse* = DataEnclosedObject[Attestation]