CE 134: Work-package sharing

Blockchain/Sources/Blockchain/RuntimeProtocols/RuntimeEvents.swift

@@ -55,19 +55,54 @@
     // When a work package is recived via CE133
     public struct WorkPackagesReceived: Event {
         public let coreIndex: CoreIndex
-        public let workPackageRef: WorkPackageRef
+        public let workPackage: WorkPackageRef
         public let extrinsics: [Data]
 
-        public init(coreIndex: CoreIndex, workPackageRef: WorkPackageRef, extrinsics: [Data]) {
+        public init(coreIndex: CoreIndex, workPackage: WorkPackageRef, extrinsics: [Data]) {
             self.coreIndex = coreIndex
-            self.workPackageRef = workPackageRef
+            self.workPackage = workPackage
             self.extrinsics = extrinsics
         }
     }
 
     // When a work package bundle is ready to shared via CE134
     public struct WorkPackageBundleReady: Event {
+        public let coreIndex: CoreIndex
         public let bundle: WorkPackageBundle
+        public let segmentsRootMappings: SegmentsRootMappings
+
+        public init(
+            coreIndex: CoreIndex,
+            bundle: WorkPackageBundle,
+            segmentsRootMappings: SegmentsRootMappings
+        ) {
+            self.coreIndex = coreIndex
+            self.bundle = bundle
+            self.segmentsRootMappings = segmentsRootMappings
+        }
+    }
+
+    // When a work package will be shared via CE134
+    public struct WorkPackageShare: Event {
+        public let coreIndex: CoreIndex
+        public let workPackage: WorkPackageRef
+        public let extrinsics: [Data]
+
+        public init(coreIndex: CoreIndex, workPackage: WorkPackageRef, extrinsics: [Data]) {
+            self.coreIndex = coreIndex
+            self.workPackage = workPackage
+            self.extrinsics = extrinsics
+        }
+    }
+
+    // When a work package bundle is recived via CE134
+    public struct WorkPackageBundleRecived: Event {
+        public let workPackageHash: Data32
+        public let edd25519Signature: Data64
+        public init(workPackageHash: Data32, edd25519Signature: Data64) {
+            self.workPackageHash = workPackageHash
+            self.edd25519Signature = edd25519Signature
+        }
     }
 
     // When a work report is generated and ready to be distrubuted via CE135

Blockchain/Sources/Blockchain/Types/SegmentsRootMappings.swift

@@ -0,0 +1,9 @@
+import Foundation
+import Utils
+
+public struct SegmentsRootMapping: Sendable, Equatable, Codable {
+    public let workPackageHash: Data32
+    public let segmentsRoot: Data32
+}
+
+public typealias SegmentsRootMappings = [SegmentsRootMapping]

Blockchain/Sources/Blockchain/Validator/GuaranteeingService.swift

@@ -6,6 +6,9 @@
 public enum GuaranteeingServiceError: Error {
     case noAuthorizerHash
     case invalidExports
+    case invalidWorkPackage
+    case invalidBundle
+    case segmentsRootNotFound
 }
 
 public final class GuaranteeingService: ServiceBase2, @unchecked Sendable {
@@ -35,9 +38,17 @@
 
         super.init(id: "GuaranteeingService", config: config, eventBus: eventBus, scheduler: scheduler)
 
+        await subscribe(RuntimeEvents.WorkPackageShare.self, id: "GuaranteeingService.ShareWorkPackage") { [weak self] event in
+            try await self?.on(workPackagSharee: event)
+        }
+
         await subscribe(RuntimeEvents.WorkPackagesReceived.self, id: "GuaranteeingService.WorkPackagesReceived") { [weak self] event in
             try await self?.on(workPackagesReceived: event)
         }
+
+        await subscribe(RuntimeEvents.WorkPackageBundleReady.self, id: "GuaranteeingService.WorkPackageBundleReady") { [weak self] event in
+            try await self?.on(workPackageBundleReady: event)
+        }
     }
 
     public func onSyncCompleted() async {
@@ -78,16 +89,101 @@
         }
     }
 
