Network handler (#182)

* async data input

* NetAddr refactoring

* decode and pipe messages

* up message

* handler safrole message

* everything linked

* logging fix

* wip

* fix message parsing

* decoder fix

* fix

* workaround

* fix

* genesis block is known block

* fix

* fix test

* update

* fix

* fix

* allow more delays

* fix

* fix

* fix

Author: xlc

Blockchain/Sources/Blockchain/Blockchain.swift

@@ -54,4 +54,8 @@ public final class Blockchain: ServiceBase, @unchecked Sendable {
 
         publish(RuntimeEvents.BlockFinalized(hash: hash))
     }
+
+    public func publish(event: some Event) {
+        publish(event)
+    }
 }

Blockchain/Sources/Blockchain/BlockchainDataProvider/BlockchainDataProvider.swift

@@ -17,17 +17,15 @@ public final class BlockchainDataProvider: Sendable {
         let heads = try await dataProvider.getHeads()
         var bestHead: (HeaderRef, Data32)?
         for head in heads {
-            guard let header = try? await dataProvider.getHeader(hash: head) else {
-                continue
-            }
+            let header = try await dataProvider.getHeader(hash: head)
             if bestHead == nil || header.value.timeslot > bestHead!.0.value.timeslot {
                 bestHead = (header, head)
             }
         }
         let finalizedHead = try await dataProvider.getFinalizedHead()
 
         storage = ThreadSafeContainer(.init(
-            bestHead: bestHead?.1 ?? Data32(),
+            bestHead: bestHead?.1 ?? dataProvider.genesisBlockHash,
             bestHeadTimeslot: bestHead?.0.value.timeslot ?? 0,
             finalizedHead: finalizedHead
         ))
@@ -129,4 +127,8 @@ extension BlockchainDataProvider {
 
         try await dataProvider.remove(hash: hash)
     }
+
+    public var genesisBlockHash: Data32 {
+        dataProvider.genesisBlockHash
+    }
 }

Blockchain/Sources/Blockchain/BlockchainDataProvider/BlockchainDataProviderProtocol.swift

@@ -34,4 +34,6 @@ public protocol BlockchainDataProviderProtocol: Sendable {
 
     /// remove header, block and state
     func remove(hash: Data32) async throws
+
+    var genesisBlockHash: Data32 { get }
 }

Blockchain/Sources/Blockchain/BlockchainDataProvider/InMemoryDataProvider.swift

