Work-package submission

Blockchain/Sources/Blockchain/RuntimeProtocols/RuntimeEvents.swift

@@ -52,6 +52,15 @@ public enum RuntimeEvents {
         public let block: BlockRef
     }
 
+    // New WorkPackagesReceived from network
+    public struct WorkPackagesReceived: Event {
+        public let items: [WorkPackage]
+
+        public init(items: [WorkPackage]) {
+            self.items = items
+        }
+    }
+
     // New WorkPackagesGenerated by Guaranteeing Service
     public struct WorkPackagesGenerated: Event {
         public let items: [WorkPackage]

Blockchain/Sources/Blockchain/Types/RefinementContext.swift

@@ -3,8 +3,8 @@ import Utils
 
 // A refinement context, denoted by the set X, describes the context of the chain
 // at the point that the report’s corresponding work-package was evaluated.
-public struct RefinementContext: Comparable, Sendable, Equatable, Codable {
-    public struct Anchor: Comparable, Sendable, Equatable, Codable {
+public struct RefinementContext: Comparable, Sendable, Equatable, Codable, Hashable {
+    public struct Anchor: Comparable, Sendable, Equatable, Codable, Hashable {
         // a
         public var headerHash: Data32
         // s
@@ -74,6 +74,12 @@ public struct RefinementContext: Comparable, Sendable, Equatable, Codable {
         }
         return lhs.lookupAnchor < rhs.lookupAnchor
     }
+
+    public func hash(into hasher: inout Hasher) {
+        hasher.combine(anchor)
+        hasher.combine(lookupAnchor)
+        hasher.combine(prerequisiteWorkPackages)
+    }
 }
 
 extension RefinementContext: Dummy {

Blockchain/Sources/Blockchain/Types/WorkItem.swift

@@ -2,9 +2,9 @@ import Foundation
 import Utils
 
 // I
-public struct WorkItem: Sendable, Equatable, Codable {
-    public struct ImportedDataSegment: Sendable, Equatable, Codable {
-        public enum DataSegmentRootKind: Sendable, Equatable {
+public struct WorkItem: Sendable, Equatable, Codable, Hashable {
+    public struct ImportedDataSegment: Sendable, Equatable, Codable, Hashable {
+        public enum DataSegmentRootKind: Sendable, Equatable, Hashable {
             case segmentRoot(Data32)
             case workPackageHash(Data32)
         }

Blockchain/Sources/Blockchain/Types/WorkPackage.swift

@@ -60,6 +60,18 @@ public struct WorkPackage: Comparable, Sendable, Equatable, Codable {
     }
 }
 
+extension WorkPackage: Hashable {
+    public func hash(into hasher: inout Hasher) {
+        hasher.combine(authorizationToken)
+        hasher.combine(authorizationServiceIndex)
+        hasher.combine(authorizationCodeHash)
+        hasher.combine(parameterizationBlob)
+        hasher.combine(context)
+        hasher.combine(workItems.count)
+        workItems.forEach { hasher.combine($0) }
+    }
+}
+
 extension WorkPackage {
     public func hash() -> Data32 {
         try! JamEncoder.encode(self).blake2b256hash()

Blockchain/Sources/Blockchain/Validator/ExtrinsicPools/WorkPackagePoolService.swift

@@ -64,8 +64,10 @@ public final class WorkPackagePoolService: ServiceBase, @unchecked Sendable {
         await subscribe(RuntimeEvents.WorkPackagesGenerated.self, id: "WorkPackagePool.WorkPackagesGenerated") { [weak self] event in
             try await self?.on(workPackagesGenerated: event)
         }
+        await subscribe(RuntimeEvents.WorkPackagesReceived.self, id: "WorkPackagePool.WorkPackagesReceived") { [weak self] event in
+            try await self?.on(workPackagesReceived: event)
+        }
         // TODO: add remove subscribe
-        // TODO: add receive subscribe?
     }
 
     private func on(workPackagesGenerated event: RuntimeEvents.WorkPackagesGenerated) async throws {
@@ -74,6 +76,12 @@ public final class WorkPackagePoolService: ServiceBase, @unchecked Sendable {
         await storage.add(packages: event.items, config: config)
     }
 
+    private func on(workPackagesReceived event: RuntimeEvents.WorkPackagesReceived) async throws {
+        let state = try await dataProvider.getBestState()
+        try await storage.update(state: state, config: config)
+        await storage.add(packages: event.items, config: config)
+    }
+
     public func update(state: StateRef, config: ProtocolConfigRef) async throws {
         try await storage.update(state: state, config: config)
     }

Blockchain/Tests/BlockchainTests/WorkPackagePoolServiceTests.swift

@@ -40,11 +40,26 @@ struct WorkPackagePoolServiceTests {
             allWorkPackages.append(workpackage)
         }
         await services.eventBus.publish(RuntimeEvents.WorkPackagesGenerated(items: allWorkPackages))
-        let workPackages = await workPackagecPoolService.getWorkPackages()
+        var workPackages = await workPackagecPoolService.getWorkPackages()
         #expect(workPackages.array == Array(allWorkPackages).sorted())
         let workpackage = WorkPackage.dummy(config: services.config)
         try await workPackagecPoolService.addWorkPackages(packages: [workpackage])
         try await workPackagecPoolService.removeWorkPackages(packages: [workpackage])
         #expect(workPackages.array.count == services.config.value.totalNumberOfCores)
+        let event = RuntimeEvents.WorkPackagesReceived(items: [workpackage])
+        await services.eventBus.publish(event)
+        await services.eventBus.publish(event) // duplicate
+        // Wait for the event to be processed
+        await services.storeMiddleware.wait()
+        workPackages = await workPackagecPoolService.getWorkPackages()
+        #expect(workPackages.array.count > services.config.value.totalNumberOfCores)
+        await services.blockchain.publish(event: RuntimeEvents.WorkPackagesReceived(
+            items: [
+                workpackage,
+            ]
+        ))
+        await services.storeMiddleware.wait()
+        workPackages = await workPackagecPoolService.getWorkPackages()
+        #expect(workPackages.array.count > services.config.value.totalNumberOfCores)
     }
 }

Node/Sources/Node/NetworkingProtocol/CommonEphemeral/CERequest.swift

@@ -7,6 +7,7 @@
     case blockRequest(BlockRequest)
     case safroleTicket1(SafroleTicketMessage)
     case safroleTicket2(SafroleTicketMessage)
+    case workPackageSubmission(WorkPackageMessage)
 }
 
 extension CERequest: RequestProtocol {
@@ -20,6 +21,8 @@
             try JamEncoder.encode(message)
         case let .safroleTicket2(message):
             try JamEncoder.encode(message)
+        case let .workPackageSubmission(message):
+            try JamEncoder.encode(message)
         }
     }
 
@@ -31,6 +34,8 @@
             .safroleTicket1
         case .safroleTicket2:
             .safroleTicket2
+        case .workPackageSubmission:
+            .workPackageSubmission
         }
     }
 
@@ -42,6 +47,8 @@
             SafroleTicketMessage.self
         case .safroleTicket2:
             SafroleTicketMessage.self
+        case .workPackageSubmission:
+            WorkPackageMessage.self
         default:
             fatalError("unimplemented")
         }
@@ -64,6 +71,9 @@
                 return nil
             }
             return .safroleTicket2(message)
+        case .workPackageSubmission:
+            guard let message = data as? WorkPackageMessage else { return nil }
+            return .workPackageSubmission(message)
         default:
             fatalError("unimplemented")
         }

Node/Sources/Node/NetworkingProtocol/CommonEphemeral/WorkPackage.swift

@@ -0,0 +1,20 @@
+import Blockchain
+import Codec
+import Foundation
+
+public struct WorkPackageMessage: Codable, Sendable, Equatable, Hashable {
+    /// The core index associated with the work-package.
+    public var coreIndex: CoreIndex
+
+    /// The work-package data.
+    public var workPackage: WorkPackage
+
+    /// The extrinsic data referenced by the work-package.
+    public var extrinsics: [Data]
+
+    public init(coreIndex: CoreIndex, workPackage: WorkPackage, extrinsics: [Data]) {
+        self.coreIndex = coreIndex
+        self.workPackage = workPackage
+        self.extrinsics = extrinsics
+    }
+}

Node/Sources/Node/NetworkingProtocol/NetworkManager.swift

@@ -226,6 +226,13 @@
                 ]
             ))
             return []