+    private func on(workPackagSharee event: RuntimeEvents.WorkPackageShare) async throws {
+        guard try validate(workPackage: event.workPackage.value) else {
+            logger.error("Invalid work package: \(event.workPackage)")
+            throw GuaranteeingServiceError.invalidWorkPackage
+        }
+        // TODO: sometings need to do
+    }
+
+    private func on(workPackageBundleReady event: RuntimeEvents.WorkPackageBundleReady) async throws {
+        try await receiveWorkPackageBundleReady(
+            coreIndex: event.coreIndex,
+            segmentsRootMappings: event.segmentsRootMappings,
+            bundle: event.bundle
+        )
+    }
+
+    private func on(workPackageBundleReceived _: RuntimeEvents.WorkPackageBundleRecived) async throws {
+        // TODO: check somethings
+    }
+
+    // Method to receive a work package bundle ready
+    private func receiveWorkPackageBundleReady(
+        coreIndex _: CoreIndex,
+        segmentsRootMappings: SegmentsRootMappings,
+        bundle: WorkPackageBundle
+    ) async throws {
+        // Perform basic verification
+        guard try validateWorkPackageBundle(bundle, segmentsRootMappings: segmentsRootMappings) else {
+            throw GuaranteeingServiceError.invalidBundle
+        }
+    }
+
+    private func validateWorkPackageBundle(
+        _ bundle: WorkPackageBundle,
+        segmentsRootMappings: SegmentsRootMappings
+    ) throws -> Bool {
+        // Validate the work package authorization
+        guard try validateAuthorization(bundle.workPackage) else {
+            return false
+        }
+
+        // Validate the segments-root mappings
+        for mapping in segmentsRootMappings {
+            guard try validateSegmentsRootMapping(mapping, for: bundle.workPackage) else {
+                return false
+            }
+        }
+
+        return true
+    }
+
     private func on(workPackagesReceived event: RuntimeEvents.WorkPackagesReceived) async throws {
-        try await refinePkg(package: event.workPackageRef)
+        try await handleWorkPackage(coreIndex: event.coreIndex, workPackage: event.workPackage, extrinsics: event.extrinsics)
     }
 