@@ -7,12 +7,15 @@ public actor InMemoryDataProvider: Sendable {
     private var blockByHash: [Data32: BlockRef] = [:]
     private var stateByBlockHash: [Data32: StateRef] = [:]
     private var hashByTimeslot: [TimeslotIndex: Set<Data32>] = [:]
+    public let genesisBlockHash: Data32
 
-    public init(genesis: StateRef) async {
-        heads = [Data32()]
-        finalizedHead = Data32()
+    public init(genesisState: StateRef, genesisBlock: BlockRef) async {
+        genesisBlockHash = genesisBlock.hash
+        heads = [genesisBlockHash]
+        finalizedHead = genesisBlockHash
 
-        add(state: genesis)
+        add(block: genesisBlock)
+        add(state: genesisState)
     }
 }
 
@@ -80,8 +83,7 @@ extension InMemoryDataProvider: BlockchainDataProviderProtocol {
         // parent needs to be either
         // - existing head
         // - known block
-        // - genesis / all zeros
-        guard heads.remove(parent) != nil || hasBlock(hash: parent) || parent == Data32() else {
+        guard heads.remove(parent) != nil || hasBlock(hash: parent) else {
             throw BlockchainDataProviderError.noData(hash: parent)
         }
         heads.insert(hash)

Blockchain/Sources/Blockchain/RuntimeProtocols/RuntimeEvents.swift

@@ -20,6 +20,10 @@ public enum RuntimeEvents {
     // New safrole ticket received from network
     public struct SafroleTicketsReceived: Event {
         public let items: [ExtrinsicTickets.TicketItem]
+
+        public init(items: [ExtrinsicTickets.TicketItem]) {
+            self.items = items
+        }
     }
 
     // New block authored by BlockAuthor service

Blockchain/Sources/Blockchain/Types/RecentHistory.swift

@@ -34,7 +34,15 @@ public struct RecentHistory: Sendable, Equatable, Codable {
 extension RecentHistory: Dummy {
     public typealias Config = ProtocolConfigRef
     public static func dummy(config: Config) -> RecentHistory {
-        RecentHistory(items: try! ConfigLimitedSizeArray(config: config))
+        RecentHistory(items: try! ConfigLimitedSizeArray(
+            config: config,
+            array: [HistoryItem(
+                headerHash: Data32(),
+                mmr: MMR([]),
+                stateRoot: Data32(),
+                workReportHashes: ConfigLimitedSizeArray(config: config)
+            )]
+        ))
     }
 }
 

Blockchain/Sources/Blockchain/Types/State+Genesis.swift

@@ -1,7 +1,7 @@
 import Utils
 
 extension State {
-    public static func devGenesis(config: ProtocolConfigRef) throws -> State {
+    public static func devGenesis(config: ProtocolConfigRef) throws -> (StateRef, BlockRef) {
         var devKeys = [ValidatorKey]()
 
         var state = State.dummy(config: config)
@@ -32,7 +32,15 @@ extension State {
         )
         state.safroleState.ticketsVerifier = commitment.data
 
-        return state
+        let block = BlockRef(Block.dummy(config: config))
+        try state.recentHistory.items.append(RecentHistory.HistoryItem(
+            headerHash: block.hash,
+            mmr: MMR([]),
+            stateRoot: Data32(),
+            workReportHashes: ConfigLimitedSizeArray(config: config)
+        ))
+
+        return (StateRef(state), block)
     }
     // TODO: add file genesis
     // public static func fileGenesis(config: ProtocolConfigRef) throws -> State

Blockchain/Sources/Blockchain/Types/State.swift

@@ -124,7 +124,7 @@ public struct State: Sendable, Equatable, Codable {
 
 extension State {
     public var lastBlockHash: Data32 {
-        recentHistory.items.last.map(\.headerHash) ?? Data32()
+        recentHistory.items.last.map(\.headerHash)!
     }
 }
 

Blockchain/Tests/BlockchainTests/BlockAuthorTests.swift

@@ -22,7 +22,8 @@ struct BlockAuthorTests {
         config = ProtocolConfigRef.dev
         timeProvider = MockTimeProvider(time: 988)
 
-        dataProvider = try await BlockchainDataProvider(InMemoryDataProvider(genesis: StateRef(State.devGenesis(config: config))))
+        let (genesisState, genesisBlock) = try State.devGenesis(config: config)
+        dataProvider = try await BlockchainDataProvider(InMemoryDataProvider(genesisState: genesisState, genesisBlock: genesisBlock))
 
         storeMiddleware = StoreMiddleware()
         eventBus = EventBus(eventMiddleware: Middleware(storeMiddleware))
@@ -45,7 +46,7 @@ struct BlockAuthorTests {
 
     @Test
     func createNewBlockWithFallbackKey() async throws {
-        let genesisState = try await dataProvider.getState(hash: Data32())
+        let genesisState = try await dataProvider.getState(hash: dataProvider.genesisBlockHash)
 
         let timeslot = timeProvider.getTime().timeToTimeslot(config: config)
 
@@ -62,7 +63,7 @@ struct BlockAuthorTests {
 
     @Test
     func createNewBlockWithTicket() async throws {
-        let genesisState = try await dataProvider.getState(hash: Data32())
+        let genesisState = try await dataProvider.getState(hash: dataProvider.genesisBlockHash)
         var state = genesisState.value
 
         state.safroleState.ticketsVerifier = try Bandersnatch.RingCommitment(
@@ -108,7 +109,7 @@ struct BlockAuthorTests {
 
     @Test
     func scheduleNewBlocks() async throws {
-        let genesisState = try await dataProvider.getState(hash: Data32())
+        let genesisState = try await dataProvider.getState(hash: dataProvider.genesisBlockHash)
 
         await blockAuthor.on(genesis: genesisState)
 

Blockchain/Tests/BlockchainTests/DispatchQueueSchedulerTests.swift

@@ -38,23 +38,23 @@ struct DispatchQueueSchedulerTests {
 
             let diff = try #require(end.value).timeIntervalSince(now) - delay
             let diffAbs = abs(diff)
-            #expect(diffAbs < 0.5)
+            #expect(diffAbs < 1)
         }
     }
 
     @Test func scheduleRepeatingTask() async throws {
-        try await confirmation(expectedCount: 3) { confirm in
-            let delay = 0.5
+        try await confirmation(expectedCount: 2) { confirm in
+            let delay = 1.5
             let now = Date()
             let executionTimes = ThreadSafeContainer<[Date]>([])
-            let expectedExecutions = 3
+            let expectedExecutions = 2
 
             let cancel = scheduler.schedule(delay: delay, repeats: true) {
                 executionTimes.value.append(Date())
                 confirm()
             }
 
-            try await Task.sleep(for: .seconds(1.6))
+            try await Task.sleep(for: .seconds(3.1))
 
             _ = cancel
 
@@ -64,7 +64,7 @@ struct DispatchQueueSchedulerTests {
                 let expectedInterval = delay * Double(index + 1)
                 let actualInterval = time.timeIntervalSince(now)
                 let difference = abs(actualInterval - expectedInterval)
-                #expect(difference < 0.5)
+                #expect(difference < 1)
             }
         }
     }
@@ -83,13 +83,13 @@ struct DispatchQueueSchedulerTests {
 
     @Test func cancelRepeatingTask() async throws {
         try await confirmation(expectedCount: 2) { confirm in
-            let delay = 0.5
+            let delay = 1.0
 
             let cancel = scheduler.schedule(delay: delay, repeats: true) {
                 confirm()
             }
 
-            try await Task.sleep(for: .seconds(1.2))
+            try await Task.sleep(for: .seconds(2.2))
 
             cancel.cancel()