+        case let .workPackageSubmission(message):
+            blockchain.publish(event: RuntimeEvents.WorkPackagesReceived(
+                items: [
+                    message.workPackage,
+                ]
+            ))
+            return []
         }
     }
 

Node/Tests/NodeTests/NetworkManagerTests.swift

@@ -74,6 +74,34 @@ struct NetworkManagerTests {
         )
     }
 
+    @Test
+    func testWorkPackagesReceived() async throws {
+        // Create dummy work packages
+        let workPackages = [
+            WorkPackage.dummy(config: services.config),
+        ]
+
+        // Publish WorkPackagesReceived event
+        await services.eventBus.publish(RuntimeEvents.WorkPackagesReceived(items: workPackages))
+
+        // Wait for event processing
+        await storeMiddleware.wait()
+
+        #expect(workPackages.first?.hash() != nil)
+        #expect(workPackages.first?.hashValue != nil)
+        // Verify network calls
+        #expect(
+            network.contain(calls: [
+                .init(function: "connect", parameters: ["address": devPeers.first!, "role": PeerRole.validator]),
+                .init(function: "sendToPeer", parameters: [
+                    "message": CERequest.workPackageSubmission(
+                        WorkPackageMessage(coreIndex: 0, workPackage: workPackages[0], extrinsics: [])
+                    ),
+                ]),
+            ])
+        )
+    }
+
     @Test
     func testBlockBroadcast() async throws {
         // Import a block

Utils/Sources/Utils/SaturatingNumber.swift

@@ -1,7 +1,7 @@
 import Codec
 import Numerics
 
-public struct SaturatingNumber<T: FixedWidthInteger & Sendable>: Sendable {
+public struct SaturatingNumber<T: FixedWidthInteger & Sendable>: Sendable, Hashable {
     public private(set) var value: T
 
     public static var max: SaturatingNumber {