syncing

xlc · xlc · commit 3e2af6cb1055 · 2024-10-25T16:45:07.000+13:00
diff --git a/Blockchain/Sources/Blockchain/Blockchain.swift b/Blockchain/Sources/Blockchain/Blockchain.swift
@@ -29,6 +29,11 @@ public final class Blockchain: ServiceBase, @unchecked Sendable {
     public func importBlock(_ block: BlockRef) async throws {
         logger.debug("importing block: #\(block.header.timeslot) \(block.hash)")
 
+        if try await dataProvider.hasBlock(hash: block.hash) {
+            logger.debug("block already imported", metadata: ["hash": "\(block.hash)"])
+            return
+        }
+
         try await withSpan("importBlock") { span in
             span.attributes.blockHash = block.hash.description
 
diff --git a/Blockchain/Sources/Blockchain/Types/Header.swift b/Blockchain/Sources/Blockchain/Types/Header.swift
@@ -133,7 +133,35 @@ extension Header {
     }
 }
 
-public typealias HeaderRef = Ref<Header>
+public final class HeaderRef: Ref<Header>, @unchecked Sendable {
+    public required init(_ value: Header) {
+        lazyHash = Lazy {
+            Ref(value.hash())
+        }
+
+        super.init(value)
+    }
+
+    private let lazyHash: Lazy<Ref<Data32>>
+
+    public var hash: Data32 {
+        lazyHash.value.value
+    }
+
+    override public var description: String {
+        "Header(hash: \(hash), timeslot: \(value.timeslot))"
+    }
+}
+
+extension HeaderRef: Codable {
+    public convenience init(from decoder: Decoder) throws {
+        try self.init(.init(from: decoder))
+    }
+
+    public func encode(to encoder: Encoder) throws {
+        try value.encode(to: encoder)
+    }
+}
 
 extension Header.Unsigned: Dummy {
     public typealias Config = ProtocolConfigRef
diff --git a/Node/Sources/Node/NetworkingProtocol/CommonEphemeral/BlockRequest.swift b/Node/Sources/Node/NetworkingProtocol/CommonEphemeral/BlockRequest.swift
@@ -0,0 +1,12 @@
+import Utils
+
+public struct BlockRequest: Codable, Sendable {
+    public enum Direction: UInt8, Codable, Sendable {
+        case ascendingExcludsive = 0
+        case descendingInclusive = 1
+    }
+
+    public var hash: Data32
+    public var direction: Direction
+    public var maxBlocks: UInt32
+}
diff --git a/Node/Sources/Node/NetworkingProtocol/CommonEphemeral/CERequest.swift b/Node/Sources/Node/NetworkingProtocol/CommonEphemeral/CERequest.swift
@@ -4,6 +4,7 @@ import Foundation
 import Networking
 
 public enum CERequest: Sendable {
+    case blockRequest(BlockRequest)
     case safroleTicket1(SafroleTicketMessage)
     case safroleTicket2(SafroleTicketMessage)
 }
@@ -13,6 +14,8 @@ extension CERequest: RequestProtocol {
 
     public func encode() throws -> Data {
         switch self {
+        case let .blockRequest(message):
+            try JamEncoder.encode(message)
         case let .safroleTicket1(message):
             try JamEncoder.encode(message)
         case let .safroleTicket2(message):
@@ -22,6 +25,8 @@ extension CERequest: RequestProtocol {
 
     public var kind: CommonEphemeralStreamKind {
         switch self {
+        case .blockRequest:
+            .blockRequest
         case .safroleTicket1:
             .safroleTicket1
         case .safroleTicket2:
@@ -31,6 +36,8 @@ extension CERequest: RequestProtocol {
 
     static func getType(kind: CommonEphemeralStreamKind) -> Decodable.Type {
         switch kind {
+        case .blockRequest:
+            BlockRequest.self
         case .safroleTicket1:
             SafroleTicketMessage.self
         case .safroleTicket2:
@@ -42,6 +49,11 @@ extension CERequest: RequestProtocol {
 
     static func from(kind: CommonEphemeralStreamKind, data: any Decodable) -> CERequest? {
         switch kind {
+        case .blockRequest:
+            guard let message = data as? BlockRequest else {
+                return nil
+            }
+            return .blockRequest(message)
         case .safroleTicket1:
             guard let message = data as? SafroleTicketMessage else {
                 return nil
@@ -57,31 +69,3 @@ extension CERequest: RequestProtocol {
         }
     }
 }
-
-extension CERequest {
-    public func handle(blockchain: Blockchain) async throws -> (any Encodable)? {
-        switch self {
-        case let .safroleTicket1(message):
-            blockchain.publish(event: RuntimeEvents.SafroleTicketsReceived(
-                items: [
-                    ExtrinsicTickets.TicketItem(
-                        attempt: message.attempt,
-                        signature: message.proof
-                    ),
-                ]
-            ))
-            // TODO: rebroadcast to other peers after some time
-            return nil
-        case let .safroleTicket2(message):
-            blockchain.publish(event: RuntimeEvents.SafroleTicketsReceived(
-                items: [
-                    ExtrinsicTickets.TicketItem(
-                        attempt: message.attempt,
-                        signature: message.proof
-                    ),
-                ]
-            ))
-            return nil
-        }
-    }
-}
diff --git a/Node/Sources/Node/NetworkingProtocol/NetworkEvents.swift b/Node/Sources/Node/NetworkingProtocol/NetworkEvents.swift
@@ -0,0 +1,15 @@
+import Blockchain
+import Utils
+
+public enum NetworkEvents {
+    public struct PeerAdded: Event {
+        public let info: PeerInfo
+    }
+
+    public struct PeerUpdated: Event {
+        public let info: PeerInfo
+        public let newBlockHeader: HeaderRef
+    }
+
+    public struct BulkSyncCompleted: Event {}
+}
diff --git a/Node/Sources/Node/NetworkingProtocol/NetworkManager.swift b/Node/Sources/Node/NetworkingProtocol/NetworkManager.swift
@@ -7,14 +7,17 @@ import Utils
 
 private let logger = Logger(label: "NetworkManager")
 
-enum SendTarget {
+let MAX_BLOCKS_PER_REQUEST: UInt32 = 50
+
+enum BroadcastTarget {
     case safroleStep1Validator
     case currentValidators
 }
 
 public final class NetworkManager: Sendable {
     private let peerManager: PeerManager
     private let network: Network
+    private let syncManager: SyncManager
     private let blockchain: Blockchain
     private let subscriptions: EventSubscriptions
 
@@ -23,16 +26,19 @@ public final class NetworkManager: Sendable {
     private let devPeers: Set<NetAddr>
 
     public init(config: Network.Config, blockchain: Blockchain, eventBus: EventBus, devPeers: Set<NetAddr>) async throws {
-        peerManager = PeerManager()
+        peerManager = PeerManager(eventBus: eventBus)
 
-        let handler = HandlerImpl(blockchain: blockchain, peerManager: peerManager)
         network = try await Network(
             config: config,
             protocolConfig: blockchain.config,
             genesisHeader: blockchain.dataProvider.genesisBlockHash,
-            handler: handler
+            handler: HandlerImpl(blockchain: blockchain, peerManager: peerManager)
+        )
+        syncManager = SyncManager(
+            blockchain: blockchain, network: network, peerManager: peerManager, eventBus: eventBus
         )
         self.blockchain = blockchain
+
         subscriptions = EventSubscriptions(eventBus: eventBus)
 
         self.devPeers = devPeers
@@ -53,7 +59,7 @@ public final class NetworkManager: Sendable {
         }
     }
 
-    private func getSendTarget(target: SendTarget) -> Set<NetAddr> {
+    private func getAddresses(target: BroadcastTarget) -> Set<NetAddr> {
         // TODO: get target from onchain state
         switch target {
         case .safroleStep1Validator:
@@ -65,31 +71,29 @@ public final class NetworkManager: Sendable {
         }
     }
 
-    private func send<R: Decodable>(
+    private func send(to: NetAddr, message: CERequest) async throws -> Data {
+        try await network.send(to: to, message: message)
+    }
+
+    private func broadcast(
+        to: BroadcastTarget,
         message: CERequest,
-        target: SendTarget,
-        responseType: R.Type,
-        responseHandler: @Sendable @escaping (Result<R, Error>) async -> Void
+        responseHandler: @Sendable @escaping (Result<Data, Error>) async -> Void
     ) async {
-        let targets = getSendTarget(target: target)
+        let targets = getAddresses(target: to)
         for target in targets {
             Task {
                 logger.trace("sending message", metadata: ["target": "\(target)", "message": "\(message)"])
                 let res = await Result {
                     try await network.send(to: target, message: message)
                 }
-                .flatMap { data in
-                    Result(catching: {
-                        try JamDecoder.decode(responseType, from: data, withConfig: blockchain.config)
-                    })
-                }
                 await responseHandler(res)
             }
         }
     }
 
-    private func send(message: CERequest, target: SendTarget) async {
-        let targets = getSendTarget(target: target)
+    private func broadcast(to: BroadcastTarget, message: CERequest) async {
+        let targets = getAddresses(target: to)
         for target in targets {
             Task {
                 logger.trace("sending message", metadata: ["target": "\(target)", "message": "\(message)"])
@@ -103,13 +107,13 @@ public final class NetworkManager: Sendable {
     private func on(safroleTicketsGenerated event: RuntimeEvents.SafroleTicketsGenerated) async {
         logger.trace("sending tickets", metadata: ["epochIndex": "\(event.epochIndex)"])
         for ticket in event.items {
-            await send(
+            await broadcast(
+                to: .safroleStep1Validator,
                 message: .safroleTicket1(.init(
                     epochIndex: event.epochIndex,
                     attempt: ticket.ticket.attempt,
                     proof: ticket.ticket.signature
-                )),
-                target: .safroleStep1Validator
+                ))
             )
         }
     }
@@ -125,6 +129,40 @@ struct HandlerImpl: NetworkProtocolHandler {
 
     func handle(ceRequest: CERequest) async throws -> (any Encodable)? {
         switch ceRequest {
+        case let .blockRequest(message):
+            let dataProvider = blockchain.dataProvider
+            let count = min(MAX_BLOCKS_PER_REQUEST, message.maxBlocks)
+            var resp = [BlockRef]()
+            resp.reserveCapacity(Int(count))
+            switch message.direction {
+            case .ascendingExcludsive:
+                let number = try await dataProvider.getBlockNumber(hash: message.hash)
+                var currentHash = message.hash
+                for i in 1 ... count {
+                    let hashes = try await dataProvider.getBlockHash(byNumber: number + i)
+                    var found = false
+                    for hash in hashes {
+                        let block = try await dataProvider.getBlock(hash: hash)
+                        if block.header.parentHash == currentHash {
+                            resp.append(block)
+                            found = true
+                            currentHash = hash
+                            break
+                        }
+                    }
+                    if !found {
+                        break
+                    }
+                }
+            case .descendingInclusive:
+                var hash = message.hash
+                for _ in 0 ..< count {
+                    let block = try await dataProvider.getBlock(hash: hash)
+                    resp.append(block)
+                    hash = block.header.parentHash
+                }
+            }
+            return resp
         case let .safroleTicket1(message):
             blockchain.publish(event: RuntimeEvents.SafroleTicketsReceived(
                 items: [
diff --git a/Node/Sources/Node/NetworkingProtocol/PeerManager.swift b/Node/Sources/Node/NetworkingProtocol/PeerManager.swift
@@ -6,16 +6,28 @@ import Utils
 
 private let logger = Logger(label: "PeerManager")
 
-public struct PeerInfo {
+public struct PeerInfo: Sendable {
     public let address: NetAddr
     public internal(set) var finalized: HashAndSlot
     public internal(set) var heads: Set<HashAndSlot> = []
+
+    public var best: HashAndSlot? {
+        heads.max { $0.timeslot < $1.timeslot }
+    }
 }
 
+// TODOs:
+// - distinguish between connect peers and offline peers
+// - peer reputation
+// - purge offline peers
 public actor PeerManager: Sendable {
-    private var peers: [NetAddr: PeerInfo] = [:]
+    private let eventBus: EventBus
+
+    public private(set) var peers: [NetAddr: PeerInfo] = [:]
 
-    init() {}
+    init(eventBus: EventBus) {
+        self.eventBus = eventBus
+    }
 
     func addPeer(address: NetAddr, handshake: BlockAnnouncementHandshake) {
         var peer = PeerInfo(
@@ -28,35 +40,35 @@ public actor PeerManager: Sendable {
         peers[address] = peer
 
         logger.debug("added peer", metadata: ["address": "\(address)", "finalized": "\(peer.finalized)"])
+        eventBus.publish(NetworkEvents.PeerAdded(info: peer))
     }
 
     func updatePeer(address: NetAddr, message: BlockAnnouncement) {
+        let updatedPeer: PeerInfo
         if var peer = peers[address] {
             peer.finalized = message.finalized
             // purge heads that are older than the finalized head
             // or if it is the parent of the new block
             // this means if some blocks are skipped, it is possible that we miss purge some heads
             // that is ancestor of the new block. but that's fine
             peer.heads = peer.heads.filter { head in
-                head.timeslot > message.finalized.timeslot && head.hash != message.header.parentHash
+                head.timeslot > message.finalized.timeslot && head.hash != message.header.value.parentHash
             }
-            peer.heads.insert(HashAndSlot(hash: message.header.hash(), timeslot: message.header.timeslot))
-            peers[address] = peer
+            peer.heads.insert(HashAndSlot(hash: message.header.hash, timeslot: message.header.value.timeslot))
+            updatedPeer = peer
         } else {
             // this shouldn't happen but let's handle it
-            peers[address] = PeerInfo(
+            updatedPeer = PeerInfo(
                 address: address,
                 finalized: message.finalized,
                 heads: [
-                    HashAndSlot(hash: message.header.hash(), timeslot: message.header.timeslot),
+                    HashAndSlot(hash: message.header.hash, timeslot: message.header.value.timeslot),
                 ]
             )
         }
+        peers[address] = updatedPeer
 
-        logger.debug("updated peer", metadata: ["address": "\(address)", "finalized": "\(peers[address]!.finalized)"])
-    }
-
-    public func getPeer(address: NetAddr) -> PeerInfo? {
-        peers[address]
+        logger.debug("updated peer", metadata: ["address": "\(address)", "finalized": "\(updatedPeer.finalized)"])
+        eventBus.publish(NetworkEvents.PeerUpdated(info: updatedPeer, newBlockHeader: message.header))
     }
 }
diff --git a/Node/Sources/Node/NetworkingProtocol/SyncManager.swift b/Node/Sources/Node/NetworkingProtocol/SyncManager.swift
diff --git a/Node/Sources/Node/NetworkingProtocol/UniquePresistent/BlockAnnouncement.swift b/Node/Sources/Node/NetworkingProtocol/UniquePresistent/BlockAnnouncement.swift