-    private func refinePkg(package: WorkPackageRef) async throws {
+    // handle Work Package
+    public func handleWorkPackage(coreIndex: CoreIndex, workPackage: WorkPackageRef, extrinsics: [Data]) async throws {
+        // Validate the work package
+        guard try validate(workPackage: workPackage.value) else {
+            logger.error("Invalid work package: \(workPackage)")
+            throw GuaranteeingServiceError.invalidWorkPackage
+        }
         guard let (validatorIndex, signingKey) = signingKey.value else {
             logger.debug("not in current validator set, skipping refine")
             return
         }
 
+        // Share work package
+        let shareWorkPackageEvent = RuntimeEvents.WorkPackageShare(coreIndex: coreIndex, workPackage: workPackage, extrinsics: extrinsics)
+        publish(shareWorkPackageEvent)
+
+        // check & refine
+        let (bundle, mappings, workReport) = try await refinePkg(
+            validatorIndex: validatorIndex,
+            workPackage: workPackage,
+            extrinsics: extrinsics
+        )
+
+        // Share work bundle
+        let shareWorkBundleEvent = RuntimeEvents.WorkPackageBundleReady(
+            coreIndex: coreIndex,
+            bundle: bundle,
+            segmentsRootMappings: mappings
+        )
+        publish(shareWorkBundleEvent)
+        // Sign work report & work-report distribution via CE 135
+        let payload = SigningContext.guarantee + workReport.hash().data
+        let signature = try signingKey.sign(message: payload)
+        let workReportEvent = RuntimeEvents.WorkReportGenerated(item: workReport, signature: signature)
+        publish(workReportEvent)
+    }
+
+    private func refinePkg(validatorIndex: ValidatorIndex, workPackage: WorkPackageRef,
+                           extrinsics: [Data]) async throws -> (WorkPackageBundle, SegmentsRootMappings, WorkReport)
+    {
         let state = try await dataProvider.getState(hash: dataProvider.bestHead.hash)
 
         // TODO: check for edge cases such as epoch end
@@ -96,24 +192,63 @@
             randomness: state.value.entropyPool.t2,
             timeslot: state.value.timeslot + 1
         )
+        // TODO: coreIndex equal with shareWorkPackage coreIndex?
         guard let coreIndex = currentCoreAssignment[safe: Int(validatorIndex)] else {
             try throwUnreachable("invalid validator index/core assignment")
         }
 
-        let workReport = try await createWorkReport(for: package, coreIndex: coreIndex)
-        let payload = SigningContext.guarantee + workReport.hash().data
-        let signature = try signingKey.sign(message: payload)
-        let event = RuntimeEvents.WorkReportGenerated(item: workReport, signature: signature)
-        publish(event)
+        // Create work report & WorkPackageBundle
+        return try await createWorkReport(
+            coreIndex: coreIndex,
+            workPackage: workPackage,
+            extrinsics: extrinsics
+        )
+    }
+
+    private func validateSegmentsRootMapping(
+        _: SegmentsRootMapping,
+        for _: WorkPackage
+    ) throws -> Bool {
+        // TODO: Implement logic to validate the segments-root mapping
+        true // Placeholder
+    }
+
+    private func validateAuthorization(_: WorkPackage) throws -> Bool {
+        // TODO: Implement logic to validate the work package authorization
+        true // Placeholder
+    }
+
+    // TODO: Add validate func
+    private func validate(workPackage _: WorkPackage) throws -> Bool {
+        // 1. Check if it is possible to generate a work-report
+        // 2. Check all import segments have been retrieved
+        true
+    }
+
+    private func retrieveImportSegments(for _: WorkPackage) async throws -> [Data4104] {
+        // TODO: Implement retrieveImportSegments
+        // Implement logic to retrieve imported data segments
+        // For example, fetch from the data availability layer
+        [] // Placeholder
+    }
+
+    private func retrieveJustifications(for _: WorkPackage) async throws -> [Data] {
+        // TODO: Implement retrieveJustifications
+        // Implement logic to retrieve justifications for the imported segments
+        // For example, fetch proofs from the data availability layer
+        [] // Placeholder
     }
 
     // workpackage -> workresult -> workreport
