diff --git a/Blockchain/Sources/Blockchain/DataStore/DataStore.swift b/Blockchain/Sources/Blockchain/DataStore/DataStore.swift new file mode 100644 index 00000000..ff1a3c1d --- /dev/null +++ b/Blockchain/Sources/Blockchain/DataStore/DataStore.swift @@ -0,0 +1,46 @@ +import Foundation + +public protocol DataStoreProtocol: Sendable { + func read(path: URL) async throws -> Data? + func write(path: URL, value: Data) async throws + func delete(path: URL) async throws +} + +public final class DataStore: Sendable { + private let impl: DataStoreProtocol + private let basePath: URL + + public init(_ impl: DataStoreProtocol, basePath: URL) { + self.impl = impl + self.basePath = basePath + } + + // partitioning files so that we won't have too many files in a single directory + private func getPath(path: String, name: String) -> URL { + var ret = basePath + ret.append(component: path) + var name = name[...] + if let first = name.first { + ret.append(component: String(first), directoryHint: .isDirectory) + name = name.dropFirst() + } + if let second = name.first { + ret.append(component: String(second), directoryHint: .isDirectory) + name = name.dropFirst() + } + ret.append(component: String(name), directoryHint: .notDirectory) + return ret + } + + public func read(path: String, name: String) async throws -> Data? { + try await impl.read(path: getPath(path: path, name: name)) + } + + public func write(path: String, name: String, value: Data) async throws { + try await impl.write(path: getPath(path: path, name: name), value: value) + } + + public func delete(path: String, name: String) async throws { + try await impl.delete(path: getPath(path: path, name: name)) + } +} diff --git a/Blockchain/Sources/Blockchain/DataStore/FilesystemDataStore.swift b/Blockchain/Sources/Blockchain/DataStore/FilesystemDataStore.swift new file mode 100644 index 00000000..04d49585 --- /dev/null +++ b/Blockchain/Sources/Blockchain/DataStore/FilesystemDataStore.swift @@ -0,0 +1,23 @@ +import Foundation + +public actor FilesystemDataStore: DataStoreProtocol { + private let fileManager: FileManager + + public init(fileManager: FileManager = .default) { + self.fileManager = fileManager + } + + public func read(path: URL) async throws -> Data? { + try Data(contentsOf: path) + } + + public func write(path: URL, value: Data) async throws { + let base = path.deletingLastPathComponent() + try fileManager.createDirectory(at: base, withIntermediateDirectories: true) + try value.write(to: path) + } + + public func delete(path: URL) async throws { + try fileManager.removeItem(at: path) + } +} diff --git a/Blockchain/Sources/Blockchain/DataStore/InMemoryDataStore.swift b/Blockchain/Sources/Blockchain/DataStore/InMemoryDataStore.swift new file mode 100644 index 00000000..9880c381 --- /dev/null +++ b/Blockchain/Sources/Blockchain/DataStore/InMemoryDataStore.swift @@ -0,0 +1,19 @@ +import Foundation + +public actor InMemoryDataStore: DataStoreProtocol { + private var store: [URL: Data] = [:] + + public init() {} + + public func read(path: URL) async throws -> Data? { + store[path] + } + + public func write(path: URL, value: Data) async throws { + store[path] = value + } + + public func delete(path: URL) async throws { + store[path] = nil + } +} diff --git a/Blockchain/Sources/Blockchain/Types/WorkPackageBundle.swift b/Blockchain/Sources/Blockchain/Types/WorkPackageBundle.swift new file mode 100644 index 00000000..4644898a --- /dev/null +++ b/Blockchain/Sources/Blockchain/Types/WorkPackageBundle.swift @@ -0,0 +1,10 @@ +import Foundation +import Utils + +// All the necessary data to audit a work package. Stored in audits DA +public struct WorkPackageBundle: Sendable, Equatable, Codable { + public var workPackage: WorkPackage + public var extrinsic: [Data] + public var importSegments: [[Data]] + public var justifications: [[Data]] +} diff --git a/Blockchain/Sources/Blockchain/Validator/DataAvailability.swift b/Blockchain/Sources/Blockchain/Validator/DataAvailability.swift new file mode 100644 index 00000000..8d731c37 --- /dev/null +++ b/Blockchain/Sources/Blockchain/Validator/DataAvailability.swift @@ -0,0 +1,54 @@ +import Foundation +import TracingUtils +import Utils + +enum DataAvailabilityStore: String, Sendable { + case imports + case audits +} + +public final class DataAvailability: ServiceBase2, @unchecked Sendable { + private let dataProvider: BlockchainDataProvider + private let dataStore: DataStore + + public init( + config: ProtocolConfigRef, + eventBus: EventBus, + scheduler: Scheduler, + dataProvider: BlockchainDataProvider, + dataStore: DataStore + ) async { + self.dataProvider = dataProvider + self.dataStore = dataStore + + super.init(id: "DataAvailability", config: config, eventBus: eventBus, scheduler: scheduler) + + scheduleForNextEpoch("BlockAuthor.scheduleForNextEpoch") { [weak self] epoch in + await self?.purge(epoch: epoch) + } + } + + public func purge(epoch _: EpochIndex) async { + // TODO: purge data + // GP 14.3.1 + // Guarantors are required to erasure-code and distribute two data sets: one blob, the auditable work-package containing + // the encoded work-package, extrinsic data and self-justifying imported segments which is placed in the short-term Audit + // da store and a second set of exported-segments data together with the Paged-Proofs metadata. Items in the first store + // are short-lived; assurers are expected to keep them only until finality of the block in which the availability of the work- + // result’s work-package is assured. Items in the second, meanwhile, are long-lived and expected to be kept for a minimum + // of 28 days (672 complete epochs) following the reporting of the work-report. + } + + public func fetchSegment(root _: Data32, index _: UInt16) async throws -> Data? { + // TODO: fetch segment + nil + } + + public func exportSegments(data _: [Data]) async throws { + // TODO: export segments + } + + public func distributeWorkpackageBundle(bundle _: WorkPackageBundle) async throws { + // TODO: distribute workpackage bundle to audits DA + } +} diff --git a/Blockchain/Sources/Blockchain/Validator/ValidatorService.swift b/Blockchain/Sources/Blockchain/Validator/ValidatorService.swift index 3ee6d2a6..cffde517 100644 --- a/Blockchain/Sources/Blockchain/Validator/ValidatorService.swift +++ b/Blockchain/Sources/Blockchain/Validator/ValidatorService.swift @@ -7,13 +7,15 @@ public final class ValidatorService: Sendable { private let safrole: SafroleService private let extrinsicPool: ExtrinsicPoolService private let blockAuthor: BlockAuthor + private let dataAvailability: DataAvailability public init( blockchain: Blockchain, keystore: KeyStore, eventBus: EventBus, scheduler: Scheduler, - dataProvider: BlockchainDataProvider + dataProvider: BlockchainDataProvider, + dataStore: DataStore ) async { self.blockchain = blockchain self.keystore = keystore @@ -38,6 +40,14 @@ public final class ValidatorService: Sendable { scheduler: scheduler, extrinsicPool: extrinsicPool ) + + dataAvailability = await DataAvailability( + config: blockchain.config, + eventBus: eventBus, + scheduler: scheduler, + dataProvider: dataProvider, + dataStore: dataStore + ) } public func onSyncCompleted() async { diff --git a/Blockchain/Tests/BlockchainTests/BlockchainServices.swift b/Blockchain/Tests/BlockchainTests/BlockchainServices.swift index d7985125..ae08472e 100644 --- a/Blockchain/Tests/BlockchainTests/BlockchainServices.swift +++ b/Blockchain/Tests/BlockchainTests/BlockchainServices.swift @@ -1,10 +1,12 @@ import Blockchain +import Foundation import Utils class BlockchainServices { let config: ProtocolConfigRef let timeProvider: MockTimeProvider let dataProvider: BlockchainDataProvider + let dataStore: DataStore let eventBus: EventBus let scheduler: MockScheduler let keystore: DevKeyStore @@ -30,6 +32,7 @@ class BlockchainServices { self.genesisBlock = genesisBlock self.genesisState = genesisState dataProvider = try! await BlockchainDataProvider(InMemoryDataProvider(genesisState: genesisState, genesisBlock: genesisBlock)) + dataStore = DataStore(InMemoryDataStore(), basePath: URL(fileURLWithPath: "/tmp/boka-test-data")) storeMiddleware = StoreMiddleware() eventBus = EventBus(eventMiddleware: .serial(Middleware(storeMiddleware), .noError), handlerMiddleware: .noError) diff --git a/Blockchain/Tests/BlockchainTests/ValidatorServiceTests.swift b/Blockchain/Tests/BlockchainTests/ValidatorServiceTests.swift index 8493157d..d927f77d 100644 --- a/Blockchain/Tests/BlockchainTests/ValidatorServiceTests.swift +++ b/Blockchain/Tests/BlockchainTests/ValidatorServiceTests.swift @@ -23,7 +23,8 @@ struct ValidatorServiceTests { keystore: services.keystore, eventBus: services.eventBus, scheduler: services.scheduler, - dataProvider: services.dataProvider + dataProvider: services.dataProvider, + dataStore: services.dataStore ) await validatorService.onSyncCompleted() return (services, validatorService) diff --git a/Boka/Sources/Boka.swift b/Boka/Sources/Boka.swift index f7b67ddd..d5442815 100644 --- a/Boka/Sources/Boka.swift +++ b/Boka/Sources/Boka.swift @@ -122,6 +122,12 @@ struct Boka: AsyncParsableCommand { return .rocksDB(path: path) } ?? .inMemory + let dataStore: DataStoreKind = basePath.map { + var path = URL(fileURLWithPath: $0) + path.append(path: "da") + return .filesystem(path: path) + } ?? .inMemory + logger.info("Peers: \(peers)") if validator { @@ -165,13 +171,14 @@ struct Boka: AsyncParsableCommand { handlerMiddleware: .tracing(prefix: "Handler") ) - let config = Node.Config( + let config = Config( rpc: rpcConfig, network: networkConfig, peers: peers, local: local, name: name, - database: database + database: database, + dataStore: dataStore ) let node: Node = if validator { diff --git a/Node/Sources/Node/Config.swift b/Node/Sources/Node/Config.swift new file mode 100644 index 00000000..a0989727 --- /dev/null +++ b/Node/Sources/Node/Config.swift @@ -0,0 +1,85 @@ +import Blockchain +import Database +import Foundation +import Networking +import RPC +import TracingUtils +import Utils + +public typealias RPCConfig = Server.Config +public typealias NetworkConfig = Network.Config + +private let logger = Logger(label: "config") + +public enum Database { + case inMemory + case rocksDB(path: URL) + + public func open(chainspec: ChainSpec) async throws -> BlockchainDataProvider { + switch self { + case let .rocksDB(path): + logger.info("Using RocksDB backend at \(path.absoluteString)") + let backend = try await RocksDBBackend( + path: path, + config: chainspec.getConfig(), + genesisBlock: chainspec.getBlock(), + genesisStateData: chainspec.getState() + ) + return try await BlockchainDataProvider(backend) + case .inMemory: + logger.info("Using in-memory backend") + let genesisBlock = try chainspec.getBlock() + let genesisStateData = try chainspec.getState() + let backend = try StateBackend(InMemoryBackend(), config: chainspec.getConfig(), rootHash: Data32()) + try await backend.writeRaw(Array(genesisStateData)) + let genesisState = try await State(backend: backend) + let genesisStateRef = StateRef(genesisState) + return try await BlockchainDataProvider(InMemoryDataProvider(genesisState: genesisStateRef, genesisBlock: genesisBlock)) + } + } +} + +public enum DataStoreKind { + case inMemory + case filesystem(path: URL) + + func create() -> DataStore { + switch self { + case let .filesystem(path): + logger.info("Using filesystem data store at \(path.absoluteString)") + let dataStore = FilesystemDataStore() + return DataStore(dataStore, basePath: path) + case .inMemory: + logger.info("Using in-memory data store") + return DataStore(InMemoryDataStore(), basePath: URL(filePath: "/tmp/boka")) + } + } +} + +public struct Config { + public var rpc: RPCConfig? + public var network: NetworkConfig + public var peers: [NetAddr] + public var local: Bool + public var name: String? + public var database: Database + public var dataStore: DataStoreKind + + public init( + rpc: RPCConfig?, + network: NetworkConfig, + peers: [NetAddr] = [], + local: Bool = false, + name: String? = nil, + database: Database = .inMemory, + dataStore: DataStoreKind = .inMemory + ) { + self.rpc = rpc + self.network = network + self.peers = peers + self.local = local + self.name = name + self.database = database + self.dataStore = dataStore + } +} diff --git a/Node/Sources/Node/Node.swift b/Node/Sources/Node/Node.swift index 46d8276f..5d998032 100644 --- a/Node/Sources/Node/Node.swift +++ b/Node/Sources/Node/Node.swift @@ -8,63 +8,7 @@ import Utils private let logger = Logger(label: "node") -public typealias RPCConfig = Server.Config -public typealias NetworkConfig = Network.Config - -public enum Database { - case inMemory - case rocksDB(path: URL) - - public func open(chainspec: ChainSpec) async throws -> BlockchainDataProvider { - switch self { - case let .rocksDB(path): - logger.info("Using RocksDB backend at \(path.absoluteString)") - let backend = try await RocksDBBackend( - path: path, - config: chainspec.getConfig(), - genesisBlock: chainspec.getBlock(), - genesisStateData: chainspec.getState() - ) - return try await BlockchainDataProvider(backend) - case .inMemory: - logger.info("Using in-memory backend") - let genesisBlock = try chainspec.getBlock() - let genesisStateData = try chainspec.getState() - let backend = try StateBackend(InMemoryBackend(), config: chainspec.getConfig(), rootHash: Data32()) - try await backend.writeRaw(Array(genesisStateData)) - let genesisState = try await State(backend: backend) - let genesisStateRef = StateRef(genesisState) - return try await BlockchainDataProvider(InMemoryDataProvider(genesisState: genesisStateRef, genesisBlock: genesisBlock)) - } - } -} - public class Node { - public struct Config { - public var rpc: RPCConfig? - public var network: NetworkConfig - public var peers: [NetAddr] - public var local: Bool - public var name: String? - public var database: Database - - public init( - rpc: RPCConfig?, - network: NetworkConfig, - peers: [NetAddr] = [], - local: Bool = false, - name: String? = nil, - database: Database = .inMemory - ) { - self.rpc = rpc - self.network = network - self.peers = peers - self.local = local - self.name = name - self.database = database - } - } - public let config: Config public let blockchain: Blockchain public let rpcServer: Server? diff --git a/Node/Sources/Node/ValidatorNode.swift b/Node/Sources/Node/ValidatorNode.swift index 5d55cdce..b91a850d 100644 --- a/Node/Sources/Node/ValidatorNode.swift +++ b/Node/Sources/Node/ValidatorNode.swift @@ -28,12 +28,13 @@ public class ValidatorNode: Node { keystore: keystore, eventBus: eventBus, scheduler: scheduler, - dataProvider: dataProvider + dataProvider: dataProvider, + dataStore: config.dataStore.create() ) self.validator = validator let syncManager = network.syncManager - let dataProvider: BlockchainDataProvider = blockchain.dataProvider + let dataProvider = dataProvider let local = config.local Task { if !local { diff --git a/Node/Tests/NodeTests/Topology.swift b/Node/Tests/NodeTests/Topology.swift index 785814d4..be8654c2 100644 --- a/Node/Tests/NodeTests/Topology.swift +++ b/Node/Tests/NodeTests/Topology.swift @@ -35,7 +35,7 @@ struct Topology { let eventBus = EventBus(eventMiddleware: .serial(Middleware(storeMiddleware), .noError), handlerMiddleware: .noError) let keystore = try await DevKeyStore(devKeysCount: 0) let keys = try await keystore.addDevKeys(seed: desc.devSeed) - let nodeConfig = await Node.Config( + let nodeConfig = await Config( rpc: nil, network: Network.Config( role: desc.isValidator ? .validator : .builder,