Multinode (#190)

* minor fix

* fix connection

* broadcase safrole tickets

* update logging

* fix unaligned load

* trace

* fix tests

Author: xlc

Blockchain/Sources/Blockchain/RuntimeProtocols/RuntimeEvents.swift

@@ -13,6 +13,7 @@ public enum RuntimeEvents {
 
     // New safrole ticket generated from SafroleService
     public struct SafroleTicketsGenerated: Event {
+        public let epochIndex: EpochIndex
         public let items: [TicketItemAndOutput]
         public let publicKey: Bandersnatch.PublicKey
     }

Blockchain/Sources/Blockchain/Validator/SafroleService.swift

@@ -77,6 +77,7 @@ public final class SafroleService: ServiceBase, @unchecked Sendable {
                 )
 
                 events.append(.init(
+                    epochIndex: state.value.timeslot.timeslotToEpochIndex(config: config),
                     items: tickets,
                     publicKey: secret.publicKey
                 ))

Blockchain/Sources/Blockchain/Validator/ServiceBase.swift

@@ -5,43 +5,27 @@ public class ServiceBase {
     public let id: UniqueId
     let logger: Logger
     public let config: ProtocolConfigRef
-    private let eventBus: EventBus
-    private let subscriptionTokens: ThreadSafeContainer<[EventBus.SubscriptionToken]> = .init([])
+    private let subscriptions: EventSubscriptions
 
     init(id: UniqueId, config: ProtocolConfigRef, eventBus: EventBus) {
         self.id = id
         logger = Logger(label: id)
         self.config = config
-        self.eventBus = eventBus
-    }
-
-    deinit {
-        let eventBus = self.eventBus
-        let subscriptionTokens = self.subscriptionTokens
-        Task {
-            for token in subscriptionTokens.value {
-                await eventBus.unsubscribe(token: token)
-            }
-        }
+        subscriptions = EventSubscriptions(eventBus: eventBus)
     }
 
     @discardableResult
-    func subscribe<T: Event>(_ eventType: T.Type, id _: UniqueId, handler: @escaping @Sendable (T) async throws -> Void) async -> EventBus
-        .SubscriptionToken
-    {
-        let token = await eventBus.subscribe(eventType, handler: handler)
-        subscriptionTokens.write { $0.append(token) }
-        return token
+    func subscribe<T: Event>(
+        _ eventType: T.Type, id _: UniqueId, handler: @escaping @Sendable (T) async throws -> Void
+    ) async -> EventBus.SubscriptionToken {
+        await subscriptions.subscribe(eventType, id: id, handler: handler)
     }
 
     func unsubscribe(token: EventBus.SubscriptionToken) async {
-        subscriptionTokens.write { tokens in
-            tokens.removeAll { $0 == token }
-        }
-        await eventBus.unsubscribe(token: token)
+        await subscriptions.unsubscribe(token: token)
     }
 
     func publish(_ event: some Event) {
-        eventBus.publish(event)
+        subscriptions.publish(event)
     }
 }

Blockchain/Tests/BlockchainTests/ExtrinsicPoolServiceTests.swift

@@ -56,7 +56,11 @@ struct ExtrinsicPoolServiceTests {
 
             allTickets.append(contentsOf: tickets)
 
-            let event = RuntimeEvents.SafroleTicketsGenerated(items: tickets, publicKey: secretKey.publicKey)
+            let event = RuntimeEvents.SafroleTicketsGenerated(
+                epochIndex: state.value.timeslot.timeslotToEpochIndex(config: config),
+                items: tickets,
+                publicKey: secretKey.publicKey
+            )
             await eventBus.publish(event)
 
             // Wait for the event to be processed
@@ -126,7 +130,11 @@ struct ExtrinsicPoolServiceTests {
             idx: 0
         )
 
-        let addEvent = RuntimeEvents.SafroleTicketsGenerated(items: tickets, publicKey: secretKey.publicKey)
+        let addEvent = RuntimeEvents.SafroleTicketsGenerated(
+            epochIndex: state.value.timeslot.timeslotToEpochIndex(config: config),
+            items: tickets,
+            publicKey: secretKey.publicKey
+        )
         await eventBus.publish(addEvent)
 
         // Wait for the event to be processed
@@ -173,7 +181,11 @@ struct ExtrinsicPoolServiceTests {
             idx: 0
         )
 
-        let addEvent = RuntimeEvents.SafroleTicketsGenerated(items: oldTickets, publicKey: secretKey.publicKey)
+        let addEvent = RuntimeEvents.SafroleTicketsGenerated(
+            epochIndex: state.value.timeslot.timeslotToEpochIndex(config: config),
+            items: oldTickets,
+            publicKey: secretKey.publicKey
+        )
         await eventBus.publish(addEvent)
         await storeMiddleware.wait()
 
@@ -216,7 +228,11 @@ struct ExtrinsicPoolServiceTests {
         )
 
         // Ensure new tickets are accepted
-        let newAddEvent = RuntimeEvents.SafroleTicketsGenerated(items: newTickets, publicKey: secretKey.publicKey)
+        let newAddEvent = RuntimeEvents.SafroleTicketsGenerated(
+            epochIndex: newState.value.timeslot.timeslotToEpochIndex(config: config),
+            items: newTickets,
+            publicKey: secretKey.publicKey
+        )
         await eventBus.publish(newAddEvent)
         await storeMiddleware.wait()
 

Boka/.swiftpm/xcode/xcshareddata/xcschemes/Boka.xcscheme

@@ -67,6 +67,10 @@
             argument = "--validator"
             isEnabled = "YES">
          </CommandLineArgument>
+         <CommandLineArgument
+            argument = "--chain=minimal"
+            isEnabled = "YES">
+         </CommandLineArgument>
       </CommandLineArguments>
       <EnvironmentVariables>
          <EnvironmentVariable

Boka/Package.resolved

@@ -27,7 +27,7 @@ enum MaybeEnabled<T: ExpressibleByArgument>: ExpressibleByArgument {
     case disabled
 
     init?(argument: String) {
-        if argument.lowercased() == "false" {
+        if argument.lowercased() == "no" {
             self = .disabled
         } else {
             guard let argument = T(argument: argument) else {
@@ -60,11 +60,11 @@ struct Boka: AsyncParsableCommand {
     @Option(name: .long, help: "A preset config or path to chain config file.")
     var chain: Genesis = .preset(.dev)
 
-    @Option(name: .long, help: "Listen address for RPC server. Pass 'false' to disable RPC server. Default to 127.0.0.1:9955.")
+    @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")!)
 
     @Option(name: .long, help: "Listen address for P2P protocol.")
-    var p2p: NetAddr = .init(address: "127.0.0.1:19955")!
+    var p2p: NetAddr = .init(address: "127.0.0.1:0")!
 
     @Option(name: .long, help: "Specify peer P2P addresses.")
     var peers: [NetAddr] = []
@@ -114,12 +114,11 @@ struct Boka: AsyncParsableCommand {
         }
 
         let rpcConfig = rpc.asOptional.map { addr -> RPCConfig in
-            logger.info("RPC listen address: \(addr)")
             let (address, port) = addr.getAddressAndPort()
             return RPCConfig(listenAddress: address, port: Int(port))
         }
 
-        let keystore = try await DevKeyStore()
+        let keystore = try await DevKeyStore(devKeysCount: devSeed == nil ? 12 : 0)
 
         let networkKey: Ed25519.SecretKey = try await {
             if let devSeed {
@@ -130,6 +129,7 @@ struct Boka: AsyncParsableCommand {
             }
         }()
 
+        logger.info("Network key: \(networkKey.publicKey.data.toHexString())")
         let networkConfig = NetworkConfig(
             mode: validator ? .validator : .builder,
             listenAddress: p2p,
@@ -144,7 +144,7 @@ struct Boka: AsyncParsableCommand {
             handlerMiddleware: .tracing(prefix: "Handler")
         )
 
-        let config = Node.Config(rpc: rpcConfig, network: networkConfig)
+        let config = Node.Config(rpc: rpcConfig, network: networkConfig, peers: peers)
 
         let node: Node = if validator {
             try await ValidatorNode(

Boka/Sources/Boka.swift

@@ -8,7 +8,7 @@ import msquic
     import Darwin
 #endif
 
-public struct NetAddr: Sendable {
+public struct NetAddr: Sendable, Equatable, Hashable {
     var quicAddr: QUIC_ADDR
 
     public init?(address: String) {
@@ -38,17 +38,13 @@ public struct NetAddr: Sendable {
         let (host, port, _) = parseQuicAddr(quicAddr) ?? ("::dead:beef", 0, false)
         return (host, port)
     }
-}
 
-extension NetAddr: Equatable {
     public static func == (lhs: NetAddr, rhs: NetAddr) -> Bool {
         var addr1 = lhs.quicAddr
         var addr2 = rhs.quicAddr
         return QuicAddrCompare(&addr1, &addr2) == 1
     }
-}
 
-extension NetAddr: Hashable {
     public func hash(into hasher: inout Hasher) {
         var addr = quicAddr
         let hash = QuicAddrHash(&addr)

Networking/Sources/MsQuicSwift/NetAddr.swift

@@ -250,6 +250,9 @@ private class ConnectionHandle {
 
         case QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE:
             logger.trace("Shutdown complete")
+            if let connection {
+                connection.handler.shutdownComplete(connection)
+            }
             if event.pointee.SHUTDOWN_COMPLETE.AppCloseInProgress == 0 {
                 // avoid closing twice
                 api.call { api in

Networking/Sources/MsQuicSwift/QuicConnection.swift

@@ -33,6 +33,7 @@ public protocol QuicEventHandler: Sendable {
     func shouldOpen(_ connection: QuicConnection, certificate: Data?) -> QuicStatus
     func connected(_ connection: QuicConnection)
     func shutdownInitiated(_ connection: QuicConnection, reason: ConnectionCloseReason)
+    func shutdownComplete(_ connection: QuicConnection)
     func streamStarted(_ connect: QuicConnection, stream: QuicStream)
 
     // stream events
@@ -55,6 +56,7 @@ extension QuicEventHandler {
     public func connected(_: QuicConnection) {}
 
     public func shutdownInitiated(_: QuicConnection, reason _: ConnectionCloseReason) {}
+    public func shutdownComplete(_: QuicConnection) {}
 
     public func streamStarted(_: QuicConnection, stream _: QuicStream) {}
 
@@ -69,6 +71,7 @@ public final class MockQuicEventHandler: QuicEventHandler {
         case shouldOpen(connection: QuicConnection, certificate: Data?)
         case connected(connection: QuicConnection)
         case shutdownInitiated(connection: QuicConnection, reason: ConnectionCloseReason)
+        case shutdownComplete(connection: QuicConnection)
         case streamStarted(connection: QuicConnection, stream: QuicStream)
         case dataReceived(stream: QuicStream, data: Data)
         case closed(stream: QuicStream, status: QuicStatus, code: QuicErrorCode)
@@ -107,6 +110,12 @@ public final class MockQuicEventHandler: QuicEventHandler {
         }
     }
 
+    public func shutdownComplete(_ connection: QuicConnection) {
+        events.write { events in
+            events.append(.shutdownComplete(connection: connection))
+        }
+    }
+
     public func streamStarted(_ connect: QuicConnection, stream: QuicStream) {
         events.write { events in
             events.append(.streamStarted(connection: connect, stream: stream))