[WIP] Block by Block Synchronous Sate Sync

build/Dockerfile.debian.dev

@@ -54,7 +54,6 @@ RUN make protogen_local && \
 RUN go get -d -v ./app/pocket
 RUN go build -o /usr/local/bin/pocket ./app/pocket
 RUN go build -tags=debug -o /usr/local/bin/p1 ./app/client
-
 RUN go build -o /usr/local/bin/cluster-manager ./build/localnet/cluster-manager
 
 CMD ["/usr/local/bin/pocket"]

build/config/config.validator1.json

@@ -56,13 +56,14 @@
   },
   "servicer": {
     "enabled": true,
+    "private_key": "0ca1a40ddecdab4f5b04fa0bfed1d235beaa2b8082e7554425607516f0862075dfe357de55649e6d2ce889acf15eb77e94ab3c5756fe46d3c7538d37f27f115e",
     "chains": ["0001"]
   },
   "ibc": {
     "enabled": true,
     "stores_dir": "/var/ibc",
     "host": {
-        "private_key": "0ca1a40ddecdab4f5b04fa0bfed1d235beaa2b8082e7554425607516f0862075dfe357de55649e6d2ce889acf15eb77e94ab3c5756fe46d3c7538d37f27f115e"
+      "private_key": "0ca1a40ddecdab4f5b04fa0bfed1d235beaa2b8082e7554425607516f0862075dfe357de55649e6d2ce889acf15eb77e94ab3c5756fe46d3c7538d37f27f115e"
     }
   }
 }

build/localnet/Tiltfile

@@ -48,14 +48,19 @@ deps = [
     "build/debug.go",
     "consensus",
     "p2p",
-    "persistance",
+    "persistence",
     "rpc",
     "runtime",
     "shared",
     "telemetry",
     "utility",
     "vendor",
     "logger",
+    "e2e",
+    "ibc",
+    "internal",
+    "state_machine",
+    "tools",
 ]
 
 deps_full_path = [root_dir + "/" + depdir for depdir in deps]

build/localnet/cluster-manager/main.go

