sync

Author: xlc

Blockchain/Sources/Blockchain/Validator/ValidatorService.swift

@@ -1,7 +1,7 @@
 import Foundation
 import Utils
 
-public class ValidatorService {
+public final class ValidatorService: Sendable {
     private let blockchain: Blockchain
     private let keystore: KeyStore
     private let safrole: SafroleService

Boka/Sources/Boka.swift

@@ -58,7 +58,7 @@ struct Boka: AsyncParsableCommand {
     var basePath: String?
 
     @Option(name: .long, help: "A preset config or path to chain config file.")
-    var chain: Genesis = .preset(.dev)
+    var chain: Genesis = .preset(.minimal)
 
     @Option(name: .long, help: "Listen address for RPC server. Pass 'no' to disable RPC server. Default to 127.0.0.1:9955.")
     var rpc: MaybeEnabled<NetAddr> = .enabled(NetAddr(address: "127.0.0.1:9955")!)
@@ -81,6 +81,12 @@ struct Boka: AsyncParsableCommand {
     @Option(name: .long, help: "Node name. For telemetry only.")
     var name: String?
 
+    @Flag(name: .long, help: "Enable local mode, whereas peers are not expected.")
+    var local: Bool = false
+
+    @Flag(name: .long, help: "Enable dev mode. This is equivalent to --local --validator")
+    var dev: Bool = false
+
     mutating func run() async throws {
         let services = try await Tracing.bootstrap("Boka", loggerOnly: true)
         for service in services {
@@ -95,6 +101,16 @@ struct Boka: AsyncParsableCommand {
 
         logger.info("Chain: \(chain)")
 
+        if dev {
+            local = true
+            validator = true
+            logger.info("Dev mode enabled. Enabling local and validator.")
+        }
+
+        if local {
+            logger.info("Local mode enabled.")
+        }
+
         if let name {
             logger.info("Node name: \(name)")
         }
@@ -146,7 +162,12 @@ struct Boka: AsyncParsableCommand {
             handlerMiddleware: .tracing(prefix: "Handler")
         )
 
-        let config = Node.Config(rpc: rpcConfig, network: networkConfig, peers: peers)
+        let config = Node.Config(
+            rpc: rpcConfig,
+            network: networkConfig,
+            peers: peers,
+            local: local
+        )
 
         let node: Node = if validator {
             try await ValidatorNode(

Node/Sources/Node/NetworkingProtocol/NetworkManager.swift

@@ -15,17 +15,22 @@ enum BroadcastTarget {
 }
 
 public final class NetworkManager: Sendable {
-    private let peerManager: PeerManager
-    private let network: Network
-    private let syncManager: SyncManager
-    private let blockchain: Blockchain
+    let peerManager: PeerManager
+    let network: Network
+    let syncManager: SyncManager
+    let blockchain: Blockchain
     private let subscriptions: EventSubscriptions
 
     // This is for development only
     // Those peers will receive all the messages regardless the target
     private let devPeers: Set<NetAddr>
 
-    public init(config: Network.Config, blockchain: Blockchain, eventBus: EventBus, devPeers: Set<NetAddr>) async throws {
+    public init(
+        config: Network.Config,
+        blockchain: Blockchain,
+        eventBus: EventBus,
+        devPeers: Set<NetAddr>
+    ) async throws {
         peerManager = PeerManager(eventBus: eventBus)
 
         network = try await Network(

Node/Sources/Node/NetworkingProtocol/SyncManager.swift

@@ -13,15 +13,17 @@ let BLOCK_REQUEST_BLOCK_COUNT: UInt32 = 50
 // - remove slow one
 // - sync peer rotation
 // - fast sync mode (no verification)
-public actor SyncManager {
+public actor SyncManager: Sendable {
     private let blockchain: Blockchain
     private let network: Network
     private let peerManager: PeerManager
 
     private let subscriptions: EventSubscriptions
 
     // starts with bulk syncing mode, until our best have catched up with the peer best
-    private var bulkSyncing = true
+    private var bulkSyncing = false
+    private var syncContinuation: [CheckedContinuation<Void, Never>] = []
+
     private var networkBest: HashAndSlot?
     private var networkFinalizedBest: HashAndSlot?
     private var currentRequest: (peer: NetAddr, request: BlockRequest)?
@@ -42,6 +44,15 @@ public actor SyncManager {
         }
     }
 
+    public func waitForSyncCompletion() async {
+        if !bulkSyncing {
+            return
+        }
+        await withCheckedContinuation { continuation in
+            syncContinuation.append(continuation)
+        }
+    }
+
     private func on(peerUpdated info: PeerInfo, newBlockHeader: HeaderRef?) async {
         // TODO: improve this to handle the case misbehaved peers seding us the wrong best
         if let networkBest {
@@ -97,7 +108,9 @@ public actor SyncManager {
                     if currentHead.timeslot >= networkBest!.timeslot {
                         if bulkSyncing {
                             bulkSyncing = false
-                            subscriptions.publish(NetworkEvents.BulkSyncCompleted())
+                            syncContinuation.forEach { $0.resume() }
+                            syncContinuation = []
+                            logger.info("bulk sync completed")
                             return
                         }
                     }

Node/Sources/Node/Node.swift

@@ -14,14 +14,17 @@ public class Node {
         public var rpc: RPCConfig?
         public var network: NetworkConfig
         public var peers: [NetAddr]
+        public var local: Bool
 
-        public init(rpc: RPCConfig?, network: NetworkConfig, peers: [NetAddr] = []) {
+        public init(rpc: RPCConfig?, network: NetworkConfig, peers: [NetAddr] = [], local: Bool = false) {
             self.rpc = rpc
             self.network = network
             self.peers = peers
+            self.local = local
         }
     }
 
+    public let config: Config
     public let blockchain: Blockchain
     public let rpcServer: Server?
     public let timeProvider: TimeProvider
@@ -35,6 +38,8 @@ public class Node {
         eventBus: EventBus,
         keystore: KeyStore
     ) async throws {
+        self.config = config
+
         let (genesisState, genesisBlock, protocolConfig) = try await genesis.load()
 
         logger.info("Genesis: \(genesisBlock.hash)")

Node/Sources/Node/ValidatorNode.swift

@@ -12,16 +12,24 @@ public class ValidatorNode: Node {
         try await super.init(config: config, genesis: genesis, eventBus: eventBus, keystore: keystore)
 
         let scheduler = DispatchQueueScheduler(timeProvider: timeProvider)
-        validator = await ValidatorService(
+        let validator = await ValidatorService(
             blockchain: blockchain,
             keystore: keystore,
             eventBus: eventBus,
             scheduler: scheduler,
             dataProvider: dataProvider
         )
+        self.validator = validator
 
-        let genesisState = try await dataProvider.getState(hash: blockchain.dataProvider.genesisBlockHash)
-
-        await validator.on(genesis: genesisState)
+        let syncManager = network.syncManager
+        let dataProvider = blockchain.dataProvider
+        let local = config.local
+        Task {
+            let genesisState = try await dataProvider.getState(hash: dataProvider.genesisBlockHash)
+            if !local {
+                await syncManager.waitForSyncCompletion()
+            }
+            await validator.on(genesis: genesisState)
+        }
     }
 }