Work-report distribution & Work-report request

Blockchain/Package.swift

@@ -20,7 +20,7 @@ let package = Package(
         .package(path: "../Utils"),
         .package(path: "../TracingUtils"),
         .package(path: "../PolkaVM"),
-        .package(url: "https://github.com/apple/swift-testing.git", branch: "0.10.0"),
+        .package(url: "https://github.com/apple/swift-testing.git", branch: "6.0.0"),
     ],
     targets: [
         // Targets are the basic building blocks of a package, defining a module or a test suite.

Blockchain/Sources/Blockchain/Blockchain.swift

@@ -33,7 +33,10 @@ public final class Blockchain: ServiceBase, @unchecked Sendable {
             logger.debug("block already imported", metadata: ["hash": "\(block.hash)"])
             return
         }
-
+        // TODO: if current block is light
+        // check if dataProvider.hasGuaranteedWorkReport
+        // send workReportDistribution waiting for response
+        // save to full block
         try await withSpan("importBlock") { span in
             span.attributes.blockHash = block.hash.description
 

Blockchain/Sources/Blockchain/BlockchainDataProvider/BlockchainDataProvider.swift

@@ -58,6 +58,18 @@ public actor BlockchainDataProvider {
 }
 
 extension BlockchainDataProvider {
+    public func hasGuaranteedWorkReport(hash: Data32) async throws -> Bool {
+        try await dataProvider.hasGuaranteedWorkReport(hash: hash)
+    }
+
+    public func getGuaranteedWorkReport(hash: Data32) async throws -> GuaranteedWorkReportRef? {
+        try await dataProvider.getGuaranteedWorkReport(hash: hash)
+    }
+
+    public func add(guaranteedWorkReport: GuaranteedWorkReportRef) async throws {
+        try await dataProvider.add(guaranteedWorkReport: guaranteedWorkReport)
+    }
+
     public func hasBlock(hash: Data32) async throws -> Bool {
         try await dataProvider.hasBlock(hash: hash)
     }

Blockchain/Sources/Blockchain/BlockchainDataProvider/BlockchainDataProviderProtocol.swift

@@ -1,6 +1,7 @@
 import Utils
 
 public protocol BlockchainDataProviderProtocol: Sendable {
+    func hasGuaranteedWorkReport(hash: Data32) async throws -> Bool
     func hasBlock(hash: Data32) async throws -> Bool
     func hasState(hash: Data32) async throws -> Bool
     func isHead(hash: Data32) async throws -> Bool
@@ -9,6 +10,8 @@ public protocol BlockchainDataProviderProtocol: Sendable {
 
     func getHeader(hash: Data32) async throws -> HeaderRef?
 
+    func getGuaranteedWorkReport(hash: Data32) async throws -> GuaranteedWorkReportRef?
+
     func getBlock(hash: Data32) async throws -> BlockRef?
 
     func getState(hash: Data32) async throws -> StateRef?
@@ -26,14 +29,15 @@ public protocol BlockchainDataProviderProtocol: Sendable {
     /// return empty set if not found
     func getBlockHash(byNumber number: UInt32) async throws -> Set<Data32>
 
+    func add(guaranteedWorkReport: GuaranteedWorkReportRef) async throws
     func add(block: BlockRef) async throws
     func add(state: StateRef) async throws
     func setFinalizedHead(hash: Data32) async throws
 
     /// throw BlockchainDataProviderError.noData if parent is not a head
     func updateHead(hash: Data32, parent: Data32) async throws
 
-    /// remove header, block and state
+    /// remove header, block, workReport, state
     func remove(hash: Data32) async throws
 
     var genesisBlockHash: Data32 { get }

Blockchain/Sources/Blockchain/BlockchainDataProvider/InMemoryDataProvider.swift

@@ -9,6 +9,7 @@ public actor InMemoryDataProvider {
     private var blockByHash: [Data32: BlockRef] = [:]
     private var stateByBlockHash: [Data32: StateRef] = [:]
     private var hashByTimeslot: [TimeslotIndex: Set<Data32>] = [:]
+    private var guaranteedWorkReports: [Data32: GuaranteedWorkReportRef] = [:]
     public let genesisBlockHash: Data32
 
     public init(genesisState: StateRef, genesisBlock: BlockRef) async {
@@ -22,6 +23,18 @@ public actor InMemoryDataProvider {
 }
 
 extension InMemoryDataProvider: BlockchainDataProviderProtocol {
+    public func hasGuaranteedWorkReport(hash: Data32) async throws -> Bool {
+        guaranteedWorkReports[hash] != nil
+    }
+
+    public func getGuaranteedWorkReport(hash: Data32) async throws -> GuaranteedWorkReportRef? {
+        guaranteedWorkReports[hash]
+    }
+
+    public func add(guaranteedWorkReport: GuaranteedWorkReportRef) async throws {
+        guaranteedWorkReports[guaranteedWorkReport.value.workReport.hash()] = guaranteedWorkReport
+    }
+
     public func getKeys(prefix: Data32, count: UInt32, startKey: Data32?, blockHash: Data32?) async throws -> [String] {
         guard let stateRef = try getState(hash: blockHash ?? genesisBlockHash) else {
             return []
@@ -120,6 +133,7 @@ extension InMemoryDataProvider: BlockchainDataProviderProtocol {
         let timeslot = blockByHash[hash]?.header.timeslot ?? stateByBlockHash[hash]?.value.timeslot
         stateByBlockHash.removeValue(forKey: hash)
         blockByHash.removeValue(forKey: hash)
+        guaranteedWorkReports.removeValue(forKey: hash)
 
         if let timeslot {
             hashByTimeslot[timeslot]?.remove(hash)

Blockchain/Sources/Blockchain/BlockchainServices.swift

@@ -23,6 +23,9 @@
     private var _guaranteeingService: GuaranteeingService?
     private weak var _guaranteeingServiceRef: GuaranteeingService?
 
+    private var _dataAvailabilityService: DataAvailabilityService?
+    private weak var _dataAvailabilityServiceRef: DataAvailabilityService?
+
     private let schedulerService: ServiceBase2
 
     public init(
@@ -63,6 +66,7 @@
         _blockchain = nil
         _blockAuthor = nil
         _guaranteeingService = nil
+        _dataAvailabilityService = nil
 
         if let _blockchainRef {
             fatalError("BlockchainServices: blockchain still alive. retain count: \(_getRetainCount(_blockchainRef))")
@@ -75,6 +79,30 @@
         if let _guaranteeingServiceRef {
             fatalError("BlockchainServices: guaranteeingService still alive. retain count: \(_getRetainCount(_guaranteeingServiceRef))")
         }
+
+        if let _dataAvailabilityServiceRef {
+            fatalError(
+                "BlockchainServices: dataAvailabilityService still alive. retain count: \(_getRetainCount(_dataAvailabilityServiceRef))"
+            )
+        }
+    }
+
+    public var dataAvailabilityService: DataAvailabilityService {
+        get async {
+            if let _dataAvailabilityService {
+                return _dataAvailabilityService
+            }
+            _dataAvailabilityService = await DataAvailabilityService(
+                config: config,
+                eventBus: eventBus,
+                scheduler: scheduler,
+                dataProvider: dataProvider,
+                dataStore: dataStore
+            )
+            _dataAvailabilityServiceRef = _dataAvailabilityService
+            await _dataAvailabilityService!.onSyncCompleted()
+            return _dataAvailabilityService!
+        }
     }
 
     public var blockchain: Blockchain {

Blockchain/Sources/Blockchain/RuntimeProtocols/RuntimeEvents.swift

@@ -236,6 +236,21 @@ public enum RuntimeEvents {
         }
     }
 
+    public struct WorkReportReceivedResponse: Event {
+        public let workReportHash: Data32
+        public let result: Result<Void, Error>
+
+        public init(workReportHash: Data32) {
+            self.workReportHash = workReportHash
+            result = .success(())
+        }
+
+        public init(workReportHash: Data32, error: Error) {
+            self.workReportHash = workReportHash
+            result = .failure(error)
+        }
+    }
+
     public struct WorkReportRequestReceived: Event {
         public let workReportHash: Data32
 
@@ -275,23 +290,6 @@ public enum RuntimeEvents {
         }
     }
 
-    //  Response to work report request
-    public struct WorkReportRequestResponse: Event {
-        public var workReportHash: Data32
-
-        public let result: Result<WorkReport, Error>
-
-        public init(workReportHash: Data32, workReport: WorkReport) {
-            self.workReportHash = workReportHash
-            result = .success(workReport)
-        }
-
-        public init(workReportHash: Data32, error: Error) {
-            self.workReportHash = workReportHash
-            result = .failure(error)
-        }
-    }
-
     public struct AuditShardRequestReceived: Event {
         public let erasureRoot: Data32
         public let shardIndex: UInt32

Blockchain/Sources/Blockchain/Types/GuaranteedWorkReport.swift

@@ -0,0 +1,42 @@
+import Codec
+import Foundation
+import Utils
+
+public struct GuaranteedWorkReport: Sendable, Equatable, Codable, Hashable {
+    public let workReport: WorkReport
+    public let slot: UInt32
+    public let signatures: [ValidatorSignature]
+
+    public init(
+        workReport: WorkReport,
+        slot: UInt32,
+        signatures: [ValidatorSignature]
+    ) {
+        self.workReport = workReport
+        self.slot = slot
+        self.signatures = signatures
+    }
+}
+
+extension GuaranteedWorkReport: Hashable32 {
+    public func hash() -> Data32 {
+        try! JamEncoder.encode(self).blake2b256hash()
+    }
+}
+
+extension GuaranteedWorkReport: Dummy {
+    public typealias Config = ProtocolConfigRef
+    public static func dummy(config: Config) -> GuaranteedWorkReport {
+        GuaranteedWorkReport(
+            workReport: WorkReport.dummy(config: config),
+            slot: 0,
+            signatures: []
+        )
+    }
+}
+
+extension GuaranteedWorkReport {
+    public func asRef() -> GuaranteedWorkReportRef {
+        GuaranteedWorkReportRef(self)
+    }
+}

Blockchain/Sources/Blockchain/Types/GuaranteedWorkReportRef.swift

@@ -0,0 +1,22 @@
+import Codec
+import Foundation
+import Utils
+
+public final class GuaranteedWorkReportRef: RefWithHash<GuaranteedWorkReport>, @unchecked Sendable {
+    public var workReport: WorkReport { value.workReport }
+    public var slot: UInt32 { value.slot }
+    public var signatures: [ValidatorSignature] { value.signatures }
+    override public var description: String {
+        "GuaranteedWorkReport(hash: \(workReport.hash()), timeslot: \(slot))"
+    }
+}
+
+extension GuaranteedWorkReportRef: Codable {
+    public convenience init(from decoder: Decoder) throws {
+        try self.init(.init(from: decoder))
+    }
+
+    public func encode(to encoder: Encoder) throws {
+        try value.encode(to: encoder)
+    }
+}

Blockchain/Sources/Blockchain/Validator/DataAvailabilityService.swift

@@ -16,9 +16,12 @@
     case invalidSegmentsRoot
     case invalidDataLength
     case pagedProofsGenerationError
+    case invalidWorkReportSlot
+    case invalidWorkReport
+    case insufficientSignatures
 }
 
-public final class DataAvailabilityService: ServiceBase2, @unchecked Sendable {
+public final class DataAvailabilityService: ServiceBase2, @unchecked Sendable, OnSyncCompleted {
     private let dataProvider: BlockchainDataProvider
     private let dataStore: DataStore
 
@@ -40,6 +43,16 @@
         }
     }
 
+    public func onSyncCompleted() async {
+        await subscribe(RuntimeEvents.WorkReportReceived.self, id: "DataAvailabilityService.WorkReportReceived") { [weak self] event in
+            await self?.handleWorkReportReceived(event)
+        }
+    }
+
+    public func handleWorkReportReceived(_ event: RuntimeEvents.WorkReportReceived) async {
+        await workReportDistribution(workReport: event.workReport, slot: event.slot, signatures: event.signatures)
+    }
+
     /// Purge old data from the data availability stores
     /// - Parameter epoch: The current epoch index
     public func purge(epoch _: EpochIndex) async {
@@ -180,6 +193,49 @@
         true
     }
 
+    // MARK: - Work-report Distribution (CE 135)
+
+    public func workReportDistribution(
+        workReport: WorkReport,
+        slot: UInt32,
+        signatures: [ValidatorSignature]
+    ) async {
+        let hash = workReport.hash()
+
+        do {
+            // verify slot
+            if await isSlotValid(slot) {
+                throw DataAvailabilityError.invalidWorkReportSlot
+            }
+            // verify signatures
+            try await validate(signatures: signatures)
+
+            // store guaranteedWorkReport
+            let report = GuaranteedWorkReport(
+                workReport: workReport,
+                slot: slot,
+                signatures: signatures
+            )
+            try await dataProvider.add(guaranteedWorkReport: GuaranteedWorkReportRef(report))
+            // response success result
+            publish(RuntimeEvents.WorkReportReceivedResponse(workReportHash: hash))
+        } catch {
+            publish(RuntimeEvents.WorkReportReceivedResponse(workReportHash: hash, error: error))
+        }
+    }
+
+    private func isSlotValid(_ slot: UInt32) async -> Bool {
+        let currentSlot = await dataProvider.bestHead.timeslot
+        return slot + 5 >= currentSlot && slot <= currentSlot + 3
+    }
+
+    private func validate(signatures: [ValidatorSignature]) async throws {
+        guard signatures.count >= 3 else {
+            throw DataAvailabilityError.insufficientSignatures
+        }
+        // TODO: more validates
+    }
+
     // MARK: - Shard Distribution (CE 137)
 
     /// Distribute shards to validators