CE 134: Work-package sharing (#303)

MacOMNI · web-flow · commit ea1b6ca96a45 · 2025-03-04T14:20:13.000+13:00
* update issue 278

* update issues 279

* Work-package submission

* update networkmanager

* update more test

* update workpackage hash

* update networkmanager

* update workpackage

* update workpackpool

* update NetworkManagerTest

* update workpackagepoolserver

* update SegmentsRootMapping

* update Guaranteeing

* udpate guaranteeing

* shareWorkPackage adjust

* update shareWorkPackage

* update local

* update shareWorkPackage

* update

* update handleWorkPackage

* update ShareWorkPackage

* update Work-package sharing

* update WorkPackageShare

* update GuaranteeingServiceTests

* no message

* TODO: add more tests

* update work package share
diff --git a/Blockchain/Sources/Blockchain/RuntimeProtocols/RuntimeEvents.swift b/Blockchain/Sources/Blockchain/RuntimeProtocols/RuntimeEvents.swift
@@ -55,19 +55,41 @@ public enum RuntimeEvents {
     // 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 struct WorkPackageBundleShare: 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 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
diff --git a/Blockchain/Sources/Blockchain/Types/SegmentsRootMappings.swift b/Blockchain/Sources/Blockchain/Types/SegmentsRootMappings.swift
@@ -0,0 +1,9 @@
+import Foundation
+import Utils
+
+public struct SegmentsRootMapping: Sendable, Equatable, Codable, Hashable {
+    public let workPackageHash: Data32
+    public let segmentsRoot: Data32
+}
+
+public typealias SegmentsRootMappings = [SegmentsRootMapping]
diff --git a/Blockchain/Sources/Blockchain/Types/WorkPackageBundle.swift b/Blockchain/Sources/Blockchain/Types/WorkPackageBundle.swift
@@ -2,9 +2,15 @@ import Foundation
 import Utils
 
 // All the necessary data to audit a work package. Stored in audits DA
-public struct WorkPackageBundle: Sendable, Equatable, Codable {
+public struct WorkPackageBundle: Sendable, Equatable, Codable, Hashable {
     public var workPackage: WorkPackage
     public var extrinsic: [Data]
     public var importSegments: [Data4104]
     public var justifications: [Data]
+    public init(workPackage: WorkPackage, extrinsic: [Data], importSegments: [Data4104], justifications: [Data]) {
+        self.workPackage = workPackage
+        self.extrinsic = extrinsic
+        self.importSegments = importSegments
+        self.justifications = justifications
+    }
 }
diff --git a/Blockchain/Sources/Blockchain/Validator/GuaranteeingService.swift b/Blockchain/Sources/Blockchain/Validator/GuaranteeingService.swift
@@ -6,6 +6,9 @@ import Utils
 public enum GuaranteeingServiceError: Error {
     case noAuthorizerHash
     case invalidExports
+    case invalidWorkPackage
+    case invalidBundle
+    case segmentsRootNotFound
 }
 
 public final class GuaranteeingService: ServiceBase2, @unchecked Sendable {
@@ -38,6 +41,10 @@ public final class GuaranteeingService: ServiceBase2, @unchecked Sendable {
         await subscribe(RuntimeEvents.WorkPackagesReceived.self, id: "GuaranteeingService.WorkPackagesReceived") { [weak self] event in
             try await self?.on(workPackagesReceived: event)
         }
+
+        await subscribe(RuntimeEvents.WorkPackageBundleShare.self, id: "GuaranteeingService.WorkPackageBundleShare") { [weak self] event in
+            try await self?.on(workPackageBundle: event)
+        }
     }
 
     public func onSyncCompleted() async {
@@ -78,16 +85,89 @@ public final class GuaranteeingService: ServiceBase2, @unchecked Sendable {
         }
     }
 
+    private func on(workPackageBundle event: RuntimeEvents.WorkPackageBundleShare) async throws {
+        try await receiveWorkPackageBundle(
+            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
+    private func receiveWorkPackageBundle(
+        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
         }
 
+        // check & refine
+        let (bundle, mappings, workReport) = try await refinePkg(
+            validatorIndex: validatorIndex,
+            workPackage: workPackage,
+            extrinsics: extrinsics
+        )
+
+        // Share work package bundle
+        let shareWorkBundleEvent = RuntimeEvents.WorkPackageBundleShare(
+            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 +176,63 @@ public final class GuaranteeingService: ServiceBase2, @unchecked Sendable {
             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 +279,16 @@ public final class GuaranteeingService: ServiceBase2, @unchecked Sendable {
 
                 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 +302,15 @@ public final class GuaranteeingService: ServiceBase2, @unchecked Sendable {
             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)")
diff --git a/Blockchain/Tests/BlockchainTests/GuaranteeingServiceTests.swift b/Blockchain/Tests/BlockchainTests/GuaranteeingServiceTests.swift
@@ -39,4 +39,20 @@ struct GuaranteeingServiceTests {
         #expect(signingKey.0 == 0)
         #expect(signingKey.1.publicKey == publicKey)
     }
+    // TODO: add more tests
+//    @Test func workPackagesReceived() async throws {
+//        let (services, guaranteeingService) = try await setup()
+//
+//        await guaranteeingService.onSyncCompleted()
+//        let workpackage = WorkPackage(
+//            authorizationToken: Data(repeating: 0x00, count: 32),
+//            authorizationServiceIndex: 0,
+//            authorizationCodeHash: Data32(),
+//            parameterizationBlob: Data(),
+//            context: RefinementContext.dummy(config: services.config),
+//            workItems: try! ConfigLimitedSizeArray(config: services.config, defaultValue: WorkItem.dummy(config: services.config))
+//        )
+//        await services.eventBus
+//            .publish(RuntimeEvents.WorkPackagesReceived(coreIndex: 0, workPackage: workpackage.asRef(), extrinsics: []))
+//    }
 }
diff --git a/Node/Sources/Node/NetworkingProtocol/CommonEphemeral/CERequest.swift b/Node/Sources/Node/NetworkingProtocol/CommonEphemeral/CERequest.swift
@@ -8,6 +8,7 @@ public enum CERequest: Sendable, Equatable, Hashable {
     case safroleTicket1(SafroleTicketMessage)
     case safroleTicket2(SafroleTicketMessage)
     case workPackageSubmission(WorkPackageMessage)
+    case workPackageSharing(WorkPackageShareMessage)
 }
 
 extension CERequest: RequestProtocol {
@@ -23,6 +24,8 @@ extension CERequest: RequestProtocol {
             try JamEncoder.encode(message)
         case let .workPackageSubmission(message):
             try JamEncoder.encode(message)
+        case let .workPackageSharing(message):
+            try JamEncoder.encode(message)
         }
     }
 
@@ -36,6 +39,8 @@ extension CERequest: RequestProtocol {
             .safroleTicket2
         case .workPackageSubmission:
             .workPackageSubmission
+        case .workPackageSharing:
+            .workPackageSharing
         }
     }
 
@@ -49,6 +54,8 @@ extension CERequest: RequestProtocol {
             SafroleTicketMessage.self
         case .workPackageSubmission:
             WorkPackageMessage.self
+        case .workPackageSharing:
+            WorkPackageShareMessage.self
         default:
             fatalError("unimplemented")
         }
@@ -74,6 +81,9 @@ extension CERequest: RequestProtocol {
         case .workPackageSubmission:
             guard let message = data as? WorkPackageMessage else { return nil }
             return .workPackageSubmission(message)
+        case .workPackageSharing:
+            guard let message = data as? WorkPackageShareMessage else { return nil }
+            return .workPackageSharing(message)
         default:
             fatalError("unimplemented")
         }
diff --git a/Node/Sources/Node/NetworkingProtocol/CommonEphemeral/WorkPackageShareMessage.swift b/Node/Sources/Node/NetworkingProtocol/CommonEphemeral/WorkPackageShareMessage.swift
@@ -0,0 +1,19 @@
+import Blockchain
+import Codec
+import Foundation
+
+public struct WorkPackageShareMessage: Codable, Sendable, Equatable, Hashable {
+    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
+    }
+}
diff --git a/Node/Sources/Node/NetworkingProtocol/NetworkManager.swift b/Node/Sources/Node/NetworkingProtocol/NetworkManager.swift
diff --git a/Node/Tests/NodeTests/NetworkManagerTests.swift b/Node/Tests/NodeTests/NetworkManagerTests.swift

Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@ public enum CERequest: Sendable, Equatable, Hashable {`
`8`	`8`	`case safroleTicket1(SafroleTicketMessage)`
`9`	`9`	`case safroleTicket2(SafroleTicketMessage)`
`10`	`10`	`case workPackageSubmission(WorkPackageMessage)`
	`11`	`+ case workPackageSharing(WorkPackageShareMessage)`
`11`	`12`	`}`
`12`	`13`
`13`	`14`	`extension CERequest: RequestProtocol {`
`@@ -23,6 +24,8 @@ extension CERequest: RequestProtocol {`
`23`	`24`	`try JamEncoder.encode(message)`
`24`	`25`	`case let .workPackageSubmission(message):`
`25`	`26`	`try JamEncoder.encode(message)`
	`27`	`+ case let .workPackageSharing(message):`
	`28`	`+ try JamEncoder.encode(message)`
`26`	`29`	`}`
`27`	`30`	`}`
`28`	`31`
`@@ -36,6 +39,8 @@ extension CERequest: RequestProtocol {`
`36`	`39`	`.safroleTicket2`
`37`	`40`	`case .workPackageSubmission:`
`38`	`41`	`.workPackageSubmission`
	`42`	`+ case .workPackageSharing:`
	`43`	`+ .workPackageSharing`
`39`	`44`	`}`
`40`	`45`	`}`
`41`	`46`
`@@ -49,6 +54,8 @@ extension CERequest: RequestProtocol {`
`49`	`54`	`SafroleTicketMessage.self`
`50`	`55`	`case .workPackageSubmission:`
`51`	`56`	`WorkPackageMessage.self`
	`57`	`+ case .workPackageSharing:`
	`58`	`+ WorkPackageShareMessage.self`
`52`	`59`	`default:`
`53`	`60`	`fatalError("unimplemented")`
`54`	`61`	`}`
`@@ -74,6 +81,9 @@ extension CERequest: RequestProtocol {`
`74`	`81`	`case .workPackageSubmission:`
`75`	`82`	`guard let message = data as? WorkPackageMessage else { return nil }`
`76`	`83`	`return .workPackageSubmission(message)`
	`84`	`+ case .workPackageSharing:`
	`85`	`+ guard let message = data as? WorkPackageShareMessage else { return nil }`
	`86`	`+ return .workPackageSharing(message)`
`77`	`87`	`default:`
`78`	`88`	`fatalError("unimplemented")`
`79`	`89`	`}`