-    private func createWorkReport(for workPackage: WorkPackageRef, coreIndex: CoreIndex) async throws -> WorkReport {
+    private func createWorkReport(coreIndex: CoreIndex, workPackage: WorkPackageRef,
+                                  extrinsics: [Data]) async throws -> (WorkPackageBundle, SegmentsRootMappings, WorkReport)
+    {
         let state = try await dataProvider.getState(hash: dataProvider.bestHead.hash)
         let packageHash = workPackage.hash
         let corePool = state.value.coreAuthorizationPool[coreIndex]
         let authorizerHash = try corePool.array.first.unwrap(orError: GuaranteeingServiceError.noAuthorizerHash)
         var exportSegmentOffset: UInt16 = 0
+        var mappings: SegmentsRootMappings = []
         // B.2. the authorization output, the result of the Is-Authorized function
         // TODO: waiting for authorizationFunction done  Mock a result
         // let res = try await authorizationFunction.invoke(config: config, serviceAccounts: state.value, package: workPackage, coreIndex:
@@ -160,16 +295,16 @@
 
                 exportSegments.append(contentsOf: refineRes.exports)
             }
-
-            let (erasureRoot, length) = try await dataAvailability.exportWorkpackageBundle(bundle: WorkPackageBundle(
+            let bundle = try await WorkPackageBundle(
                 workPackage: workPackage.value,
-                extrinsic: [], // TODO: get extrinsic data
-                importSegments: [],
-                justifications: []
-            ))
+                extrinsic: extrinsics,
+                importSegments: retrieveImportSegments(for: workPackage.value),
+                justifications: retrieveJustifications(for: workPackage.value)
+            )
+            let (erasureRoot, length) = try await dataAvailability.exportWorkpackageBundle(bundle: bundle)
 
             let segmentRoot = try await dataAvailability.exportSegments(data: exportSegments, erasureRoot: erasureRoot)
-
+            mappings.append(SegmentsRootMapping(workPackageHash: packageHash, segmentsRoot: segmentRoot))
             // TODO: generate or find AvailabilitySpecifications  14.4.1 work-package bundle
             let packageSpecification = AvailabilitySpecifications(
                 workPackageHash: packageHash,
@@ -183,15 +318,15 @@
             for item in state.value.recentHistory.items {
                 oldLookups.merge(item.lookup, uniquingKeysWith: { _, new in new })
             }
-            return try WorkReport(
+            return try (bundle, mappings, WorkReport(
                 authorizerHash: authorizerHash,
                 coreIndex: coreIndex,
                 authorizationOutput: authorizationOutput,
                 refinementContext: workPackage.value.context,
                 packageSpecification: packageSpecification,
                 lookup: oldLookups,
                 results: ConfigLimitedSizeArray(config: config, array: workResults)
-            )
+            ))
 
         case let .failure(error):
             logger.error("Authorization failed with error: \(error)")

Blockchain/Tests/BlockchainTests/GuaranteeingServiceTests.swift

@@ -39,4 +39,14 @@ struct GuaranteeingServiceTests {
         #expect(signingKey.0 == 0)
         #expect(signingKey.1.publicKey == publicKey)
     }
+
+    @Test func workPackagesReceived() async throws {
+        let (services, guaranteeingService) = try await setup(keysCount: 0)
+
+        await guaranteeingService.onSyncCompleted()
+
+        let workpackage = WorkPackage.dummy(config: services.config)
+        await services.eventBus
+            .publish(RuntimeEvents.WorkPackagesReceived(coreIndex: 0, workPackage: workpackage.asRef(), extrinsics: []))
+    }
 }

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

@@ -8,6 +8,7 @@
     case safroleTicket1(SafroleTicketMessage)
     case safroleTicket2(SafroleTicketMessage)
     case workPackageSubmission(WorkPackageMessage)
+    case workPackageSharing(WorkPackageMessage)
 }
 
 extension CERequest: RequestProtocol {
@@ -23,6 +24,8 @@
             try JamEncoder.encode(message)
         case let .workPackageSubmission(message):
             try JamEncoder.encode(message)
+        case let .workPackageSharing(message):
+            try JamEncoder.encode(message)
         }
     }
 
@@ -36,6 +39,8 @@
             .safroleTicket2
         case .workPackageSubmission:
             .workPackageSubmission
+        case .workPackageSharing:
+            .workPackageSharing
         }
     }
 
@@ -49,6 +54,8 @@
             SafroleTicketMessage.self
         case .workPackageSubmission:
             WorkPackageMessage.self
+        case .workPackageSharing:
+            WorkPackageMessage.self
         default:
             fatalError("unimplemented")
         }
@@ -74,6 +81,9 @@
         case .workPackageSubmission:
             guard let message = data as? WorkPackageMessage else { return nil }
             return .workPackageSubmission(message)
+        case .workPackageSharing:
+            guard let message = data as? WorkPackageMessage else { return nil }
+            return .workPackageSharing(message)
         default:
             fatalError("unimplemented")
         }

Node/Sources/Node/NetworkingProtocol/NetworkManager.swift

@@ -232,7 +232,18 @@
                     event: RuntimeEvents
                         .WorkPackagesReceived(
                             coreIndex: message.coreIndex,
-                            workPackageRef: message.workPackage.asRef(),
+                            workPackage: message.workPackage.asRef(),
+                            extrinsics: message.extrinsics
+                        )
+                )
+            return []
+        case let .workPackageSharing(message):
+            blockchain
+                .publish(
+                    event: RuntimeEvents
+                        .WorkPackageShare(
+                            coreIndex: message.coreIndex,
+                            workPackage: message.workPackage.asRef(),
                             extrinsics: message.extrinsics
                         )
                 )