@@ -57,7 +57,8 @@ func init() {
 	clusterManagerCmd.PersistentFlags().StringVar(
 		&flags.RemoteCLIURL,
 		"remote_cli_url",
-		defaults.Validator1EndpointK8SHostname,
+		// defaults.Validator1EndpointK8SHostname,
+		defaults.FullNode1EndpointK8SHostname,
 		"takes a remote endpoint in the form of <protocol>://<host>:<port> (uses RPC Port)",
 	)
 

charts/pocket/README.md

@@ -43,6 +43,7 @@ privateKeySecretKeyRef:
 | config.consensus.pacemaker_config.manual | bool | `true` |  |
 | config.consensus.pacemaker_config.timeout_msec | int | `10000` |  |
 | config.consensus.private_key | string | `""` |  |
+| config.consensus.server_mode_enabled | bool | `true` |  |
 | config.fisherman.enabled | bool | `false` |  |
 | config.ibc.enabled | bool | `true` |  |
 | config.ibc.host.private_key | string | `""` |  |

charts/pocket/values.yaml

@@ -80,6 +80,7 @@ config:
       manual: true
       debug_time_between_steps_msec: 1000
     private_key: "" # @ignored This value is needed but ignored - use privateKeySecretKeyRef instead
+    server_mode_enabled: true
   utility:
     max_mempool_transaction_bytes: 1073741824
     max_mempool_transactions: 9000

consensus/block.go

@@ -34,9 +34,12 @@ func (m *consensusModule) commitBlock(block *coreTypes.Block) error {
 	return nil
 }
 
+// isBlockInMessageValidBasic does basic validation of the block in the hotstuff message such as:
+// - validating if the block could/should be nil
+// - the state hash of the block
+// - the size of the block
 // ADDTEST: Add unit tests specific to block validation
-// IMPROVE: Rename to provide clarity of operation. ValidateBasic() is typically a stateless check not stateful
-func (m *consensusModule) isValidMessageBlock(msg *typesCons.HotstuffMessage) (bool, error) {
+func (m *consensusModule) isBlockInMessageValidBasic(msg *typesCons.HotstuffMessage) (bool, error) {
 	block := msg.GetBlock()
 	step := msg.GetStep()
 
@@ -73,12 +76,17 @@ func (m *consensusModule) isValidMessageBlock(msg *typesCons.HotstuffMessage) (b
 	return true, nil
 }
 
-// Creates a new Utility Unit Of Work and clears/nullifies any previous UOW if they exist
+// refreshUtilityUnitOfWork is a helper that creates a new Utility Unit Of Work and clears/nullifies a previous one if it exists
 func (m *consensusModule) refreshUtilityUnitOfWork() error {
+	// m.m.Lock()
+	// defer m.m.Unlock()
+
 	// Catch-all structure to release the previous utility UOW if it wasn't properly cleaned up.
 	utilityUnitOfWork := m.utilityUnitOfWork
+
+	// TECHDEBT: This should, theoretically, never happen. Need to identify all
+	// code paths where it does and fix it.
 	if utilityUnitOfWork != nil {
-		// TODO: This should, ideally, never be called
 		m.logger.Warn().Bool("TODO", true).Msg(typesCons.NilUtilityUOWWarning)
 		if err := utilityUnitOfWork.Release(); err != nil {
 			m.logger.Warn().Err(err).Msg("failed to release utility unit of work")

consensus/debugging.go

@@ -2,7 +2,6 @@ package consensus
 
 import (
 	typesCons "github.com/pokt-network/pocket/consensus/types"
-	cryptoPocket "github.com/pokt-network/pocket/shared/crypto"
 	"github.com/pokt-network/pocket/shared/messaging"
 	"google.golang.org/protobuf/types/known/anypb"
 )
@@ -83,74 +82,43 @@ func (m *consensusModule) togglePacemakerManualMode(_ *messaging.DebugMessage) {
 	m.paceMaker.SetManualMode(newMode)
 }
 
-// requests current block from all validators
+// sendGetBlockStateSyncMessage sends a messages to request specific blocks from peers
 func (m *consensusModule) sendGetBlockStateSyncMessage(_ *messaging.DebugMessage) {
-	currentHeight := m.CurrentHeight()
-	requestHeight := currentHeight - 1
-	peerAddress := m.GetNodeAddress()
-
 	stateSyncGetBlockMessage := &typesCons.StateSyncMessage{
 		Message: &typesCons.StateSyncMessage_GetBlockReq{
 			GetBlockReq: &typesCons.GetBlockRequest{
-				PeerAddress: peerAddress,
-				Height:      requestHeight,
+				PeerAddress: m.GetNodeAddress(),
+				Height:      m.CurrentHeight() - 1,
 			},
 		},
 	}
-
-	validators, err := m.getValidatorsAtHeight(currentHeight)
+	anyMsg, err := anypb.New(stateSyncGetBlockMessage)
 	if err != nil {
-		m.logger.Debug().Msgf(typesCons.ErrPersistenceGetAllValidators.Error(), err)
+		m.logger.Error().Err(err).Str("proto_type", "GetBlockRequest").Msg("failed to create StateSyncGetBlockMessage")
+		return
 	}
-
-	for _, val := range validators {
-		if m.GetNodeAddress() == val.GetAddress() {
-			continue
-		}
-		valAddress := cryptoPocket.AddressFromString(val.GetAddress())
-
-		anyMsg, err := anypb.New(stateSyncGetBlockMessage)
-		if err != nil {
-			m.logger.Error().Err(err).Str("proto_type", "GetBlockRequest").Msg("failed to send StateSyncMessage")
-		}
-
-		if err := m.GetBus().GetP2PModule().Send(valAddress, anyMsg); err != nil {
-			m.logger.Error().Err(err).Msg(typesCons.ErrSendMessage.Error())
-		}
+	if err := m.GetBus().GetP2PModule().Broadcast(anyMsg); err != nil {
+		m.logger.Error().Err(err).Msg(typesCons.ErrBroadcastMessage.Error())
+		return
 	}
 }
 
-// requests metadata from all validators
+// sendGetMetadataStateSyncMessage sends a message to request metadata from their peers
 func (m *consensusModule) sendGetMetadataStateSyncMessage(_ *messaging.DebugMessage) {
-	currentHeight := m.CurrentHeight()
-	peerAddress := m.GetNodeAddress()
-
 	stateSyncMetaDataReqMessage := &typesCons.StateSyncMessage{
 		Message: &typesCons.StateSyncMessage_MetadataReq{
 			MetadataReq: &typesCons.StateSyncMetadataRequest{
-				PeerAddress: peerAddress,
+				PeerAddress: m.GetNodeAddress(),
 			},
 		},
 	}
-
-	validators, err := m.getValidatorsAtHeight(currentHeight)
+	anyMsg, err := anypb.New(stateSyncMetaDataReqMessage)
 	if err != nil {
-		m.logger.Debug().Msgf(typesCons.ErrPersistenceGetAllValidators.Error(), err)
+		m.logger.Error().Err(err).Str("proto_type", "StateSyncMessage").Msg("failed to create StateSyncMetadataRequest")
+		return
 	}
-
-	for _, val := range validators {
-		if m.GetNodeAddress() == val.GetAddress() {
-			continue
-		}
-		valAddress := cryptoPocket.AddressFromString(val.GetAddress())
-
-		anyMsg, err := anypb.New(stateSyncMetaDataReqMessage)
-		if err != nil {
-			m.logger.Error().Err(err).Str("proto_type", "GetMetadataRequest").Msg("failed to send StateSyncMessage")
-		}
-
-		if err := m.GetBus().GetP2PModule().Send(valAddress, anyMsg); err != nil {
-			m.logger.Error().Err(err).Msg(typesCons.ErrSendMessage.Error())
-		}
+	if m.GetBus().GetP2PModule().Broadcast(anyMsg) != nil {
+		m.logger.Error().Err(err).Msg(typesCons.ErrBroadcastMessage.Error())
+		return
 	}
 }

consensus/doc/PROTOCOL_STATE_SYNC.md

@@ -1,6 +1,13 @@
 # State Sync Protocol Design <!-- omit in toc -->
 
-_NOTE: This document makes some assumption of P2P implementation details, so please see [p2p](../../p2p/README.md) for the latest source of truth._
+⚠️ IMPORTANT NOTES TO THE (last updated on 08/03/2023):
+
+- TECHDEBT(#821): Once the FSM is remove, state sync will look completely different
+- State Sync implementation is a WIP and has taken several different shapes.
+- This document is out of date and needs to be updated to reflect the latest implementation. This will be done once a functional implementation is in place.
+- This document makes some assumption of P2P implementation details, so please see [p2p](../../p2p/README.md) for the latest source of truth.
+
+## Table of Contents <!-- omit in toc -->
 
 - [Background](#background)
 - [State Sync - Peer Metadata](#state-sync---peer-metadata)
@@ -55,9 +62,9 @@ Node gathers peer metadata from its peers in `StateSyncMetadataResponse` type, d
 
 ```golang
 type StateSyncMetadataResponse struct {
-    PeerAddress string
-	MinHeight   uint64
-	MaxHeight   uint64
+  PeerAddress string
+  MinHeight   uint64
+  MaxHeight   uint64
 }
 ```
 
@@ -87,7 +94,7 @@ type StateSyncModule interface {
   // ...
   GetAggregatedStateSyncMetadata() *StateSyncMetadataResponse // Aggregated metadata received from peers.
   IsSynced() (bool, error)
-  StartSyncing() error
+  Start() error
   // ...
 }
 ```
@@ -105,7 +112,7 @@ For every new block and block proposal `Validator`s receive:
 
 According to the result of the `IsSynced()` function:
 
-- If the node is out of sync, it runs `StartSyncing()` function. Node requests blocks one by one using the minimum and maximum height in aggregated state sync metadata.
+- If the node is out of sync, it runs `Start()` function. Node requests blocks one by one using the minimum and maximum height in aggregated state sync metadata.
 - If the node is in sync with its peers it rejects the block and/or block proposal.
 
 ```mermaid
@@ -114,13 +121,13 @@ flowchart TD
     A[Node] --> B[Periodic <br> Sync]
     A[Node] --> |New Block| C{IsSynced}
 
-    %% periodic snyc
+    %% periodic sync
     B --> |Request <br> metadata| D[Peers]
     D[Peers] --> |Collect metadata| B[Periodic <br> Sync]
 
 
-    %% is node sycnhed
-    C -->  |No| E[StartSyncing]
+    %% is node synched
+    C -->  |No| E[Start]
     C -->  |Yes| F[Apply Block]
 
     %% syncing
@@ -154,7 +161,7 @@ In `Unsynced` Mode, node transitions to `Sync Mode` by sending `Consensus_IsSync
 
 ### Sync Mode
 
-In `Sync` Mode, the Node is catching up to the latest block by making `GetBlock` requests, via `StartSyncing()` function to eligible peers in its address book. A peer can handle a `GetBlock` request if `PeerSyncMetadata.MinHeight` <= `localSyncState.MaxHeight` <= `PeerSyncMetadata.MaxHeight`.
+In `Sync` Mode, the Node is catching up to the latest block by making `GetBlock` requests, via `Start()` function to eligible peers in its address book. A peer can handle a `GetBlock` request if `PeerSyncMetadata.MinHeight` <= `localSyncState.MaxHeight` <= `PeerSyncMetadata.MaxHeight`.
 
 Though it is unspecified whether or not a Node may make `GetBlock` requests in order or in parallel, the cryptographic restraints of block processing require the Node to call `CommitBlock` sequentially until it is `Synced`.