diff --git a/storage/blockexchange/engine/activedownload.nim b/storage/blockexchange/engine/activedownload.nim index fc8ba828..a260189b 100644 --- a/storage/blockexchange/engine/activedownload.nim +++ b/storage/blockexchange/engine/activedownload.nim @@ -94,7 +94,7 @@ proc manifestCid*(download: ActiveDownload): Cid = download.ctx.md.manifestCid proc makeBlockAddress*(download: ActiveDownload, index: uint64): BlockAddress = - BlockAddress(treeCid: download.treeCid, index: index.int) + BlockAddress(treeCid: download.treeCid, index: index) proc getOrCreateBlockReq(download: ActiveDownload, address: BlockAddress): BlockReq = download.blocks.withValue(address, blkReq): diff --git a/storage/blockexchange/engine/downloadcontext.nim b/storage/blockexchange/engine/downloadcontext.nim index 346b530d..d5f9a410 100644 --- a/storage/blockexchange/engine/downloadcontext.nim +++ b/storage/blockexchange/engine/downloadcontext.nim @@ -20,6 +20,7 @@ import ../../manifest import ../../storagetypes import ../../blocktype import ../protocol/constants +import ../protocol/message import ../utils export scheduler, peercontext, manifest @@ -67,7 +68,7 @@ type lastBroadcastTime: Moment broadcastInterval: Duration of spRandomWindow: - pendingRanges: seq[tuple[start: uint64, count: uint64]] + pendingRanges: seq[IndexRange] DownloadContext* = ref object md*: ManifestDescriptor @@ -112,16 +113,16 @@ proc shouldBroadcast(t: BroadcastAvailabilityTracker, scheduler: Scheduler): boo proc getRanges( t: var BroadcastAvailabilityTracker, scheduler: Scheduler -): seq[tuple[start: uint64, count: uint64]] = +): seq[IndexRange] = case t.policy of spRandomWindow: t.pendingRanges of spSequential: let watermark = scheduler.completedWatermark() - var ranges: seq[tuple[start: uint64, count: uint64]] = @[] + var ranges: seq[IndexRange] = @[] if watermark > t.lastBroadcastedWatermark: ranges.add( - ( + IndexRange( start: t.lastBroadcastedWatermark, count: watermark - t.lastBroadcastedWatermark, ) @@ -129,7 +130,7 @@ proc getRanges( t.pendingOOOSnapshot.clear() for batchStart in scheduler.completedOutOfOrderItems: if batchStart notin t.broadcastedOutOfOrder: - ranges.add((start: batchStart, count: scheduler.batchSizeCount)) + ranges.add(IndexRange(start: batchStart, count: scheduler.batchSizeCount)) t.pendingOOOSnapshot.incl(batchStart) ranges @@ -156,7 +157,7 @@ proc addPendingRange( ) = case t.policy of spRandomWindow: - t.pendingRanges.add(range) + t.pendingRanges.add(IndexRange(start: range.start, count: range.count)) of spSequential: discard @@ -251,20 +252,18 @@ proc trimPresenceBeforeWatermark*(ctx: DownloadContext) = let peer = peerOpt.get() # only trim range-based availability if peer.availability.kind == bakRanges: - var newRanges: seq[tuple[start: uint64, count: uint64]] = @[] - for (start, count) in peer.availability.ranges: - let rangeEnd = start + count + var newRanges: seq[IndexRange] = @[] + for r in peer.availability.ranges: + let rangeEnd = r.start + r.count if rangeEnd > watermark: # keep ranges not entirely below watermark - newRanges.add((start, count)) + newRanges.add(r) peer.availability = BlockAvailability.fromRanges(newRanges) proc shouldBroadcastAvailability*(ctx: DownloadContext): bool = ctx.availabilityTracker.shouldBroadcast(ctx.scheduler) -proc getAvailabilityBroadcast*( - ctx: DownloadContext -): seq[tuple[start: uint64, count: uint64]] = +proc getAvailabilityBroadcast*(ctx: DownloadContext): seq[IndexRange] = ctx.availabilityTracker.getRanges(ctx.scheduler) proc markAvailabilityBroadcasted*(ctx: DownloadContext) = diff --git a/storage/blockexchange/engine/engine.nim b/storage/blockexchange/engine/engine.nim index 4b48e420..5d34db37 100644 --- a/storage/blockexchange/engine/engine.nim +++ b/storage/blockexchange/engine/engine.nim @@ -188,7 +188,7 @@ proc validateBlockDeliveryView(self: BlockExcEngine, view: BlockDeliveryView): ? without proof =? view.proof: return failure("Missing proof") - if proof.index != view.address.index: + if proof.index.uint64 != view.address.index: return failure( "Proof index " & $proof.index & " doesn't match leaf index " & $view.address.index ) @@ -230,7 +230,7 @@ proc sendWantBlocksRequest( let treeCid = download.treeCid # missingIndices must be sorted ascending with no duplicates for correct coalescing - var ranges: seq[tuple[start: uint64, count: uint64]] = @[] + var ranges: seq[IndexRange] = @[] if missingIndices.len > 0: var rangeStart = missingIndices[0] @@ -240,11 +240,11 @@ proc sendWantBlocksRequest( if missingIndices[i] == rangeStart + rangeCount: rangeCount += 1 else: - ranges.add((rangeStart, rangeCount)) + ranges.add(IndexRange(start: rangeStart, count: rangeCount)) rangeStart = missingIndices[i] rangeCount = 1 - ranges.add((rangeStart, rangeCount)) + ranges.add(IndexRange(start: rangeStart, count: rangeCount)) trace "Requesting missing blocks", treeCid = treeCid, @@ -444,7 +444,7 @@ proc broadcastWantHave( count: uint64, peers: seq[PeerContext], ) {.async: (raises: [CancelledError]).} = - let rangeAddress = BlockAddress.init(download.treeCid, start.int) + let rangeAddress = BlockAddress.init(download.treeCid, start) for peerCtx in peers: if not download.addPeerIfAbsent(peerCtx.id, BlockAvailability.unknown()): # skip presence request for peer with Complete availability @@ -537,7 +537,7 @@ proc downloadWorker( swarmPeers = ctx.swarm.peerCount() let presence = BlockPresence( - address: BlockAddress(treeCid: treeCid, index: broadcastRanges[0].start.int), + address: BlockAddress(treeCid: treeCid, index: broadcastRanges[0].start), kind: BlockPresenceType.HaveRange, ranges: broadcastRanges, ) @@ -811,7 +811,7 @@ proc startTreeDownloadGeneric[T: Block | void]( proc genNext(): Future[?!T] {.async: (raises: [CancelledError]).} = while pendingHandle.isNone and nextBlockToRequest < totalBlocks: - let address = BlockAddress(treeCid: treeCid, index: nextBlockToRequest.int) + let address = BlockAddress(treeCid: treeCid, index: nextBlockToRequest) nextBlockToRequest += 1 let handle = @@ -975,7 +975,7 @@ proc wantListHandler*( let runtimeQuota = 100.milliseconds var - ranges: seq[tuple[start: uint64, count: uint64]] = @[] + ranges: seq[IndexRange] = @[] rangeStart: uint64 = 0 inRange = false lastIdle = Moment.now() @@ -985,7 +985,7 @@ proc wantListHandler*( await idleAsync() lastIdle = Moment.now() - let address = BlockAddress(treeCid: treeCid, index: (startIdx + i).int) + let address = BlockAddress(treeCid: treeCid, index: startIdx + i) let have = try: await address in self.localStore @@ -998,11 +998,17 @@ proc wantListHandler*( inRange = true else: if inRange: - ranges.add((rangeStart, (startIdx + i) - rangeStart)) + ranges.add( + IndexRange(start: rangeStart, count: (startIdx + i) - rangeStart) + ) inRange = false if inRange: - ranges.add((rangeStart, (startIdx + effectiveCount) - rangeStart)) + ranges.add( + IndexRange( + start: rangeStart, count: (startIdx + effectiveCount) - rangeStart + ) + ) iterBudget -= effectiveCount @@ -1039,7 +1045,7 @@ proc wantListHandler*( BlockPresence( address: e.address, kind: BlockPresenceType.HaveRange, - ranges: @[(e.address.index.uint64, 1'u64)], + ranges: @[IndexRange(start: e.address.index, count: 1'u64)], downloadId: e.downloadId, ) ) @@ -1175,10 +1181,10 @@ proc new*( notFoundCount = 0 totalRequested: uint64 = 0 - for (start, count) in req.ranges: - totalRequested += count - for i in start ..< start + count: - let address = BlockAddress(treeCid: req.treeCid, index: i.Natural) + for r in req.ranges: + totalRequested += r.count + for i in r.start ..< r.start + r.count: + let address = BlockAddress(treeCid: req.treeCid, index: i) let res = await self.localLookup(address) if res.isOk: diff --git a/storage/blockexchange/network/networkpeer.nim b/storage/blockexchange/network/networkpeer.nim index d8042743..467399e2 100644 --- a/storage/blockexchange/network/networkpeer.nim +++ b/storage/blockexchange/network/networkpeer.nim @@ -97,7 +97,7 @@ proc readLoop*(self: NetworkPeer, conn: Connection) {.async: (raises: []).} = if dataLen > 0: await conn.readExactly(addr data[0], dataLen) - let msg = Message.protobufDecode(data).mapFailure().tryGet() + let msg = Message.decode(data).mapFailure().tryGet() await self.handler(self, msg) of mtWantBlocksRequest: let reqResult = await readWantBlocksRequest(conn, dataLen) @@ -164,7 +164,7 @@ proc send*( return try: - let msgData = protobufEncode(msg) + let msgData = msg.encode() let frameLen = 1 + msgData.len diff --git a/storage/blockexchange/protocol/message.nim b/storage/blockexchange/protocol/message.nim index ffd2a2b6..0d673be2 100644 --- a/storage/blockexchange/protocol/message.nim +++ b/storage/blockexchange/protocol/message.nim @@ -1,193 +1,100 @@ # Protocol of data exchange between Logos Storage nodes # and Protobuf encoder/decoder for these messages. -# -# Eventually all this code should be auto-generated from message.proto. -import std/sugar -import pkg/libp2p/protobuf/minprotobuf -import pkg/libp2p/cid +{.push raises: [].} -import pkg/questionable +import pkg/faststreams +import pkg/protobuf_serialization +import pkg/protobuf_serialization/codec +import pkg/protobuf_serialization/std/enums +import pkg/protobuf_serialization/pkg/results import ../../merkletree import ../../blocktype +import ../../utils/protobuf/cid +import ../../utils/protobuf/serializer import ./constants type - WantType* = enum + WantType* {.pure.} = enum WantHave = 0 # Presence query - the only type used with batch transfer protocol - WantListEntry* = object - address*: BlockAddress - priority*: int32 # The priority (normalized). default to 1 - cancel*: bool # Whether this revokes an entry - wantType*: WantType # Defaults to WantHave (only type supported) - sendDontHave*: bool # Note: defaults to false - rangeCount*: uint64 + WantListEntry* {.proto2.} = object + address* {.fieldNumber: 1, required.}: BlockAddress + priority* {.fieldNumber: 2, required, pint.}: int32 + # The priority (normalized). default to 1 + cancel* {.fieldNumber: 3, required.}: bool # Whether this revokes an entry + wantType* {.fieldNumber: 4, required, ext.}: WantType + # Defaults to WantHave (only type supported) + sendDontHave* {.fieldNumber: 5, required.}: bool # Note: defaults to false + rangeCount* {.fieldNumber: 6, required, pint.}: uint64 # For range queries: number of sequential blocks starting from address.index (0 = single block) - downloadId*: uint64 # Unique download ID for request/response correlation + downloadId* {.fieldNumber: 7, required, pint.}: uint64 + # Unique download ID for request/response correlation - WantList* = object - entries*: seq[WantListEntry] # A list of wantList entries - full*: bool # Whether this is the full wantList. default to false + WantList* {.proto2.} = object + entries* {.fieldNumber: 1.}: seq[WantListEntry] # A list of wantList entries + full* {.fieldNumber: 2, required.}: bool + # Whether this is the full wantList. default to false BlockDelivery* = object blk*: Block address*: BlockAddress proof*: ?StorageMerkleProof - BlockPresenceType* = enum + BlockPresenceType* {.pure.} = enum DontHave = 0 HaveRange = 1 Complete = 2 - BlockPresence* = object - address*: BlockAddress - kind*: BlockPresenceType - ranges*: seq[tuple[start: uint64, count: uint64]] - downloadId*: uint64 # echoed for request/response correlation + IndexRange* {.proto2.} = object + start* {.fieldNumber: 1, required, pint.}: uint64 + count* {.fieldNumber: 2, required, pint.}: uint64 - Message* = object - wantList*: WantList - blockPresences*: seq[BlockPresence] + BlockPresence* {.proto2.} = object + address* {.fieldNumber: 1, required.}: BlockAddress + kind* {.fieldNumber: 2, required, ext.}: BlockPresenceType + ranges* {.fieldNumber: 3.}: seq[IndexRange] + downloadId* {.fieldNumber: 4, required, pint.}: uint64 + # echoed for request/response correlation -# -# Encoding Message into seq[byte] in Protobuf format -# + Message* {.proto2.} = object + wantList* {.fieldNumber: 1, required.}: WantList + blockPresences* {.fieldNumber: 4.}: seq[BlockPresence] -proc write*(pb: var ProtoBuffer, field: int, value: BlockAddress) = - var ipb = initProtoBuffer() - ipb.write(1, value.treeCid.data.buffer) - ipb.write(2, value.index.uint64) - ipb.finish() - pb.write(field, ipb) +proc readFieldInto*( + stream: InputStream, + value: var seq[WantListEntry], + header: FieldHeader, + ProtoType: type SomeProto, +): bool {.raises: [SerializationError, IOError].} = + if value.len >= MaxWantListEntries: + raise newException( + SerializationError, "WantList exceeds " & $MaxWantListEntries & " entries" + ) + var val = default(WantListEntry) + if stream.readFieldInto(val, header, ProtoType): + value.add move(val) + true + else: + false -proc write*(pb: var ProtoBuffer, field: int, value: WantListEntry) = - var ipb = initProtoBuffer() - ipb.write(1, value.address) - ipb.write(2, value.priority.uint64) - ipb.write(3, value.cancel.uint) - ipb.write(4, value.wantType.uint) - ipb.write(5, value.sendDontHave.uint) - ipb.write(6, value.rangeCount) - ipb.write(7, value.downloadId) - ipb.finish() - pb.write(field, ipb) +proc readFieldInto*( + stream: InputStream, + value: var seq[BlockPresence], + header: FieldHeader, + ProtoType: type SomeProto, +): bool {.raises: [SerializationError, IOError].} = + if value.len >= MaxBlockPresenceEntries: + raise newException( + SerializationError, + "blockPresences exceeds " & $MaxBlockPresenceEntries & " entries", + ) + var val = default(BlockPresence) + if stream.readFieldInto(val, header, ProtoType): + value.add move(val) + true + else: + false -proc write*(pb: var ProtoBuffer, field: int, value: WantList) = - var ipb = initProtoBuffer() - for v in value.entries: - ipb.write(1, v) - ipb.write(2, value.full.uint) - ipb.finish() - pb.write(field, ipb) - -proc write*(pb: var ProtoBuffer, field: int, value: BlockPresence) = - var ipb = initProtoBuffer() - ipb.write(1, value.address) - ipb.write(2, value.kind.uint) - # Encode ranges if present - for (start, count) in value.ranges: - var rangePb = initProtoBuffer() - rangePb.write(1, start) - rangePb.write(2, count) - rangePb.finish() - ipb.write(3, rangePb) - ipb.write(4, value.downloadId) - ipb.finish() - pb.write(field, ipb) - -proc protobufEncode*(value: Message): seq[byte] = - var ipb = initProtoBuffer() - ipb.write(1, value.wantList) - for v in value.blockPresences: - ipb.write(4, v) - ipb.finish() - ipb.buffer - -# -# Decoding Message from seq[byte] in Protobuf format -# -proc decode*(_: type BlockAddress, pb: ProtoBuffer): ProtoResult[BlockAddress] = - var - value: BlockAddress - field: uint64 - cidBuf = newSeq[byte]() - - if ?pb.getField(1, cidBuf): - value.treeCid = ?Cid.init(cidBuf).mapErr(x => ProtoError.IncorrectBlob) - if ?pb.getField(2, field): - value.index = field - - ok(value) - -proc decode*(_: type WantListEntry, pb: ProtoBuffer): ProtoResult[WantListEntry] = - var - value = WantListEntry() - field: uint64 - ipb: ProtoBuffer - if ?pb.getField(1, ipb): - value.address = ?BlockAddress.decode(ipb) - if ?pb.getField(2, field): - value.priority = int32(field) - if ?pb.getField(3, field): - value.cancel = bool(field) - if ?pb.getField(4, field): - value.wantType = WantType(field) - if ?pb.getField(5, field): - value.sendDontHave = bool(field) - if ?pb.getField(6, field): - value.rangeCount = field - if ?pb.getField(7, field): - value.downloadId = field - ok(value) - -proc decode*(_: type WantList, pb: ProtoBuffer): ProtoResult[WantList] = - var - value = WantList() - field: uint64 - sublist: seq[seq[byte]] - if ?pb.getRepeatedField(1, sublist): - if sublist.len > MaxWantListEntries: - return err(ProtoError.BufferOverflow) - for item in sublist: - value.entries.add(?WantListEntry.decode(initProtoBuffer(item))) - if ?pb.getField(2, field): - value.full = bool(field) - ok(value) - -proc decode*(_: type BlockPresence, pb: ProtoBuffer): ProtoResult[BlockPresence] = - var - value = BlockPresence() - field: uint64 - ipb: ProtoBuffer - rangelist: seq[seq[byte]] - if ?pb.getField(1, ipb): - value.address = ?BlockAddress.decode(ipb) - if ?pb.getField(2, field): - value.kind = BlockPresenceType(field) - if ?pb.getRepeatedField(3, rangelist): - for item in rangelist: - var rangePb = initProtoBuffer(item) - var start, count: uint64 - discard ?rangePb.getField(1, start) - discard ?rangePb.getField(2, count) - value.ranges.add((start, count)) - if ?pb.getField(4, field): - value.downloadId = field - ok(value) - -proc protobufDecode*(_: type Message, msg: seq[byte]): ProtoResult[Message] = - var - value = Message() - pb = initProtoBuffer(msg) - ipb: ProtoBuffer - sublist: seq[seq[byte]] - if ?pb.getField(1, ipb): - value.wantList = ?WantList.decode(ipb) - if ?pb.getRepeatedField(4, sublist): - if sublist.len > MaxBlockPresenceEntries: - return err(ProtoError.BufferOverflow) - for item in sublist: - value.blockPresences.add(?BlockPresence.decode(initProtoBuffer(item))) - ok(value) +Protobuf.serializerForResult([Message]) diff --git a/storage/blockexchange/protocol/presence.nim b/storage/blockexchange/protocol/presence.nim index f0c9c7a3..9228144e 100644 --- a/storage/blockexchange/protocol/presence.nim +++ b/storage/blockexchange/protocol/presence.nim @@ -15,7 +15,7 @@ type address*: BlockAddress have*: bool presenceType*: BlockPresenceType - ranges*: seq[tuple[start: uint64, count: uint64]] + ranges*: seq[IndexRange] func init*(_: type Presence, message: PresenceMessage): ?Presence = some Presence( diff --git a/storage/blockexchange/protocol/wantblocks.nim b/storage/blockexchange/protocol/wantblocks.nim index 246bf6c9..12e94883 100644 --- a/storage/blockexchange/protocol/wantblocks.nim +++ b/storage/blockexchange/protocol/wantblocks.nim @@ -55,7 +55,7 @@ type WantBlocksRequest* = object requestId*: uint64 treeCid*: Cid - ranges*: seq[tuple[start: uint64, count: uint64]] + ranges*: seq[IndexRange] SharedBlocksBuffer* = ref object data*: seq[byte] @@ -173,11 +173,11 @@ proc encodeRequestInto*( copyMem(addr buf[offset], unsafeAddr rangeCountLE, 4) offset += 4 - for (start, count) in req.ranges: - let startLE = start.toLE + for r in req.ranges: + let startLE = r.start.toLE copyMem(addr buf[offset], unsafeAddr startLE, 8) offset += 8 - let countLE = count.toLE + let countLE = r.count.toLE copyMem(addr buf[offset], unsafeAddr countLE, 8) offset += 8 @@ -214,13 +214,13 @@ proc decodeRequest*(data: openArray[byte]): WantBlocksResult[WantBlocksRequest] if offset + (rangeCount * SizeRange) > data.len: return err(wantBlocksError(RequestTruncated, "Request truncated (ranges)")) - var ranges = newSeqOfCap[tuple[start: uint64, count: uint64]](rangeCount) + var ranges = newSeqOfCap[IndexRange](rangeCount) for _ in 0 ..< rangeCount: let start = uint64.fromBytes(data.toOpenArray(offset, offset + 7), littleEndian) offset += 8 let count = uint64.fromBytes(data.toOpenArray(offset, offset + 7), littleEndian) offset += 8 - ranges.add((start, count)) + ranges.add(IndexRange(start: start, count: count)) ok(WantBlocksRequest(requestId: requestId, treeCid: treeCid, ranges: ranges)) @@ -615,7 +615,7 @@ proc toBlockDeliveryView*( ok( BlockDeliveryView( cid: entry.cid, - address: BlockAddress(treeCid: treeCid, index: entry.index.Natural), + address: BlockAddress(treeCid: treeCid, index: entry.index), proof: some(entry.proof), sharedBuf: sharedBuf, dataOffset: entry.dataOffset, diff --git a/storage/blockexchange/types.nim b/storage/blockexchange/types.nim index fe48771f..1daaee8c 100644 --- a/storage/blockexchange/types.nim +++ b/storage/blockexchange/types.nim @@ -13,10 +13,12 @@ import std/algorithm import pkg/libp2p/cid +import ./protocol/message + type BlockRange* = object treeCid*: Cid - ranges*: seq[tuple[start: uint64, count: uint64]] + ranges*: seq[IndexRange] BlockAvailabilityKind* = enum bakUnknown @@ -31,7 +33,7 @@ type of bakComplete: discard of bakRanges: - ranges*: seq[tuple[start: uint64, count: uint64]] + ranges*: seq[IndexRange] of bakBitmap: bitmap*: seq[byte] totalBlocks*: uint64 @@ -43,7 +45,7 @@ proc complete*(_: type BlockAvailability): BlockAvailability = BlockAvailability(kind: bakComplete) proc fromRanges*( - _: type BlockAvailability, ranges: seq[tuple[start: uint64, count: uint64]] + _: type BlockAvailability, ranges: seq[IndexRange] ): BlockAvailability = BlockAvailability(kind: bakRanges, ranges: ranges) @@ -59,10 +61,10 @@ proc hasBlock*(avail: BlockAvailability, index: uint64): bool = of bakComplete: true of bakRanges: - for (start, count) in avail.ranges: - if count > high(uint64) - start: + for r in avail.ranges: + if r.count > high(uint64) - r.start: continue - if index >= start and index < start + count: + if index >= r.start and index < r.start + r.count: return true false of bakBitmap: @@ -86,11 +88,11 @@ proc hasRange*(avail: BlockAvailability, start: uint64, count: uint64): bool = true of bakRanges: let reqEnd = start + count - for (rangeStart, rangeCount) in avail.ranges: - if rangeCount > high(uint64) - rangeStart: + for r in avail.ranges: + if r.count > high(uint64) - r.start: continue - let rangeEnd = rangeStart + rangeCount - if start >= rangeStart and reqEnd <= rangeEnd: + let rangeEnd = r.start + r.count + if start >= r.start and reqEnd <= rangeEnd: return true false of bakBitmap: @@ -110,12 +112,12 @@ proc hasAnyInRange*(avail: BlockAvailability, start: uint64, count: uint64): boo true of bakRanges: let reqEnd = start + count - for (rangeStart, rangeCount) in avail.ranges: - if rangeCount > high(uint64) - rangeStart: + for r in avail.ranges: + if r.count > high(uint64) - r.start: continue - let rangeEnd = rangeStart + rangeCount + let rangeEnd = r.start + r.count # check if they overlap - if start < rangeEnd and rangeStart < reqEnd: + if start < rangeEnd and r.start < reqEnd: return true false of bakBitmap: @@ -124,15 +126,13 @@ proc hasAnyInRange*(avail: BlockAvailability, start: uint64, count: uint64): boo return true false -proc mergeRanges( - ranges: seq[tuple[start: uint64, count: uint64]] -): seq[tuple[start: uint64, count: uint64]] = +proc mergeRanges(ranges: seq[IndexRange]): seq[IndexRange] = if ranges.len == 0: return @[] var sorted = ranges sorted.sort( - proc(a, b: tuple[start: uint64, count: uint64]): int = + proc(a, b: IndexRange): int = if a.start < b.start: -1 elif a.start > b.start: @@ -172,9 +172,7 @@ proc merge*(current: BlockAvailability, other: BlockAvailability): BlockAvailabi if other.kind == bakUnknown: return current - proc bitmapToRanges( - avail: BlockAvailability - ): seq[tuple[start: uint64, count: uint64]] = + proc bitmapToRanges(avail: BlockAvailability): seq[IndexRange] = result = @[] var inRange = false @@ -185,10 +183,10 @@ proc merge*(current: BlockAvailability, other: BlockAvailability): BlockAvailabi rangeStart = i inRange = true elif not hasIt and inRange: - result.add((rangeStart, i - rangeStart)) + result.add(IndexRange(start: rangeStart, count: i - rangeStart)) inRange = false if inRange: - result.add((rangeStart, avail.totalBlocks - rangeStart)) + result.add(IndexRange(start: rangeStart, count: avail.totalBlocks - rangeStart)) let currentRanges = if current.kind == bakRanges: diff --git a/storage/blockexchange/utils.nim b/storage/blockexchange/utils.nim index 5cba3283..db2e51e3 100644 --- a/storage/blockexchange/utils.nim +++ b/storage/blockexchange/utils.nim @@ -10,11 +10,12 @@ import std/algorithm import ./protocol/constants +import ./protocol/message func isIndexInRanges*( - index: uint64, ranges: openArray[(uint64, uint64)], sortedRanges: bool = false + index: uint64, ranges: openArray[IndexRange], sortedRanges: bool = false ): bool = - func binarySearch(r: openArray[(uint64, uint64)]): bool = + func binarySearch(r: openArray[IndexRange]): bool = var lo = 0 hi = r.len - 1 @@ -22,15 +23,14 @@ func isIndexInRanges*( while lo <= hi: let mid = (lo + hi) div 2 - if r[mid][0] <= index: + if r[mid].start <= index: candidate = mid lo = mid + 1 else: hi = mid - 1 if candidate >= 0: - let (start, count) = r[candidate] - return index < start + count + return index < r[candidate].start + r[candidate].count return false @@ -41,8 +41,8 @@ func isIndexInRanges*( binarySearch(ranges) else: let sorted = @ranges.sorted( - proc(a, b: (uint64, uint64)): int = - cmp(a[0], b[0]) + proc(a, b: IndexRange): int = + cmp(a.start, b.start) ) binarySearch(sorted) diff --git a/storage/blocktype.nim b/storage/blocktype.nim index 3cc6958a..92797e05 100644 --- a/storage/blocktype.nim +++ b/storage/blocktype.nim @@ -12,6 +12,7 @@ import std/[tables, sugar, hashes] {.push raises: [], gcsafe.} import pkg/libp2p/[cid, multicodec, multihash] +import pkg/protobuf_serialization import pkg/stew/[byteutils, endians2] import pkg/questionable import pkg/questionable/results @@ -30,9 +31,9 @@ type cid*: Cid data*: ref seq[byte] - BlockAddress* = object - treeCid* {.serialize.}: Cid - index* {.serialize.}: Natural + BlockAddress* {.proto2.} = object + treeCid* {.serialize, fieldNumber: 1, required, ext.}: Cid + index* {.serialize, fieldNumber: 2, required, pint.}: uint64 logutils.formatIt(LogFormat.textLines, BlockAddress): "treeCid: " & shortLog($it.treeCid) & ", index: " & $it.index @@ -50,7 +51,7 @@ proc hash*(a: BlockAddress): Hash = let data = a.treeCid.data.buffer & @(a.index.uint64.toBytesBE) hash(data) -proc init*(_: type BlockAddress, treeCid: Cid, index: Natural): BlockAddress = +proc init*(_: type BlockAddress, treeCid: Cid, index: uint64): BlockAddress = BlockAddress(treeCid: treeCid, index: index) proc `$`*(b: Block): string = diff --git a/storage/dht_proxy/protocol.nim b/storage/dht_proxy/protocol.nim index 90327c73..dd76b694 100644 --- a/storage/dht_proxy/protocol.nim +++ b/storage/dht_proxy/protocol.nim @@ -10,12 +10,13 @@ {.push raises: [].} import pkg/chronos -import pkg/libp2p/protobuf/minprotobuf import pkg/libp2p_mix -import pkg/libp2p/routing_record -import pkg/stew/objects +import pkg/protobuf_serialization +import pkg/protobuf_serialization/std/enums +import pkg/protobuf_serialization/pkg/results import ../logutils +import ../utils/protobuf/serializer const DhtProxyCodec* = "/storage/dht-proxy/1.0.0" @@ -43,56 +44,15 @@ type ErrResponseTooLarge = 201 ErrTooBusy = 202 - LookupRequest* = object - queryType*: QueryType - queryBytes*: seq[byte] + LookupRequest* {.proto2.} = object + queryType* {.fieldNumber: 1, required, ext.}: QueryType + queryBytes* {.fieldNumber: 2, required.}: seq[byte] - LookupResponse* = object - code*: LookupCode - providers*: seq[seq[byte]] + LookupResponse* {.proto2.} = object + code* {.fieldNumber: 1, required, ext.}: LookupCode + providers* {.fieldNumber: 2.}: seq[seq[byte]] -proc encode*(req: LookupRequest): seq[byte] = - var pb = initProtoBuffer() - pb.write(1, req.queryType.uint32) - pb.write(2, req.queryBytes) - pb.finish() - pb.buffer - -proc encode*(resp: LookupResponse): seq[byte] = - var pb = initProtoBuffer() - pb.write(1, resp.code.uint32) - for spr in resp.providers: - pb.write(2, spr) - pb.finish() - pb.buffer - -proc decode*(_: type LookupRequest, data: openArray[byte]): ProtoResult[LookupRequest] = - let pb = initProtoBuffer(data) - var - req = LookupRequest() - qt: uint32 - - if ?pb.getField(1, qt): - if not checkedEnumAssign(req.queryType, qt): - return err(ProtoError.IncorrectBlob) - - discard ?pb.getField(2, req.queryBytes) - ok(req) - -proc decode*( - _: type LookupResponse, data: openArray[byte] -): ProtoResult[LookupResponse] = - let pb = initProtoBuffer(data) - var - resp = LookupResponse() - codeOrd: uint32 - - if ?pb.getField(1, codeOrd): - if not checkedEnumAssign(resp.code, codeOrd): - return err(ProtoError.IncorrectBlob) - - discard ?pb.getRepeatedField(2, resp.providers) - ok(resp) +Protobuf.serializerForResult([LookupRequest, LookupResponse]) proc packProviders*( providers: seq[seq[byte]], budget_bytes: int diff --git a/storage/manifest/coders.nim b/storage/manifest/coders.nim index 8f0ae128..ef490708 100644 --- a/storage/manifest/coders.nim +++ b/storage/manifest/coders.nim @@ -7,144 +7,84 @@ ## This file may not be copied, modified, or distributed except according to ## those terms. -# This module implements serialization and deserialization of Manifest - -import times +## This module implements serialization and deserialization of Manifest. +## +## ```protobuf +## Message Manifest { +## required uint32 manifestVersion = 1; # manifest format version +## required bytes treeCid = 2; # cid (root) of the tree +## required uint32 blockSize = 3; # size of a single block +## required uint64 datasetSize = 4; # size of the dataset +## required codec: MultiCodec = 5; # Dataset codec +## required hcodec: MultiCodec = 6; # Multihash codec +## required version: CidVersion = 7; # Cid version +## optional filename: string = 8; # original filename +## optional mimetype: string = 9; # original mimetype +## } +## ``` {.push raises: [].} -import std/tables -import std/options - -import pkg/libp2p +import pkg/faststreams +import pkg/libp2p/cid +import pkg/protobuf_serialization +import pkg/protobuf_serialization/codec +import pkg/protobuf_serialization/std/enums import pkg/questionable import pkg/questionable/results -import pkg/chronos import ./manifest -import ../errors import ../blocktype -import ../logutils +import ../utils/protobuf/cid +import ../utils/protobuf/multicodec +import ../utils/protobuf/nbytes +import ../utils/protobuf/option +import ../utils/protobuf/refobject + +type ManifestEnvelope {.proto2.} = object + data {.fieldNumber: 1, required.}: seq[byte] + +proc encodeManifestFields(manifest: Manifest): seq[byte] {.raises: [IOError].} = + encode(Protobuf, manifest) + +proc decodeManifestFields( + fields: seq[byte] +): Manifest {.raises: [SerializationError].} = + decode(Protobuf, fields, Manifest) + +proc decodeEnvelope( + data: openArray[byte] +): ManifestEnvelope {.raises: [SerializationError].} = + decode(Protobuf, data, ManifestEnvelope) proc encode*(manifest: Manifest): ?!seq[byte] = - ## Encode the manifest into a ``ManifestCodec`` - ## multicodec container (Dag-pb) for now - ## - - var pbNode = initProtoBuffer() - - # NOTE: The `Data` field in the the `dag-pb` - # contains the following protobuf `Message` - # - # ```protobuf - # Message Header { - # required uint32 manifestVersion = 1; # manifest format version - # optional bytes treeCid = 2; # cid (root) of the tree - # optional uint32 blockSize = 3; # size of a single block - # optional uint64 datasetSize = 4; # size of the dataset - # optional codec: MultiCodec = 5; # Dataset codec - # optional hcodec: MultiCodec = 6; # Multihash codec - # optional version: CidVersion = 7; # Cid version - # optional filename: string = 8; # original filename - # optional mimetype: string = 9; # original mimetype - # } - # ``` - # - var header = initProtoBuffer() - header.write(1, manifest.manifestVersion) - header.write(2, manifest.treeCid.data.buffer) - header.write(3, manifest.blockSize.uint32) - header.write(4, manifest.datasetSize.uint64) - header.write(5, manifest.codec.uint32) - header.write(6, manifest.hcodec.uint32) - header.write(7, manifest.version.uint32) - - if manifest.filename.isSome: - header.write(8, manifest.filename.get()) - - if manifest.mimetype.isSome: - header.write(9, manifest.mimetype.get()) - - pbNode.write(1, header) # set the treeCid as the data field - pbNode.finish() - - return pbNode.buffer.success + try: + encode(Protobuf, ManifestEnvelope(data: encodeManifestFields(manifest))).success + except IOError as exc: + failure(exc.msg) proc decode*(_: type Manifest, data: openArray[byte]): ?!Manifest = - ## Decode a manifest from a data blob - ## + if data.len == 0: + return failure("Empty manifest input") - var - pbNode = initProtoBuffer(data) - pbHeader: ProtoBuffer - treeCidBuf: seq[byte] - datasetSize: uint64 - codec: uint32 - hcodec: uint32 - version: uint32 - blockSize: uint32 - manifestVersion: uint32 - filename: string - mimetype: string + let envelope = + try: + decodeEnvelope(data) + except SerializationError as exc: + return failure("Unable to decode manifest envelope: " & exc.msg) - # Decode `Header` message - if pbNode.getField(1, pbHeader).isErr: - return failure("Unable to decode `Header` from dag-pb manifest!") + let manifest = + try: + decodeManifestFields(envelope.data) + except SerializationError as exc: + return failure("Unable to decode manifest fields: " & exc.msg) - # Decode `Header` contents - if pbHeader.getField(1, manifestVersion).isErr: - return failure("Unable to decode `manifestVersion` from manifest!") + if manifest.manifestVersion != 0: + return failure("Unsupported manifest version: " & $manifest.manifestVersion) - if pbHeader.getField(2, treeCidBuf).isErr: - return failure("Unable to decode `treeCid` from manifest!") - - if pbHeader.getField(3, blockSize).isErr: - return failure("Unable to decode `blockSize` from manifest!") - - if pbHeader.getField(4, datasetSize).isErr: - return failure("Unable to decode `datasetSize` from manifest!") - - if pbHeader.getField(5, codec).isErr: - return failure("Unable to decode `codec` from manifest!") - - if pbHeader.getField(6, hcodec).isErr: - return failure("Unable to decode `hcodec` from manifest!") - - if pbHeader.getField(7, version).isErr: - return failure("Unable to decode `version` from manifest!") - - if pbHeader.getField(8, filename).isErr: - return failure("Unable to decode `filename` from manifest!") - - if pbHeader.getField(9, mimetype).isErr: - return failure("Unable to decode `mimetype` from manifest!") - - if manifestVersion != 0: - return failure("Unsupported manifest version: " & $manifestVersion) - - let treeCid = ?Cid.init(treeCidBuf).mapFailure - - var filenameOption = if filename.len == 0: string.none else: filename.some - var mimetypeOption = if mimetype.len == 0: string.none else: mimetype.some - - let self = Manifest.new( - treeCid = treeCid, - datasetSize = datasetSize.NBytes, - blockSize = blockSize.NBytes, - version = CidVersion(version), - hcodec = hcodec.MultiCodec, - codec = codec.MultiCodec, - filename = filenameOption, - mimetype = mimetypeOption, - manifestVersion = manifestVersion, - ) - - self.success - -func decode*(_: type Manifest, blk: Block): ?!Manifest = - ## Decode a manifest using `decoder` - ## + manifest.success +proc decode*(_: type Manifest, blk: Block): ?!Manifest = if not ?blk.cid.isManifest: return failure "Cid not a manifest codec" diff --git a/storage/manifest/manifest.nim b/storage/manifest/manifest.nim index 600e5a65..0ac05a53 100644 --- a/storage/manifest/manifest.nim +++ b/storage/manifest/manifest.nim @@ -11,8 +11,8 @@ {.push raises: [], gcsafe.} -import pkg/libp2p/protobuf/minprotobuf import pkg/libp2p/[cid, multihash, multicodec] +import pkg/protobuf_serialization import pkg/questionable/results import ../errors @@ -24,17 +24,21 @@ import ../logutils # TODO: Manifest should be reworked to more concrete types, # perhaps using inheritance -type Manifest* = ref object of RootObj - manifestVersion {.serialize.}: uint32 # Manifest format version - treeCid {.serialize.}: Cid # Root of the merkle tree - datasetSize {.serialize.}: NBytes # Total size of all blocks - blockSize {.serialize.}: NBytes +type Manifest* {.proto2.} = ref object of RootObj + manifestVersion {.serialize, fieldNumber: 1, required, pint.}: uint32 + # Manifest format version + treeCid {.serialize, fieldNumber: 2, required, ext.}: Cid # Root of the merkle tree + blockSize {.serialize, fieldNumber: 3, required, ext.}: NBytes # Size of each contained block (might not be needed if blocks are len-prefixed) - codec: MultiCodec # Dataset codec - hcodec: MultiCodec # Multihash codec - version: CidVersion # Cid version - filename {.serialize.}: ?string # The filename of the content uploaded (optional) - mimetype {.serialize.}: ?string # The mimetype of the content uploaded (optional) + datasetSize {.serialize, fieldNumber: 4, required, ext.}: NBytes + # Total size of all blocks + codec {.fieldNumber: 5, required, ext.}: MultiCodec # Dataset codec + hcodec {.fieldNumber: 6, required, ext.}: MultiCodec # Multihash codec + version {.fieldNumber: 7, required, ext.}: CidVersion # Cid version + filename {.serialize, fieldNumber: 8, ext.}: ?string + # The filename of the content uploaded (optional) + mimetype {.serialize, fieldNumber: 9, ext.}: ?string + # The mimetype of the content uploaded (optional) type ManifestDescriptor* = ref object manifest*: Manifest diff --git a/storage/stores/cachestore.nim b/storage/stores/cachestore.nim index 5550d87d..27e451f5 100644 --- a/storage/stores/cachestore.nim +++ b/storage/stores/cachestore.nim @@ -89,7 +89,8 @@ method getCidAndProof*( else: failure( newException( - BlockNotFoundError, "Block not in cache: " & $BlockAddress.init(treeCid, index) + BlockNotFoundError, + "Block not in cache: " & $BlockAddress.init(treeCid, index.uint64), ) ) diff --git a/storage/stores/networkstore.nim b/storage/stores/networkstore.nim index 19faacd5..c9627cb4 100644 --- a/storage/stores/networkstore.nim +++ b/storage/stores/networkstore.nim @@ -66,7 +66,7 @@ method getBlock*( ## Get a block from the blockstore ## - self.getBlock(BlockAddress.init(treeCid, index)) + self.getBlock(BlockAddress.init(treeCid, index.uint64)) method putBlock*( self: NetworkStore, blk: Block, ttl = Duration.none diff --git a/storage/streams/storestream.nim b/storage/streams/storestream.nim index c17b43c2..96097283 100644 --- a/storage/streams/storestream.nim +++ b/storage/streams/storestream.nim @@ -94,7 +94,7 @@ method readOnce*( self.manifest.blockSize.int - blockOffset, ] ) - address = BlockAddress(treeCid: self.manifest.treeCid, index: blockNum) + address = BlockAddress(treeCid: self.manifest.treeCid, index: blockNum.uint64) # Read contents of block `blockNum` without blk =? (await self.store.getBlock(address)).tryGet.catch, error: diff --git a/storage/utils/json.nim b/storage/utils/json.nim index 3450713e..d1f47028 100644 --- a/storage/utils/json.nim +++ b/storage/utils/json.nim @@ -1,4 +1,3 @@ -import std/typetraits from pkg/libp2p import Cid, PeerId, SignedPeerRecord, MultiAddress, AddressInfo, init, `$` import pkg/contractabi diff --git a/storage/utils/protobuf/cid.nim b/storage/utils/protobuf/cid.nim new file mode 100644 index 00000000..22b831db --- /dev/null +++ b/storage/utils/protobuf/cid.nim @@ -0,0 +1,43 @@ +## Logos Storage +## Copyright (c) 2026 Status Research & Development GmbH +## Licensed under either of +## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE)) +## * MIT license ([LICENSE-MIT](LICENSE-MIT)) +## at your option. +## This file may not be copied, modified, or distributed except according to +## those terms. + +{.push raises: [].} + +import pkg/faststreams +import pkg/libp2p/cid +import pkg/protobuf_serialization +import pkg/protobuf_serialization/codec + +Protobuf.extensionDefaults(Cid, defaultSeq = false, packed = false) + +func computeFieldSize*( + field: int, cid: Cid, ProtoType: type ProtobufExt, skipDefault: static bool +): int = + computeFieldSize(field, cid.data.buffer, pbytes, skipDefault) + +proc writeField*( + stream: OutputStream, + field: int, + cid: Cid, + ProtoType: type ProtobufExt, + skipDefault: static bool = false, +) {.raises: [IOError].} = + writeField(stream, field, cid.data.buffer, pbytes, skipDefault) + +proc readFieldInto*( + stream: InputStream, cid: var Cid, header: FieldHeader, ProtoType: type ProtobufExt +): bool {.raises: [SerializationError, IOError].} = + var bytes: seq[byte] + if not readFieldInto(stream, bytes, header, pbytes): + return false + let res = Cid.init(bytes) + if res.isErr: + raise newException(SerializationError, "Invalid Cid bytes: " & $res.error) + cid = res.get() + true diff --git a/storage/utils/protobuf/multicodec.nim b/storage/utils/protobuf/multicodec.nim new file mode 100644 index 00000000..f3c7f1c1 --- /dev/null +++ b/storage/utils/protobuf/multicodec.nim @@ -0,0 +1,44 @@ +## Logos Storage +## Copyright (c) 2026 Status Research & Development GmbH +## Licensed under either of +## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE)) +## * MIT license ([LICENSE-MIT](LICENSE-MIT)) +## at your option. +## This file may not be copied, modified, or distributed except according to +## those terms. + +{.push raises: [].} + +import pkg/faststreams +import pkg/libp2p/multicodec +import pkg/protobuf_serialization +import pkg/protobuf_serialization/codec + +Protobuf.extensionDefaults(MultiCodec) + +func computeFieldSize*( + field: int, value: MultiCodec, ProtoType: type ProtobufExt, skipDefault: static bool +): int = + computeFieldSize(field, uint32(value), puint32, skipDefault) + +proc writeField*( + stream: OutputStream, + field: int, + value: MultiCodec, + ProtoType: type ProtobufExt, + skipDefault: static bool = false, +) {.raises: [IOError].} = + writeField(stream, field, uint32(value), puint32, skipDefault) + +proc readFieldInto*( + stream: InputStream, + value: var MultiCodec, + header: FieldHeader, + ProtoType: type ProtobufExt, +): bool {.raises: [SerializationError, IOError].} = + var raw: uint32 + if readFieldInto(stream, raw, header, puint32): + value = MultiCodec(raw) + true + else: + false diff --git a/storage/utils/protobuf/nbytes.nim b/storage/utils/protobuf/nbytes.nim new file mode 100644 index 00000000..d49005c9 --- /dev/null +++ b/storage/utils/protobuf/nbytes.nim @@ -0,0 +1,45 @@ +## Logos Storage +## Copyright (c) 2026 Status Research & Development GmbH +## Licensed under either of +## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE)) +## * MIT license ([LICENSE-MIT](LICENSE-MIT)) +## at your option. +## This file may not be copied, modified, or distributed except according to +## those terms. + +{.push raises: [].} + +import pkg/faststreams +import pkg/protobuf_serialization +import pkg/protobuf_serialization/codec + +import ../../units + +Protobuf.extensionDefaults(NBytes) + +func computeFieldSize*( + field: int, value: NBytes, ProtoType: type ProtobufExt, skipDefault: static bool +): int = + computeFieldSize(field, uint64(value), puint64, skipDefault) + +proc writeField*( + stream: OutputStream, + field: int, + value: NBytes, + ProtoType: type ProtobufExt, + skipDefault: static bool = false, +) {.raises: [IOError].} = + writeField(stream, field, uint64(value), puint64, skipDefault) + +proc readFieldInto*( + stream: InputStream, + value: var NBytes, + header: FieldHeader, + ProtoType: type ProtobufExt, +): bool {.raises: [SerializationError, IOError].} = + var raw: uint64 + if readFieldInto(stream, raw, header, puint64): + value = NBytes(raw) + true + else: + false diff --git a/storage/utils/protobuf/option.nim b/storage/utils/protobuf/option.nim new file mode 100644 index 00000000..99a3b9f5 --- /dev/null +++ b/storage/utils/protobuf/option.nim @@ -0,0 +1,59 @@ +## Logos Storage +## Copyright (c) 2026 Status Research & Development GmbH +## Licensed under either of +## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE)) +## * MIT license ([LICENSE-MIT](LICENSE-MIT)) +## at your option. +## This file may not be copied, modified, or distributed except according to +## those terms. + +{.push raises: [].} + +import std/options + +import pkg/faststreams +import pkg/protobuf_serialization +import pkg/protobuf_serialization/codec + +Protobuf.extensionDefaults(Option[string]) + +template isExtension*(T: type Protobuf, FieldType: type Option[string]): bool = + true + +template flatType*(T: type Protobuf, value: Option[string]): type = + string + +func computeFieldSize*( + field: int, + value: Option[string], + ProtoType: type ProtobufExt, + skipDefault: static bool, +): int = + if value.isSome: + computeFieldSize(field, value.get, pstring, skipDefault) + else: + 0 + +proc writeField*( + stream: OutputStream, + field: int, + value: Option[string], + ProtoType: type ProtobufExt, + skipDefault: static bool = false, +) {.raises: [IOError].} = + if value.isSome: + writeField(stream, field, value.get, pstring, skipDefault) + +proc readFieldInto*( + stream: InputStream, + value: var Option[string], + header: FieldHeader, + ProtoType: type ProtobufExt, +): bool {.raises: [SerializationError, IOError].} = + var raw: string + if readFieldInto(stream, raw, header, pstring): + if raw.len > 0: + value = some(raw) + true + else: + false diff --git a/storage/utils/protobuf/refobject.nim b/storage/utils/protobuf/refobject.nim new file mode 100644 index 00000000..e09ee30e --- /dev/null +++ b/storage/utils/protobuf/refobject.nim @@ -0,0 +1,25 @@ +## Logos Storage +## Copyright (c) 2026 Status Research & Development GmbH +## Licensed under either of +## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE)) +## * MIT license ([LICENSE-MIT](LICENSE-MIT)) +## at your option. +## This file may not be copied, modified, or distributed except according to +## those terms. + +{.push raises: [].} + +import pkg/protobuf_serialization + +proc readValue*[T: ref object]( + reader: ProtobufReader, value: var T +) {.raises: [SerializationError, IOError].} = + if value.isNil: + new(value) + readValue(reader, value[]) + +proc writeValue*[T: ref object]( + writer: ProtobufWriter, value: T +) {.raises: [IOError].} = + if not value.isNil: + writeValue(writer, value[]) diff --git a/storage/utils/protobuf/serializer.nim b/storage/utils/protobuf/serializer.nim new file mode 100644 index 00000000..b2bc1999 --- /dev/null +++ b/storage/utils/protobuf/serializer.nim @@ -0,0 +1,39 @@ +## Logos Storage +## Copyright (c) 2026 Status Research & Development GmbH +## Licensed under either of +## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE)) +## * MIT license ([LICENSE-MIT](LICENSE-MIT)) +## at your option. +## This file may not be copied, modified, or distributed except according to +## those terms. + +{.push raises: [].} + +import std/macros +import pkg/protobuf_serialization +import pkg/protobuf_serialization/pkg/results + +type ProtobufDecodeError* = object + msg*: string + +proc `$`*(e: ProtobufDecodeError): string = + e.msg + +macro serializerForResult*(_: type Protobuf, Types: untyped): untyped = + var stmts = newStmtList() + for T in Types: + let helperName = genSym(nskProc, "decode" & $T) + stmts.add quote do: + proc encode*(c: `T`): seq[byte] = + encode(Protobuf, c) + + proc `helperName`(buf: seq[byte]): `T` {.raises: [SerializationError].} = + decode(Protobuf, buf, `T`) + + proc decode*(_: type `T`, buf: seq[byte]): Result[`T`, ProtobufDecodeError] = + try: + ok(`helperName`(buf)) + except SerializationError as exc: + err(ProtobufDecodeError(msg: exc.msg)) + + stmts diff --git a/tests/cbindings/storage.c b/tests/cbindings/storage.c index aae45529..9fe14085 100644 --- a/tests/cbindings/storage.c +++ b/tests/cbindings/storage.c @@ -745,7 +745,7 @@ int check_download_manifest(void *storage_ctx, const char *cid) int ret = is_resp_ok(r, &res); - const char *expected_manifest = "{\"manifestVersion\":0,\"treeCid\":\"zDzSvJTf8JYwvysKPmG7BtzpbiAHfuwFMRphxm4hdvnMJ4XPJjKX\",\"datasetSize\":12,\"blockSize\":65536,\"filename\":\"hello_world.txt\",\"mimetype\":\"text/plain\"}"; + const char *expected_manifest = "{\"manifestVersion\":0,\"treeCid\":\"zDzSvJTf8JYwvysKPmG7BtzpbiAHfuwFMRphxm4hdvnMJ4XPJjKX\",\"blockSize\":65536,\"datasetSize\":12,\"filename\":\"hello_world.txt\",\"mimetype\":\"text/plain\"}"; if (res == NULL || strncmp(res, expected_manifest, strlen(expected_manifest)) != 0) { @@ -771,7 +771,7 @@ int check_list(void *storage_ctx) int ret = is_resp_ok(r, &res); - const char *expected_manifest = "{\"manifestVersion\":0,\"treeCid\":\"zDzSvJTf8JYwvysKPmG7BtzpbiAHfuwFMRphxm4hdvnMJ4XPJjKX\",\"datasetSize\":12,\"blockSize\":65536,\"filename\":\"hello_world.txt\",\"mimetype\":\"text/plain\"}"; + const char *expected_manifest = "{\"manifestVersion\":0,\"treeCid\":\"zDzSvJTf8JYwvysKPmG7BtzpbiAHfuwFMRphxm4hdvnMJ4XPJjKX\",\"blockSize\":65536,\"datasetSize\":12,\"filename\":\"hello_world.txt\",\"mimetype\":\"text/plain\"}"; if (res == NULL || strstr(res, expected_manifest) == NULL) { diff --git a/tests/storage/blockexchange/engine/testblockexc.nim b/tests/storage/blockexchange/engine/testblockexc.nim index 43cde5aa..002893b2 100644 --- a/tests/storage/blockexchange/engine/testblockexc.nim +++ b/tests/storage/blockexchange/engine/testblockexc.nim @@ -199,13 +199,14 @@ asyncchecksuite "BlockExchange - Multi-Peer Download": desc = DownloadDesc(md: dataset.manifestDesc, count: dataset.blocks.len.uint64) download = leecher.downloadManager.startDownload(desc) halfPoint = (dataset.blocks.len div 2).uint64 - ranges1 = @[(start: 0'u64, count: halfPoint)] + ranges1 = @[IndexRange(start: 0'u64, count: halfPoint)] download.updatePeerAvailability( seeder1.switch.peerInfo.peerId, BlockAvailability.fromRanges(ranges1) ) - let ranges2 = @[(start: halfPoint, count: dataset.blocks.len.uint64 - halfPoint)] + let ranges2 = + @[IndexRange(start: halfPoint, count: dataset.blocks.len.uint64 - halfPoint)] download.updatePeerAvailability( seeder2.switch.peerInfo.peerId, BlockAvailability.fromRanges(ranges2) ) @@ -364,7 +365,7 @@ asyncchecksuite "BlockExchange - Error Handling": blockSize = dataset.manifest.blockSize.uint32 desc = DownloadDesc(md: dataset.manifestDesc, count: dataset.blocks.len.uint64) download = leecher.downloadManager.startDownload(desc) - ranges = @[(start: 0'u64, count: 2'u64)] + ranges = @[IndexRange(start: 0'u64, count: 2'u64)] download.updatePeerAvailability( seeder.switch.peerInfo.peerId, BlockAvailability.fromRanges(ranges) diff --git a/tests/storage/blockexchange/engine/testbroadcastavailability.nim b/tests/storage/blockexchange/engine/testbroadcastavailability.nim index ce36e98f..713f5dfe 100644 --- a/tests/storage/blockexchange/engine/testbroadcastavailability.nim +++ b/tests/storage/blockexchange/engine/testbroadcastavailability.nim @@ -6,6 +6,7 @@ import pkg/chronos import pkg/storage/blockexchange/engine/downloadcontext {.all.} import pkg/storage/blockexchange/engine/scheduler {.all.} +import pkg/storage/blockexchange/protocol/message privateAccess(BroadcastAvailabilityTracker) @@ -45,7 +46,7 @@ suite "BroadcastAvailabilityTracker (sequential OOO)": check tracker.shouldBroadcast(sched) let ranges = tracker.getRanges(sched) - check ranges == @[(start: 100'u64, count: BatchSize)] + check ranges == @[IndexRange(start: 100'u64, count: BatchSize)] check tracker.pendingOOOSnapshot == toHashSet([100'u64]) tracker.markBroadcasted(sched) @@ -62,7 +63,7 @@ suite "BroadcastAvailabilityTracker (sequential OOO)": sched.markComplete(200) check tracker.shouldBroadcast(sched) let ranges = tracker.getRanges(sched) - check ranges == @[(start: 200'u64, count: BatchSize)] + check ranges == @[IndexRange(start: 200'u64, count: BatchSize)] test "Multiple new OOO batches are all emitted in a single broadcast": for _ in 0 ..< 4: @@ -96,7 +97,7 @@ suite "BroadcastAvailabilityTracker (sequential OOO)": check tracker.shouldBroadcast(sched) let ranges = tracker.getRanges(sched) - check ranges == @[(start: 0'u64, count: 200'u64)] + check ranges == @[IndexRange(start: 0'u64, count: 200'u64)] tracker.markBroadcasted(sched) check tracker.broadcastedOutOfOrder.len == 0 @@ -115,7 +116,7 @@ suite "BroadcastAvailabilityTracker (sequential OOO)": check tracker.shouldBroadcast(sched) let ranges = tracker.getRanges(sched) - check ranges == @[(start: 0'u64, count: BatchSize)] + check ranges == @[IndexRange(start: 0'u64, count: BatchSize)] check tracker.pendingOOOSnapshot.len == 0 tracker.markBroadcasted(sched) @@ -128,7 +129,7 @@ suite "BroadcastAvailabilityTracker (sequential OOO)": sched.markComplete(100) let ranges = tracker.getRanges(sched) - check ranges == @[(start: 100'u64, count: BatchSize)] + check ranges == @[IndexRange(start: 100'u64, count: BatchSize)] check tracker.pendingOOOSnapshot == toHashSet([100'u64]) sched.markComplete(200) @@ -139,7 +140,7 @@ suite "BroadcastAvailabilityTracker (sequential OOO)": check tracker.shouldBroadcast(sched) let ranges2 = tracker.getRanges(sched) - check ranges2 == @[(start: 200'u64, count: BatchSize)] + check ranges2 == @[IndexRange(start: 200'u64, count: BatchSize)] check tracker.pendingOOOSnapshot == toHashSet([200'u64]) test "getRanges clears stale snapshot when mark was skipped": @@ -153,5 +154,5 @@ suite "BroadcastAvailabilityTracker (sequential OOO)": check sched.completedWatermark() == 200 let ranges = tracker.getRanges(sched) - check ranges == @[(start: 0'u64, count: 200'u64)] + check ranges == @[IndexRange(start: 0'u64, count: 200'u64)] check tracker.pendingOOOSnapshot.len == 0 diff --git a/tests/storage/blockexchange/engine/testengine.nim b/tests/storage/blockexchange/engine/testengine.nim index b806e52b..d6c22d53 100644 --- a/tests/storage/blockexchange/engine/testengine.nim +++ b/tests/storage/blockexchange/engine/testengine.nim @@ -259,8 +259,8 @@ asyncchecksuite "NetworkStore engine handlers": ) {.async: (raises: [CancelledError]).} = check presence.len == 1 check presence[0].kind == BlockPresenceType.HaveRange - for (start, count) in presence[0].ranges: - check start < 2 + for r in presence[0].ranges: + check r.start < 2 done.complete() engine.network = @@ -270,8 +270,11 @@ asyncchecksuite "NetworkStore engine handlers": await done test "WantBlocks: rejects range with count = 0": - let req = - WantBlocksRequest(requestId: 1, treeCid: Cid.example, ranges: @[(0'u64, 0'u64)]) + let req = WantBlocksRequest( + requestId: 1, + treeCid: Cid.example, + ranges: @[IndexRange(start: 0'u64, count: 0'u64)], + ) let blocks = await network.handlers.onWantBlocksRequest(peerId, req) check blocks.len == 0 @@ -280,7 +283,7 @@ asyncchecksuite "NetworkStore engine handlers": let req = WantBlocksRequest( requestId: 1, treeCid: Cid.example, - ranges: @[(0'u64, MaxBlocksPerBatch.uint64 + 1)], + ranges: @[IndexRange(start: 0'u64, count: MaxBlocksPerBatch.uint64 + 1)], ) let blocks = await network.handlers.onWantBlocksRequest(peerId, req) @@ -288,7 +291,9 @@ asyncchecksuite "NetworkStore engine handlers": test "WantBlocks: rejects range whose start+count overflows": let req = WantBlocksRequest( - requestId: 1, treeCid: Cid.example, ranges: @[(uint64.high, 1'u64)] + requestId: 1, + treeCid: Cid.example, + ranges: @[IndexRange(start: uint64.high, count: 1'u64)], ) let blocks = await network.handlers.onWantBlocksRequest(peerId, req) @@ -296,17 +301,19 @@ asyncchecksuite "NetworkStore engine handlers": test "WantBlocks: rejects range whose max index exceeds Natural": let req = WantBlocksRequest( - requestId: 1, treeCid: Cid.example, ranges: @[(high(Natural).uint64 + 1, 1'u64)] + requestId: 1, + treeCid: Cid.example, + ranges: @[IndexRange(start: high(Natural).uint64 + 1, count: 1'u64)], ) let blocks = await network.handlers.onWantBlocksRequest(peerId, req) check blocks.len == 0 test "WantBlocks: rejects when total count across ranges exceeds cap": - var ranges: seq[tuple[start: uint64, count: uint64]] = @[] + var ranges: seq[IndexRange] = @[] let halfMaxBlocksPerBatchPlusOne = (MaxBlocksPerBatch div 2).uint64 + 1 - ranges.add((0'u64, halfMaxBlocksPerBatchPlusOne)) - ranges.add((10_000'u64, halfMaxBlocksPerBatchPlusOne)) + ranges.add(IndexRange(start: 0'u64, count: halfMaxBlocksPerBatchPlusOne)) + ranges.add(IndexRange(start: 10_000'u64, count: halfMaxBlocksPerBatchPlusOne)) let req = WantBlocksRequest(requestId: 1, treeCid: Cid.example, ranges: ranges) @@ -323,7 +330,9 @@ asyncchecksuite "NetworkStore engine handlers": (await localStore.putCidAndProof(rootCid, i, blk.cid, tree.getProof(i).tryGet())).tryGet() let req = WantBlocksRequest( - requestId: 1, treeCid: rootCid, ranges: @[(0'u64, blocks.len.uint64)] + requestId: 1, + treeCid: rootCid, + ranges: @[IndexRange(start: 0'u64, count: blocks.len.uint64)], ) let delivered = await network.handlers.onWantBlocksRequest(peerId, req) @@ -331,24 +340,28 @@ asyncchecksuite "NetworkStore engine handlers": suite "IsIndexInRanges": test "Empty ranges returns false": - let ranges: seq[(uint64, uint64)] = @[] + let ranges: seq[IndexRange] = @[] check not isIndexInRanges(0, ranges) check not isIndexInRanges(100, ranges) test "Single range - index inside": - let ranges = @[(10'u64, 5'u64)] + let ranges = @[IndexRange(start: 10'u64, count: 5'u64)] check isIndexInRanges(10, ranges, sortedRanges = true) check isIndexInRanges(12, ranges, sortedRanges = true) check isIndexInRanges(14, ranges, sortedRanges = true) test "Single range - index outside": - let ranges = @[(10'u64, 5'u64)] + let ranges = @[IndexRange(start: 10'u64, count: 5'u64)] check not isIndexInRanges(9, ranges, sortedRanges = true) check not isIndexInRanges(15, ranges, sortedRanges = true) check not isIndexInRanges(100, ranges, sortedRanges = true) test "Multiple sorted ranges - index in each": - let ranges = @[(0'u64, 3'u64), (10'u64, 5'u64), (100'u64, 10'u64)] + let ranges = @[ + IndexRange(start: 0'u64, count: 3'u64), + IndexRange(start: 10'u64, count: 5'u64), + IndexRange(start: 100'u64, count: 10'u64), + ] check isIndexInRanges(0, ranges, sortedRanges = true) check isIndexInRanges(2, ranges, sortedRanges = true) check isIndexInRanges(10, ranges, sortedRanges = true) @@ -357,7 +370,11 @@ suite "IsIndexInRanges": check isIndexInRanges(109, ranges, sortedRanges = true) test "Multiple ranges - index in gaps": - let ranges = @[(0'u64, 3'u64), (10'u64, 5'u64), (100'u64, 10'u64)] + let ranges = @[ + IndexRange(start: 0'u64, count: 3'u64), + IndexRange(start: 10'u64, count: 5'u64), + IndexRange(start: 100'u64, count: 10'u64), + ] check not isIndexInRanges(3, ranges, sortedRanges = true) check not isIndexInRanges(9, ranges, sortedRanges = true) check not isIndexInRanges(15, ranges, sortedRanges = true) @@ -365,7 +382,11 @@ suite "IsIndexInRanges": check not isIndexInRanges(110, ranges, sortedRanges = true) test "Unsorted ranges with sortedRanges=false": - let ranges = @[(100'u64, 10'u64), (0'u64, 3'u64), (10'u64, 5'u64)] + let ranges = @[ + IndexRange(start: 100'u64, count: 10'u64), + IndexRange(start: 0'u64, count: 3'u64), + IndexRange(start: 10'u64, count: 5'u64), + ] check isIndexInRanges(0, ranges, sortedRanges = false) check isIndexInRanges(2, ranges, sortedRanges = false) check isIndexInRanges(10, ranges, sortedRanges = false) @@ -373,13 +394,17 @@ suite "IsIndexInRanges": check not isIndexInRanges(50, ranges, sortedRanges = false) test "Adjacent ranges": - let ranges = @[(0'u64, 5'u64), (5'u64, 5'u64), (10'u64, 5'u64)] + let ranges = @[ + IndexRange(start: 0'u64, count: 5'u64), + IndexRange(start: 5'u64, count: 5'u64), + IndexRange(start: 10'u64, count: 5'u64), + ] for i in 0'u64 ..< 15: check isIndexInRanges(i, ranges, sortedRanges = true) check not isIndexInRanges(15, ranges, sortedRanges = true) test "Large range values": - let ranges = @[(1_000_000_000'u64, 1_000_000'u64)] + let ranges = @[IndexRange(start: 1_000_000_000'u64, count: 1_000_000'u64)] check isIndexInRanges(1_000_000_000, ranges, sortedRanges = true) check isIndexInRanges(1_000_500_000, ranges, sortedRanges = true) check not isIndexInRanges(999_999_999, ranges, sortedRanges = true) diff --git a/tests/storage/blockexchange/engine/testswarm.nim b/tests/storage/blockexchange/engine/testswarm.nim index 3458a4c8..0e30cbd6 100644 --- a/tests/storage/blockexchange/engine/testswarm.nim +++ b/tests/storage/blockexchange/engine/testswarm.nim @@ -9,6 +9,7 @@ import pkg/storage/blockexchange/engine/swarm import pkg/storage/blockexchange/engine/peertracker import pkg/storage/blockexchange/peers/peercontext import pkg/storage/blockexchange/peers/peerstats +import pkg/storage/blockexchange/protocol/message import pkg/storage/blockexchange/utils import pkg/storage/storagetypes @@ -38,7 +39,10 @@ suite "BlockAvailability": test "ranges availability - hasBlock": let avail = BlockAvailability.fromRanges( - @[(start: 10'u64, count: 20'u64), (start: 50'u64, count: 10'u64)] + @[ + IndexRange(start: 10'u64, count: 20'u64), + IndexRange(start: 50'u64, count: 10'u64), + ] ) check avail.kind == bakRanges @@ -56,7 +60,10 @@ suite "BlockAvailability": test "ranges availability - hasRange": let avail = BlockAvailability.fromRanges( - @[(start: 10'u64, count: 20'u64), (start: 50'u64, count: 10'u64)] + @[ + IndexRange(start: 10'u64, count: 20'u64), + IndexRange(start: 50'u64, count: 10'u64), + ] ) check avail.hasRange(10, 20) == true @@ -72,7 +79,10 @@ suite "BlockAvailability": test "ranges availability - hasAnyInRange": let avail = BlockAvailability.fromRanges( - @[(start: 10'u64, count: 20'u64), (start: 50'u64, count: 10'u64)] + @[ + IndexRange(start: 10'u64, count: 20'u64), + IndexRange(start: 50'u64, count: 10'u64), + ] ) check avail.hasAnyInRange(5, 10) == true @@ -129,15 +139,15 @@ suite "BlockAvailability": test "merge unknown with ranges": let unknown = BlockAvailability.unknown() - ranges = BlockAvailability.fromRanges(@[(start: 10'u64, count: 20'u64)]) + ranges = BlockAvailability.fromRanges(@[IndexRange(start: 10'u64, count: 20'u64)]) merged = unknown.merge(ranges) check merged.kind == bakRanges check merged.hasBlock(15) == true test "merge ranges with ranges": let - r1 = BlockAvailability.fromRanges(@[(start: 0'u64, count: 10'u64)]) - r2 = BlockAvailability.fromRanges(@[(start: 20'u64, count: 10'u64)]) + r1 = BlockAvailability.fromRanges(@[IndexRange(start: 0'u64, count: 10'u64)]) + r2 = BlockAvailability.fromRanges(@[IndexRange(start: 20'u64, count: 10'u64)]) merged = r1.merge(r2) check merged.kind == bakRanges @@ -147,8 +157,8 @@ suite "BlockAvailability": test "merge overlapping ranges": let - r1 = BlockAvailability.fromRanges(@[(start: 0'u64, count: 15'u64)]) - r2 = BlockAvailability.fromRanges(@[(start: 10'u64, count: 15'u64)]) + r1 = BlockAvailability.fromRanges(@[IndexRange(start: 0'u64, count: 15'u64)]) + r2 = BlockAvailability.fromRanges(@[IndexRange(start: 10'u64, count: 15'u64)]) merged = r1.merge(r2) check merged.kind == bakRanges @@ -159,13 +169,13 @@ suite "BlockAvailability": test "merge bitmap with ranges converts bitmap to ranges": let bitmap = BlockAvailability.fromBitmap(@[0x0F'u8], 8) - ranges = BlockAvailability.fromRanges(@[(start: 6'u64, count: 2'u64)]) + ranges = BlockAvailability.fromRanges(@[IndexRange(start: 6'u64, count: 2'u64)]) merged = bitmap.merge(ranges) check merged.kind == bakRanges check merged.ranges.len == 2 - check merged.ranges[0] == (start: 0'u64, count: 4'u64) - check merged.ranges[1] == (start: 6'u64, count: 2'u64) + check merged.ranges[0] == IndexRange(start: 0'u64, count: 4'u64) + check merged.ranges[1] == IndexRange(start: 6'u64, count: 2'u64) suite "SwarmPeer": test "touch updates lastSeen": @@ -176,10 +186,11 @@ suite "SwarmPeer": check peer.lastSeen >= before test "updateAvailability merges": - let peer = - SwarmPeer.new(BlockAvailability.fromRanges(@[(start: 0'u64, count: 10'u64)])) + let peer = SwarmPeer.new( + BlockAvailability.fromRanges(@[IndexRange(start: 0'u64, count: 10'u64)]) + ) peer.updateAvailability( - BlockAvailability.fromRanges(@[(start: 20'u64, count: 10'u64)]) + BlockAvailability.fromRanges(@[IndexRange(start: 20'u64, count: 10'u64)]) ) check peer.availability.hasBlock(5) == true @@ -243,11 +254,11 @@ suite "Swarm": test "updatePeerAvailability": let peerId = PeerId.example discard swarm.addPeer( - peerId, BlockAvailability.fromRanges(@[(start: 0'u64, count: 10'u64)]) + peerId, BlockAvailability.fromRanges(@[IndexRange(start: 0'u64, count: 10'u64)]) ) swarm.updatePeerAvailability( - peerId, BlockAvailability.fromRanges(@[(start: 20'u64, count: 10'u64)]) + peerId, BlockAvailability.fromRanges(@[IndexRange(start: 20'u64, count: 10'u64)]) ) let peer = swarm.getPeer(peerId).get() @@ -272,7 +283,7 @@ suite "Swarm": discard swarm.addPeer(peer1, BlockAvailability.complete()) discard swarm.addPeer( - peer2, BlockAvailability.fromRanges(@[(start: 0'u64, count: 100'u64)]) + peer2, BlockAvailability.fromRanges(@[IndexRange(start: 0'u64, count: 100'u64)]) ) let peersForRange = swarm.peersWithRange(0, 50) @@ -287,10 +298,10 @@ suite "Swarm": peer2 = PeerId.example discard swarm.addPeer( - peer1, BlockAvailability.fromRanges(@[(start: 0'u64, count: 50'u64)]) + peer1, BlockAvailability.fromRanges(@[IndexRange(start: 0'u64, count: 50'u64)]) ) discard swarm.addPeer( - peer2, BlockAvailability.fromRanges(@[(start: 100'u64, count: 50'u64)]) + peer2, BlockAvailability.fromRanges(@[IndexRange(start: 100'u64, count: 50'u64)]) ) let peers1 = swarm.peersWithAnyInRange(25, 50) diff --git a/tests/storage/blockexchange/protocol/testmessage.nim b/tests/storage/blockexchange/protocol/testmessage.nim index 942e2fa0..dd268852 100644 --- a/tests/storage/blockexchange/protocol/testmessage.nim +++ b/tests/storage/blockexchange/protocol/testmessage.nim @@ -1,4 +1,5 @@ import std/sequtils +import std/strutils import pkg/unittest2 @@ -8,244 +9,12 @@ import pkg/storage/blockexchange/protocol/message import ../../examples import ../../helpers -suite "BlockAddress protobuf encoding": - test "Should encode and decode block address": - let - treeCid = Cid.example - address = BlockAddress(treeCid: treeCid, index: 42) - - var buffer = initProtoBuffer() - buffer.write(1, address) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = BlockAddress.decode(decoded) - check res.isOk - check res.get.treeCid == treeCid - check res.get.index == 42 - - test "Should encode and decode block address with index 0": - let - blockCid = Cid.example - address = BlockAddress(treeCid: blockCid, index: 0) - - var buffer = initProtoBuffer() - buffer.write(1, address) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = BlockAddress.decode(decoded) - check res.isOk - check res.get.treeCid == blockCid - check res.get.index == 0 - -suite "WantListEntry protobuf encoding": - test "Should encode and decode WantListEntry": - let - treeCid = Cid.example - entry = WantListEntry( - address: BlockAddress(treeCid: treeCid, index: 10), - priority: 5, - cancel: false, - wantType: WantType.WantHave, - sendDontHave: true, - rangeCount: 100, - ) - - var buffer = initProtoBuffer() - buffer.write(1, entry) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = WantListEntry.decode(decoded) - check res.isOk - check res.get.address.treeCid == treeCid - check res.get.address.index == 10 - check res.get.priority == 5 - check res.get.cancel == false - check res.get.wantType == WantType.WantHave - check res.get.sendDontHave == true - check res.get.rangeCount == 100 - - test "Should handle WantListEntry with cancel flag": - let - blockCid = Cid.example - entry = WantListEntry( - address: BlockAddress(treeCid: blockCid, index: 0), - priority: 1, - cancel: true, - wantType: WantType.WantHave, - sendDontHave: false, - rangeCount: 0, - ) - - var buffer = initProtoBuffer() - buffer.write(1, entry) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = WantListEntry.decode(decoded) - check res.isOk - check res.get.cancel == true - -suite "WantList protobuf encoding": - test "Should encode and decode empty WantList": - let wantList = WantList(entries: @[], full: false) - - var buffer = initProtoBuffer() - buffer.write(1, wantList) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = WantList.decode(decoded) - check res.isOk - check res.get.entries.len == 0 - check res.get.full == false - - test "Should encode and decode WantList with entries": - let - treeCid = Cid.example - wantList = WantList( - entries: @[ - WantListEntry( - address: BlockAddress(treeCid: treeCid, index: 0), - priority: 1, - cancel: false, - wantType: WantType.WantHave, - sendDontHave: false, - rangeCount: 10, - ), - WantListEntry( - address: BlockAddress(treeCid: treeCid, index: 1), - priority: 2, - cancel: false, - wantType: WantType.WantHave, - sendDontHave: true, - rangeCount: 0, - ), - ], - full: true, - ) - - var buffer = initProtoBuffer() - buffer.write(1, wantList) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = WantList.decode(decoded) - check res.isOk - check res.get.entries.len == 2 - check res.get.entries[0].rangeCount == 10 - check res.get.entries[1].sendDontHave == true - check res.get.full == true - - test "Should reject WantList with too many entries": - let - treeCid = Cid.example - wantList = WantList( - entries: newSeqWith( - MaxWantListEntries + 1, - WantListEntry(address: BlockAddress(treeCid: treeCid, index: 0)), - ), - full: false, - ) - - var buffer = initProtoBuffer() - buffer.write(1, wantList) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = WantList.decode(decoded) - check res.isErr - check res.error == ProtoError.BufferOverflow - -suite "BlockPresence protobuf encoding": - test "Should encode and decode BlockPresence with DontHave": - let - treeCid = Cid.example - presence = BlockPresence( - address: BlockAddress(treeCid: treeCid, index: 0), - kind: BlockPresenceType.DontHave, - ranges: @[], - ) - - var buffer = initProtoBuffer() - buffer.write(1, presence) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = BlockPresence.decode(decoded) - check res.isOk - check res.get.kind == BlockPresenceType.DontHave - check res.get.ranges.len == 0 - - test "Should encode and decode BlockPresence with HaveRange": - let - treeCid = Cid.example - presence = BlockPresence( - address: BlockAddress(treeCid: treeCid, index: 0), - kind: BlockPresenceType.HaveRange, - ranges: @[(start: 0'u64, count: 100'u64), (start: 200'u64, count: 50'u64)], - ) - - var buffer = initProtoBuffer() - buffer.write(1, presence) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = BlockPresence.decode(decoded) - check res.isOk - check res.get.kind == BlockPresenceType.HaveRange - check res.get.ranges.len == 2 - check res.get.ranges[0].start == 0 - check res.get.ranges[0].count == 100 - check res.get.ranges[1].start == 200 - check res.get.ranges[1].count == 50 - - test "Should encode and decode BlockPresence with Complete": - let - treeCid = Cid.example - presence = BlockPresence( - address: BlockAddress(treeCid: treeCid, index: 0), - kind: BlockPresenceType.Complete, - ranges: @[], - ) - - var buffer = initProtoBuffer() - buffer.write(1, presence) - buffer.finish() - - var decoded: ProtoBuffer - check buffer.getField(1, decoded).isOk - - let res = BlockPresence.decode(decoded) - check res.isOk - check res.get.kind == BlockPresenceType.Complete - suite "Full Message protobuf encoding": test "Should encode and decode empty Message": let msg = Message(wantList: WantList(entries: @[], full: false), blockPresences: @[]) - encoded = msg.protobufEncode() - decoded = Message.protobufDecode(encoded) + encoded = msg.encode() + decoded = Message.decode(encoded) check decoded.isOk check decoded.get.wantList.entries.len == 0 @@ -270,8 +39,8 @@ suite "Full Message protobuf encoding": ), blockPresences: @[], ) - encoded = msg.protobufEncode() - decoded = Message.protobufDecode(encoded) + encoded = msg.encode() + decoded = Message.decode(encoded) check decoded.isOk check decoded.get.wantList.entries.len == 1 @@ -286,12 +55,12 @@ suite "Full Message protobuf encoding": BlockPresence( address: BlockAddress(treeCid: treeCid, index: 0), kind: BlockPresenceType.HaveRange, - ranges: @[(start: 0'u64, count: 500'u64)], + ranges: @[IndexRange(start: 0'u64, count: 500'u64)], ) ], ) - encoded = msg.protobufEncode() - decoded = Message.protobufDecode(encoded) + encoded = msg.encode() + decoded = Message.decode(encoded) check decoded.isOk check decoded.get.blockPresences.len == 1 @@ -313,8 +82,34 @@ suite "Full Message protobuf encoding": ), ), ) - encoded = msg.protobufEncode() - decoded = Message.protobufDecode(encoded) + encoded = msg.encode() + decoded = Message.decode(encoded) check decoded.isErr - check decoded.error == ProtoError.BufferOverflow + check "exceeds" in decoded.error.msg + + test "Should reject Message with too many wantList entries": + let + treeCid = Cid.example + msg = Message( + wantList: WantList( + entries: newSeqWith( + MaxWantListEntries + 1, + WantListEntry( + address: BlockAddress(treeCid: treeCid, index: 0), + priority: 1, + cancel: false, + wantType: WantType.WantHave, + sendDontHave: false, + rangeCount: 0, + ), + ), + full: false, + ), + blockPresences: @[], + ) + encoded = msg.encode() + decoded = Message.decode(encoded) + + check decoded.isErr + check "exceeds" in decoded.error.msg diff --git a/tests/storage/blockexchange/protocol/testpresence.nim b/tests/storage/blockexchange/protocol/testpresence.nim index ab50fb0c..3f5db651 100644 --- a/tests/storage/blockexchange/protocol/testpresence.nim +++ b/tests/storage/blockexchange/protocol/testpresence.nim @@ -1,3 +1,5 @@ +import std/options + import pkg/chronos import pkg/storage/blockexchange/protocol/presence @@ -22,11 +24,17 @@ suite "Block presence protobuf messages": check PresenceMessage.init(presence).kind == BlockPresenceType.DontHave test "decodes CID": - check Presence.init(message) .? address == address.some + let p = Presence.init(message) + check p.isSome + check p.get.address == address test "decodes have/donthave": var message = message message.kind = BlockPresenceType.HaveRange - check Presence.init(message) .? have == true.some + let pHave = Presence.init(message) + check pHave.isSome + check pHave.get.have == true message.kind = BlockPresenceType.DontHave - check Presence.init(message) .? have == false.some + let pDontHave = Presence.init(message) + check pDontHave.isSome + check pDontHave.get.have == false diff --git a/tests/storage/blockexchange/protocol/testwireformat.nim b/tests/storage/blockexchange/protocol/testwireformat.nim new file mode 100644 index 00000000..fff11889 --- /dev/null +++ b/tests/storage/blockexchange/protocol/testwireformat.nim @@ -0,0 +1,636 @@ +import pkg/unittest2 +import pkg/libp2p/cid + +import pkg/storage/blocktype +import pkg/storage/blockexchange/protocol/message + +## All Expected_* consts were generated from the minprotobuf encoder and are +## the current wire format. +## DO NOT MODIFY +## DO NOT MODIFY +## DO NOT MODIFY +## DO NOT MODIFY +## DO NOT MODIFY any of the Expected_* consts. If tests are failing after +## touching message.nim, the protobuf library, the custom Cid encoder or anything +## else in the encode path, the wire format has diverged from what +## existing peers expect and they will not be able to decode any messages. + +const Expected_emptyMessage = @[0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8] + +const Expected_wantListEmptyFullFalse = @[0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8] + +const Expected_wantListEmptyFullTrue = @[0x0A'u8, 0x02'u8, 0x10'u8, 0x01'u8] + +const Expected_wantListSingleEntry = @[ + 0x0A'u8, 0x3A'u8, 0x0A'u8, 0x36'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, + 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, + 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, + 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, + 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, + 0x05'u8, 0x10'u8, 0x01'u8, 0x18'u8, 0x00'u8, 0x20'u8, 0x00'u8, 0x28'u8, 0x01'u8, + 0x30'u8, 0x00'u8, 0x38'u8, 0x2A'u8, 0x10'u8, 0x00'u8, +] + +const Expected_wantListMultipleFullTrue = @[ + 0x0A'u8, 0x72'u8, 0x0A'u8, 0x36'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, + 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, + 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, + 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, + 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, + 0x00'u8, 0x10'u8, 0x0A'u8, 0x18'u8, 0x00'u8, 0x20'u8, 0x00'u8, 0x28'u8, 0x00'u8, + 0x30'u8, 0x05'u8, 0x38'u8, 0x01'u8, 0x0A'u8, 0x36'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xBB'u8, 0x10'u8, 0x64'u8, 0x10'u8, 0x32'u8, 0x18'u8, 0x01'u8, 0x20'u8, 0x00'u8, + 0x28'u8, 0x01'u8, 0x30'u8, 0x0A'u8, 0x38'u8, 0x02'u8, 0x10'u8, 0x01'u8, +] + +const Expected_presenceDontHave = @[ + 0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8, 0x22'u8, 0x2E'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xAA'u8, 0x10'u8, 0x07'u8, 0x10'u8, 0x00'u8, 0x20'u8, 0x64'u8, +] + +const Expected_presenceHaveRange = @[ + 0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8, 0x22'u8, 0x3B'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xBB'u8, 0x10'u8, 0x00'u8, 0x10'u8, 0x01'u8, 0x1A'u8, 0x04'u8, 0x08'u8, 0x00'u8, + 0x10'u8, 0x0A'u8, 0x1A'u8, 0x04'u8, 0x08'u8, 0x64'u8, 0x10'u8, 0x32'u8, 0x20'u8, + 0xF4'u8, 0x03'u8, +] + +const Expected_presenceComplete = @[ + 0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8, 0x22'u8, 0x30'u8, 0x0A'u8, 0x29'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xAA'u8, 0x10'u8, 0xE7'u8, 0x07'u8, 0x10'u8, 0x02'u8, 0x20'u8, 0x8F'u8, 0x4E'u8, +] + +const Expected_fullMessage = @[ + 0x0A'u8, 0x3A'u8, 0x0A'u8, 0x36'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, + 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, + 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, + 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, + 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, + 0x01'u8, 0x10'u8, 0x05'u8, 0x18'u8, 0x00'u8, 0x20'u8, 0x00'u8, 0x28'u8, 0x00'u8, + 0x30'u8, 0x02'u8, 0x38'u8, 0x01'u8, 0x10'u8, 0x00'u8, 0x22'u8, 0x34'u8, 0x0A'u8, + 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, + 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, + 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, + 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, + 0x1D'u8, 0x1E'u8, 0xBB'u8, 0x10'u8, 0x0A'u8, 0x10'u8, 0x01'u8, 0x1A'u8, 0x04'u8, + 0x08'u8, 0x05'u8, 0x10'u8, 0x03'u8, 0x20'u8, 0x01'u8, 0x22'u8, 0x2E'u8, 0x0A'u8, + 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, + 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, + 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, + 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, + 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, 0x14'u8, 0x10'u8, 0x02'u8, 0x20'u8, 0x01'u8, +] + +const Expected_largeVarints = @[ + 0x0A'u8, 0x55'u8, 0x0A'u8, 0x51'u8, 0x0A'u8, 0x2D'u8, 0x0A'u8, 0x24'u8, 0x01'u8, + 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, + 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, + 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, + 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, + 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0x1F'u8, 0x10'u8, 0xFF'u8, 0xFF'u8, + 0xFF'u8, 0xFF'u8, 0x07'u8, 0x18'u8, 0x00'u8, 0x20'u8, 0x00'u8, 0x28'u8, 0x00'u8, + 0x30'u8, 0xEF'u8, 0xFD'u8, 0xB6'u8, 0xF5'u8, 0xED'u8, 0xD7'u8, 0xAE'u8, 0xFF'u8, + 0xCA'u8, 0x01'u8, 0x38'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, + 0xFF'u8, 0xFF'u8, 0xFF'u8, 0x01'u8, 0x10'u8, 0x00'u8, +] + +const Expected_allZeroValues = @[ + 0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8, 0x22'u8, 0x34'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xAA'u8, 0x10'u8, 0x00'u8, 0x10'u8, 0x00'u8, 0x1A'u8, 0x04'u8, 0x08'u8, 0x00'u8, + 0x10'u8, 0x00'u8, 0x20'u8, 0x00'u8, +] + +const Expected_wantListFullTrueWithPresences = @[ + 0x0A'u8, 0x3A'u8, 0x0A'u8, 0x36'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, + 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, + 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, + 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, + 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, + 0x03'u8, 0x10'u8, 0x07'u8, 0x18'u8, 0x00'u8, 0x20'u8, 0x00'u8, 0x28'u8, 0x01'u8, + 0x30'u8, 0x04'u8, 0x38'u8, 0x0B'u8, 0x10'u8, 0x01'u8, 0x22'u8, 0x34'u8, 0x0A'u8, + 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, + 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, + 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, + 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, + 0x1D'u8, 0x1E'u8, 0xBB'u8, 0x10'u8, 0x04'u8, 0x10'u8, 0x01'u8, 0x1A'u8, 0x04'u8, + 0x08'u8, 0x00'u8, 0x10'u8, 0x10'u8, 0x20'u8, 0x0B'u8, +] + +const Expected_threeBlockPresences = @[ + 0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8, 0x22'u8, 0x34'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xAA'u8, 0x10'u8, 0x01'u8, 0x10'u8, 0x01'u8, 0x1A'u8, 0x04'u8, 0x08'u8, 0x00'u8, + 0x10'u8, 0x08'u8, 0x20'u8, 0x15'u8, 0x22'u8, 0x2E'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xBB'u8, 0x10'u8, 0x02'u8, 0x10'u8, 0x00'u8, 0x20'u8, 0x16'u8, 0x22'u8, 0x2E'u8, + 0x0A'u8, 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, + 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, + 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, + 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, + 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, 0x03'u8, 0x10'u8, 0x02'u8, 0x20'u8, + 0x17'u8, +] + +const Expected_multipleEntriesCancelTrue = @[ + 0x0A'u8, 0x72'u8, 0x0A'u8, 0x36'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, + 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, + 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, + 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, + 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, + 0x01'u8, 0x10'u8, 0x01'u8, 0x18'u8, 0x01'u8, 0x20'u8, 0x00'u8, 0x28'u8, 0x00'u8, + 0x30'u8, 0x01'u8, 0x38'u8, 0x1F'u8, 0x0A'u8, 0x36'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xBB'u8, 0x10'u8, 0x02'u8, 0x10'u8, 0x02'u8, 0x18'u8, 0x01'u8, 0x20'u8, 0x00'u8, + 0x28'u8, 0x00'u8, 0x30'u8, 0x01'u8, 0x38'u8, 0x20'u8, 0x10'u8, 0x00'u8, +] + +const Expected_presenceDownloadIdMax = @[ + 0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8, 0x22'u8, 0x37'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xAA'u8, 0x10'u8, 0x01'u8, 0x10'u8, 0x00'u8, 0x20'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, + 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0x01'u8, +] + +const Expected_entryRangeCountMax = @[ + 0x0A'u8, 0x43'u8, 0x0A'u8, 0x3F'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, 0x24'u8, 0x01'u8, + 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, + 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, + 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, + 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x10'u8, + 0x01'u8, 0x10'u8, 0x01'u8, 0x18'u8, 0x00'u8, 0x20'u8, 0x00'u8, 0x28'u8, 0x00'u8, + 0x30'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, + 0xFF'u8, 0x01'u8, 0x38'u8, 0x01'u8, 0x10'u8, 0x00'u8, +] + +const Expected_rangeStartNearMax = @[ + 0x0A'u8, 0x02'u8, 0x10'u8, 0x00'u8, 0x22'u8, 0x3D'u8, 0x0A'u8, 0x28'u8, 0x0A'u8, + 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, + 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, + 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, + 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, + 0xAA'u8, 0x10'u8, 0x00'u8, 0x10'u8, 0x01'u8, 0x1A'u8, 0x0D'u8, 0x08'u8, 0xFE'u8, + 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0xFF'u8, 0x01'u8, + 0x10'u8, 0x01'u8, 0x20'u8, 0x29'u8, +] + +## Deterministic Cids for stable byte fixtures. +## CIDv1, raw codec (0x55), sha2-256 (0x12, 0x20) + 32 fixed bytes. +proc fixedCid(seed: byte): Cid = + var bytes = @[ + 0x01.byte, 0x55, 0x12, 0x20, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, + 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, + 0x17, 0x18, 0x19, 0x1A, 0x1B, 0x1C, 0x1D, 0x1E, 0x1F, + ] + bytes[^1] = seed + Cid.init(bytes).expect("valid Cid bytes") + +let cidA = fixedCid(0xAA) +let cidB = fixedCid(0xBB) + +suite "Block exchange Message wire-format contract": + test "Should encode an empty Message": + let msg = Message() + check msg.encode() == Expected_emptyMessage + + test "Should encode a WantList with no entries": + let msg1 = Message(wantList: WantList(entries: @[], full: false)) + check msg1.encode() == Expected_wantListEmptyFullFalse + + let msg2 = Message(wantList: WantList(entries: @[], full: true)) + check msg2.encode() == Expected_wantListEmptyFullTrue + + test "Should encode a WantList containing a single entry": + let msg = Message( + wantList: WantList( + entries: @[ + WantListEntry( + address: BlockAddress(treeCid: cidA, index: 5), + priority: 1, + cancel: false, + wantType: WantType.WantHave, + sendDontHave: true, + rangeCount: 0, + downloadId: 42, + ) + ] + ) + ) + check msg.encode() == Expected_wantListSingleEntry + + test "Should encode a WantList containing multiple entries": + let msg = Message( + wantList: WantList( + entries: @[ + WantListEntry( + address: BlockAddress(treeCid: cidA, index: 0), + priority: 10, + cancel: false, + wantType: WantType.WantHave, + sendDontHave: false, + rangeCount: 5, + downloadId: 1, + ), + WantListEntry( + address: BlockAddress(treeCid: cidB, index: 100), + priority: 50, + cancel: true, + wantType: WantType.WantHave, + sendDontHave: true, + rangeCount: 10, + downloadId: 2, + ), + ], + full: true, + ) + ) + check msg.encode() == Expected_wantListMultipleFullTrue + + test "Should encode a BlockPresence with DontHave kind": + let msg = Message( + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidA, index: 7), + kind: BlockPresenceType.DontHave, + ranges: @[], + downloadId: 100, + ) + ] + ) + check msg.encode() == Expected_presenceDontHave + + test "Should encode a BlockPresence with HaveRange kind and multiple ranges": + let msg = Message( + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidB, index: 0), + kind: BlockPresenceType.HaveRange, + ranges: @[ + IndexRange(start: 0'u64, count: 10'u64), + IndexRange(start: 100'u64, count: 50'u64), + ], + downloadId: 500, + ) + ] + ) + check msg.encode() == Expected_presenceHaveRange + + test "Should encode a BlockPresence with Complete kind": + let msg = Message( + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidA, index: 999), + kind: BlockPresenceType.Complete, + ranges: @[], + downloadId: 9999, + ) + ] + ) + check msg.encode() == Expected_presenceComplete + + test "Should encode a Message combining WantList and multiple BlockPresences": + let msg = Message( + wantList: WantList( + entries: @[ + WantListEntry( + address: BlockAddress(treeCid: cidA, index: 1), + priority: 5, + cancel: false, + wantType: WantType.WantHave, + sendDontHave: false, + rangeCount: 2, + downloadId: 1, + ) + ], + full: false, + ), + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidB, index: 10), + kind: BlockPresenceType.HaveRange, + ranges: @[IndexRange(start: 5'u64, count: 3'u64)], + downloadId: 1, + ), + BlockPresence( + address: BlockAddress(treeCid: cidA, index: 20), + kind: BlockPresenceType.Complete, + ranges: @[], + downloadId: 1, + ), + ], + ) + check msg.encode() == Expected_fullMessage + + test "Should encode a Message with maximum-width varint values": + let msg = Message( + wantList: WantList( + entries: @[ + WantListEntry( + address: BlockAddress(treeCid: cidA, index: 0xFFFFFFFFFF'u64), + priority: int32(0x7FFFFFFF), + cancel: false, + wantType: WantType.WantHave, + sendDontHave: false, + rangeCount: 0xCAFEBABE_DEADBEEF'u64, + downloadId: 0xFFFFFFFFFFFFFFFF'u64, + ) + ] + ) + ) + check msg.encode() == Expected_largeVarints + + test "Should encode a Message with all-zero default values": + let msg = Message( + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidA, index: 0), + kind: BlockPresenceType.DontHave, + ranges: @[IndexRange(start: 0'u64, count: 0'u64)], + downloadId: 0, + ) + ] + ) + check msg.encode() == Expected_allZeroValues + + test "Should round-trip every captured byte fixture (decode + re-encode)": + let fixtures = @[ + ("emptyMessage", Expected_emptyMessage), + ("wantListEmptyFullFalse", Expected_wantListEmptyFullFalse), + ("wantListEmptyFullTrue", Expected_wantListEmptyFullTrue), + ("wantListSingleEntry", Expected_wantListSingleEntry), + ("wantListMultipleFullTrue", Expected_wantListMultipleFullTrue), + ("presenceDontHave", Expected_presenceDontHave), + ("presenceHaveRange", Expected_presenceHaveRange), + ("presenceComplete", Expected_presenceComplete), + ("fullMessage", Expected_fullMessage), + ("largeVarints", Expected_largeVarints), + ("allZeroValues", Expected_allZeroValues), + ("wantListFullTrueWithPresences", Expected_wantListFullTrueWithPresences), + ("threeBlockPresences", Expected_threeBlockPresences), + ("multipleEntriesCancelTrue", Expected_multipleEntriesCancelTrue), + ("presenceDownloadIdMax", Expected_presenceDownloadIdMax), + ("entryRangeCountMax", Expected_entryRangeCountMax), + ("rangeStartNearMax", Expected_rangeStartNearMax), + ] + for (name, expected) in fixtures: + checkpoint("fixture: " & name) + let decoded = Message.decode(expected) + check decoded.isOk + check decoded.get.encode() == expected + + test "Should encode a WantList combined with BlockPresences": + let msg = Message( + wantList: WantList( + entries: @[ + WantListEntry( + address: BlockAddress(treeCid: cidA, index: 3), + priority: 7, + cancel: false, + wantType: WantType.WantHave, + sendDontHave: true, + rangeCount: 4, + downloadId: 11, + ) + ], + full: true, + ), + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidB, index: 4), + kind: BlockPresenceType.HaveRange, + ranges: @[IndexRange(start: 0'u64, count: 16'u64)], + downloadId: 11, + ) + ], + ) + check msg.encode() == Expected_wantListFullTrueWithPresences + + test "Should encode a Message containing three BlockPresences": + let msg = Message( + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidA, index: 1), + kind: BlockPresenceType.HaveRange, + ranges: @[IndexRange(start: 0'u64, count: 8'u64)], + downloadId: 21, + ), + BlockPresence( + address: BlockAddress(treeCid: cidB, index: 2), + kind: BlockPresenceType.DontHave, + ranges: @[], + downloadId: 22, + ), + BlockPresence( + address: BlockAddress(treeCid: cidA, index: 3), + kind: BlockPresenceType.Complete, + ranges: @[], + downloadId: 23, + ), + ] + ) + check msg.encode() == Expected_threeBlockPresences + + test "Should encode a WantList where multiple entries have cancel=true": + let msg = Message( + wantList: WantList( + entries: @[ + WantListEntry( + address: BlockAddress(treeCid: cidA, index: 1), + priority: 1, + cancel: true, + wantType: WantType.WantHave, + sendDontHave: false, + rangeCount: 1, + downloadId: 31, + ), + WantListEntry( + address: BlockAddress(treeCid: cidB, index: 2), + priority: 2, + cancel: true, + wantType: WantType.WantHave, + sendDontHave: false, + rangeCount: 1, + downloadId: 32, + ), + ], + full: false, + ) + ) + check msg.encode() == Expected_multipleEntriesCancelTrue + + test "Should encode a BlockPresence with downloadId at uint64 maximum": + let msg = Message( + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidA, index: 1), + kind: BlockPresenceType.DontHave, + ranges: @[], + downloadId: 0xFFFFFFFFFFFFFFFF'u64, + ) + ] + ) + check msg.encode() == Expected_presenceDownloadIdMax + + test "Should encode a WantListEntry with rangeCount at uint64 maximum": + let msg = Message( + wantList: WantList( + entries: @[ + WantListEntry( + address: BlockAddress(treeCid: cidA, index: 1), + priority: 1, + cancel: false, + wantType: WantType.WantHave, + sendDontHave: false, + rangeCount: 0xFFFFFFFFFFFFFFFF'u64, + downloadId: 1, + ) + ] + ) + ) + check msg.encode() == Expected_entryRangeCountMax + + test "Should encode a IndexRange with start near uint64 maximum and count=1": + let msg = Message( + blockPresences: @[ + BlockPresence( + address: BlockAddress(treeCid: cidA, index: 0), + kind: BlockPresenceType.HaveRange, + ranges: @[IndexRange(start: 0xFFFFFFFFFFFFFFFE'u64, count: 1'u64)], + downloadId: 41, + ) + ] + ) + check msg.encode() == Expected_rangeStartNearMax + +suite "Block exchange Message decode error handling": + test "Should reject empty bytes": + let decoded = Message.decode(@[]) + check decoded.isErr + + test "Should reject random garbage bytes": + let garbage = @[ + 0xFF.byte, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, + 0xFF, 0xFF, 0xFF, + ] + let decoded = Message.decode(garbage) + check decoded.isErr + + test "Should reject truncated bytes": + let truncated = Expected_fullMessage[0 ..< 8] + let decoded = Message.decode(@truncated) + check decoded.isErr + + test "Should reject bytes with a length-prefix exceeding buffer": + # Tag 0x0A = field 1, length-delimited. Length 0xFF claims 255 bytes but + # only 2 follow. + let malformed = @[0x0A.byte, 0xFF, 0x00, 0x00] + let decoded = Message.decode(malformed) + check decoded.isErr + + test "Should reject bytes with an unknown wire type": + # Tag 0x0F = field 1, wire type 7 (reserved/invalid). + let malformed = @[0x0F.byte, 0x00] + let decoded = Message.decode(malformed) + check decoded.isErr + + test "Should not crash on a varint with continuation bits set indefinitely": + # 10 bytes with continuation bit set — exceeds varint max length. + let malformed = + @[0x08.byte, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF] + let decoded = Message.decode(malformed) + check decoded.isErr + + test "Should reject a Message containing an invalid Cid": + var corrupted = @Expected_presenceDontHave + for i in 10 .. 45: + corrupted[i] = 0xCC + let decoded = Message.decode(corrupted) + check decoded.isErr + +# ============================================================================ +# For reference: minprotobuf encoders that produced the Expected_* bytes above. +# Kept as documentation of the wire-format derivation. +# ============================================================================ + +when false: + import pkg/libp2p/protobuf/minprotobuf + + proc write*(pb: var ProtoBuffer, field: int, value: BlockAddress) = + var ipb = initProtoBuffer() + ipb.write(1, value.treeCid.data.buffer) + ipb.write(2, value.index.uint64) + ipb.finish() + pb.write(field, ipb) + + proc write*(pb: var ProtoBuffer, field: int, value: WantListEntry) = + var ipb = initProtoBuffer() + ipb.write(1, value.address) + ipb.write(2, value.priority.uint64) + ipb.write(3, value.cancel.uint) + ipb.write(4, value.wantType.uint) + ipb.write(5, value.sendDontHave.uint) + ipb.write(6, value.rangeCount) + ipb.write(7, value.downloadId) + ipb.finish() + pb.write(field, ipb) + + proc write*(pb: var ProtoBuffer, field: int, value: WantList) = + var ipb = initProtoBuffer() + for v in value.entries: + ipb.write(1, v) + ipb.write(2, value.full.uint) + ipb.finish() + pb.write(field, ipb) + + proc write*(pb: var ProtoBuffer, field: int, value: BlockPresence) = + var ipb = initProtoBuffer() + ipb.write(1, value.address) + ipb.write(2, value.kind.uint) + for r in value.ranges: + var rangePb = initProtoBuffer() + rangePb.write(1, r.start) + rangePb.write(2, r.count) + rangePb.finish() + ipb.write(3, rangePb) + ipb.write(4, value.downloadId) + ipb.finish() + pb.write(field, ipb) + + proc protobufEncode*(value: Message): seq[byte] = + var ipb = initProtoBuffer() + ipb.write(1, value.wantList) + for v in value.blockPresences: + ipb.write(4, v) + ipb.finish() + ipb.buffer diff --git a/tests/storage/blockexchange/testdownloadmanager.nim b/tests/storage/blockexchange/testdownloadmanager.nim index 7565a178..619a39fc 100644 --- a/tests/storage/blockexchange/testdownloadmanager.nim +++ b/tests/storage/blockexchange/testdownloadmanager.nim @@ -542,7 +542,7 @@ suite "DownloadManager - Retry Management": treeCid = md.manifest.treeCid for i in 0'u64 ..< 5: - let address = BlockAddress(treeCid: treeCid, index: i.int) + let address = BlockAddress(treeCid: treeCid, index: i) discard download.getWantHandle(address) let addresses = download.getBlockAddressesForRange(0, 10) @@ -690,7 +690,10 @@ suite "DownloadContext - Windowed Presence": md = testManifestDesc(Cid.example, 65536, 100000) ctx = DownloadContext.new(DownloadDesc(md: md, count: 100000)) peerId = PeerId.example - ranges = @[(start: 0'u64, count: 400'u64), (start: 2000'u64, count: 500'u64)] + ranges = @[ + IndexRange(start: 0'u64, count: 400'u64), + IndexRange(start: 2000'u64, count: 500'u64), + ] discard ctx.swarm.addPeer(peerId, BlockAvailability.fromRanges(ranges)) @@ -714,7 +717,7 @@ suite "DownloadContext - Windowed Presence": md = testManifestDesc(Cid.example, 65536, 100000) ctx = DownloadContext.new(DownloadDesc(md: md, count: 100000)) peerId = PeerId.example - ranges = @[(start: 0'u64, count: 1000'u64)] + ranges = @[IndexRange(start: 0'u64, count: 1000'u64)] discard ctx.swarm.addPeer(peerId, BlockAvailability.fromRanges(ranges)) ctx.scheduler.init(ctx.totalBlocks, 256, WindowSize, Threshold) @@ -800,7 +803,7 @@ suite "DownloadManager - Completion Future": var addresses: seq[BlockAddress] = @[] for i in 0'u64 ..< 10: - let address = BlockAddress(treeCid: treeCid, index: i.int) + let address = BlockAddress(treeCid: treeCid, index: i) discard download.getWantHandle(address) addresses.add(address) @@ -866,7 +869,7 @@ suite "DownloadManager - Completion Future": var addresses: seq[BlockAddress] = @[] for i in 0'u64 ..< 10: - let address = BlockAddress(treeCid: treeCid, index: i.int) + let address = BlockAddress(treeCid: treeCid, index: i) discard download.getWantHandle(address) addresses.add(address) diff --git a/tests/storage/blockexchange/testnetwork.nim b/tests/storage/blockexchange/testnetwork.nim index 3c8039fe..810892cc 100644 --- a/tests/storage/blockexchange/testnetwork.nim +++ b/tests/storage/blockexchange/testnetwork.nim @@ -65,14 +65,15 @@ asyncchecksuite "Network - Handlers": makeWantList(treeCid, blocks.len, 1, true, WantType.WantHave, true, true) let msg = Message(wantlist: wantList) - await buffer.pushData(frameProtobufMessage(protobufEncode(msg))) + await buffer.pushData(frameProtobufMessage(msg.encode())) await done.wait(500.millis) test "Presence Handler": let treeCid = Cid.example - addresses = (0 ..< blocks.len).mapIt(BlockAddress(treeCid: treeCid, index: it)) + addresses = + (0 ..< blocks.len).mapIt(BlockAddress(treeCid: treeCid, index: it.uint64)) proc presenceHandler( peer: PeerId, presence: seq[BlockPresence] @@ -89,7 +90,7 @@ asyncchecksuite "Network - Handlers": blockPresences: addresses.mapIt(BlockPresence(address: it, kind: BlockPresenceType.HaveRange)) ) - await buffer.pushData(frameProtobufMessage(protobufEncode(msg))) + await buffer.pushData(frameProtobufMessage(msg.encode())) await done.wait(500.millis) @@ -130,7 +131,8 @@ asyncchecksuite "Network - Senders": test "Send want list": let treeCid = Cid.example - addresses = (0 ..< blocks.len).mapIt(BlockAddress(treeCid: treeCid, index: it)) + addresses = + (0 ..< blocks.len).mapIt(BlockAddress(treeCid: treeCid, index: it.uint64)) proc wantListHandler(peer: PeerId, wantList: WantList) {.async: (raises: []).} = check wantList.entries.len == 4 @@ -154,7 +156,8 @@ asyncchecksuite "Network - Senders": test "send presence": let treeCid = Cid.example - addresses = (0 ..< blocks.len).mapIt(BlockAddress(treeCid: treeCid, index: it)) + addresses = + (0 ..< blocks.len).mapIt(BlockAddress(treeCid: treeCid, index: it.uint64)) proc presenceHandler( peer: PeerId, precense: seq[BlockPresence] diff --git a/tests/storage/helpers.nim b/tests/storage/helpers.nim index 2fbe050b..376a5748 100644 --- a/tests/storage/helpers.nim +++ b/tests/storage/helpers.nim @@ -61,7 +61,7 @@ proc makeWantList*( WantList( entries: (0 ..< count).mapIt( WantListEntry( - address: BlockAddress(treeCid: treeCid, index: it), + address: BlockAddress(treeCid: treeCid, index: it.uint64), priority: priority.int32, cancel: cancel, wantType: wantType, diff --git a/tests/storage/manifest/testwireformat.nim b/tests/storage/manifest/testwireformat.nim new file mode 100644 index 00000000..d09dd73b --- /dev/null +++ b/tests/storage/manifest/testwireformat.nim @@ -0,0 +1,200 @@ +import std/options + +import pkg/unittest2 +import pkg/libp2p/cid +import pkg/storage/manifest +import pkg/storage/units + +## All Expected_* consts were generated from the minprotobuf encoder and are +## the current manifest wire format. +## DO NOT MODIFY +## DO NOT MODIFY +## DO NOT MODIFY +## DO NOT MODIFY any of the Expected_* consts. If tests are failing after +## touching coders.nim, the protobuf library or anything else in the encode +## path, the wire format has diverged from what existing peers and stored +## manifests expect — both network decoding and manifest CIDs would break. + +const Expected_minimal = @[ + 0x0A'u8, 0x38'u8, 0x08'u8, 0x00'u8, 0x12'u8, 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, + 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, + 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, + 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, + 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x18'u8, 0x80'u8, 0x80'u8, + 0x04'u8, 0x20'u8, 0x80'u8, 0x80'u8, 0x40'u8, 0x28'u8, 0x82'u8, 0x9A'u8, 0x03'u8, + 0x30'u8, 0x12'u8, 0x38'u8, 0x02'u8, +] + +const Expected_withFilenameMimetype = @[ + 0x0A'u8, 0x4E'u8, 0x08'u8, 0x00'u8, 0x12'u8, 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, + 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, + 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, + 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, + 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x18'u8, 0x80'u8, 0x80'u8, + 0x04'u8, 0x20'u8, 0x80'u8, 0x80'u8, 0x40'u8, 0x28'u8, 0x82'u8, 0x9A'u8, 0x03'u8, + 0x30'u8, 0x12'u8, 0x38'u8, 0x02'u8, 0x42'u8, 0x08'u8, 0x74'u8, 0x65'u8, 0x73'u8, + 0x74'u8, 0x2E'u8, 0x74'u8, 0x78'u8, 0x74'u8, 0x4A'u8, 0x0A'u8, 0x74'u8, 0x65'u8, + 0x78'u8, 0x74'u8, 0x2F'u8, 0x70'u8, 0x6C'u8, 0x61'u8, 0x69'u8, 0x6E'u8, +] + +const Expected_onlyFilename = @[ + 0x0A'u8, 0x42'u8, 0x08'u8, 0x00'u8, 0x12'u8, 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, + 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, + 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, + 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, + 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x18'u8, 0x80'u8, 0x80'u8, + 0x04'u8, 0x20'u8, 0x80'u8, 0x80'u8, 0x40'u8, 0x28'u8, 0x82'u8, 0x9A'u8, 0x03'u8, + 0x30'u8, 0x12'u8, 0x38'u8, 0x02'u8, 0x42'u8, 0x08'u8, 0x64'u8, 0x61'u8, 0x74'u8, + 0x61'u8, 0x2E'u8, 0x62'u8, 0x69'u8, 0x6E'u8, +] + +const Expected_largeDataset = @[ + 0x0A'u8, 0x3A'u8, 0x08'u8, 0x00'u8, 0x12'u8, 0x24'u8, 0x01'u8, 0x55'u8, 0x12'u8, + 0x20'u8, 0x00'u8, 0x01'u8, 0x02'u8, 0x03'u8, 0x04'u8, 0x05'u8, 0x06'u8, 0x07'u8, + 0x08'u8, 0x09'u8, 0x0A'u8, 0x0B'u8, 0x0C'u8, 0x0D'u8, 0x0E'u8, 0x0F'u8, 0x10'u8, + 0x11'u8, 0x12'u8, 0x13'u8, 0x14'u8, 0x15'u8, 0x16'u8, 0x17'u8, 0x18'u8, 0x19'u8, + 0x1A'u8, 0x1B'u8, 0x1C'u8, 0x1D'u8, 0x1E'u8, 0xAA'u8, 0x18'u8, 0x80'u8, 0x80'u8, + 0x04'u8, 0x20'u8, 0x80'u8, 0x80'u8, 0x80'u8, 0x80'u8, 0x14'u8, 0x28'u8, 0x82'u8, + 0x9A'u8, 0x03'u8, 0x30'u8, 0x12'u8, 0x38'u8, 0x02'u8, +] + +proc fixedCid(seed: byte): Cid = + var bytes = @[ + 0x01.byte, 0x55, 0x12, 0x20, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, + 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, + 0x17, 0x18, 0x19, 0x1A, 0x1B, 0x1C, 0x1D, 0x1E, 0x1F, + ] + bytes[^1] = seed + Cid.init(bytes).expect("valid Cid bytes") + +let cidA = fixedCid(0xAA) + +suite "Manifest wire-format contract": + test "Should encode a minimal manifest": + let m = Manifest.new( + treeCid = cidA, blockSize = 65536.NBytes, datasetSize = 1048576.NBytes + ) + let encoded = m.encode() + check encoded.isOk + check encoded.get == Expected_minimal + + test "Should encode a manifest with filename and mimetype": + let m = Manifest.new( + treeCid = cidA, + blockSize = 65536.NBytes, + datasetSize = 1048576.NBytes, + filename = "test.txt".some, + mimetype = "text/plain".some, + ) + let encoded = m.encode() + check encoded.isOk + check encoded.get == Expected_withFilenameMimetype + + test "Should encode a manifest with only a filename": + let m = Manifest.new( + treeCid = cidA, + blockSize = 65536.NBytes, + datasetSize = 1048576.NBytes, + filename = "data.bin".some, + ) + let encoded = m.encode() + check encoded.isOk + check encoded.get == Expected_onlyFilename + + test "Should encode a manifest with a large dataset size": + let m = Manifest.new( + treeCid = cidA, + blockSize = 65536.NBytes, + datasetSize = (5'u64 * 1024 * 1024 * 1024).NBytes, + ) + let encoded = m.encode() + check encoded.isOk + check encoded.get == Expected_largeDataset + + test "Should round-trip every captured byte fixture (decode + re-encode)": + let fixtures = [ + ("minimal", Expected_minimal), + ("withFilenameMimetype", Expected_withFilenameMimetype), + ("onlyFilename", Expected_onlyFilename), + ("largeDataset", Expected_largeDataset), + ] + for (name, expected) in fixtures: + checkpoint("fixture: " & name) + let decoded = Manifest.decode(expected) + check decoded.isOk + let reEncoded = decoded.get.encode() + check reEncoded.isOk + check reEncoded.get == expected + +suite "Manifest decode error handling": + test "Should reject empty bytes": + check Manifest.decode(newSeq[byte]()).isErr + + test "Should reject random garbage bytes": + check Manifest.decode(@[0xFF'u8, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF]).isErr + + test "Should reject truncated bytes": + check Manifest.decode(Expected_minimal[0 ..< 8]).isErr + + test "Should reject an unsupported manifest version": + # index 3 is the manifestVersion varint value (0x00 -> 0x01). + var corrupted = Expected_minimal + corrupted[3] = 0x01 + check Manifest.decode(corrupted).isErr + + test "Should reject a manifest with a corrupted first byte": + var corrupted = Expected_minimal + corrupted[0] = 0x12 + check Manifest.decode(corrupted).isErr + + test "Should skip an unknown field and decode to the canonical manifest": + var withUnknown = Expected_minimal + withUnknown[1] = withUnknown[1] + 0x02'u8 + withUnknown.add(0x50'u8) + withUnknown.add(0x01'u8) + let decoded = Manifest.decode(withUnknown) + check decoded.isOk + let reEncoded = decoded.get.encode() + check reEncoded.isOk + check reEncoded.get == Expected_minimal + + test "Should decode an empty filename/mimetype string as none": + let m = Manifest.new( + treeCid = cidA, + blockSize = 65536.NBytes, + datasetSize = 1048576.NBytes, + filename = "".some, + mimetype = "".some, + ) + let encoded = m.encode() + check encoded.isOk + let decoded = Manifest.decode(encoded.get) + check decoded.isOk + check decoded.get.filename.isNone + check decoded.get.mimetype.isNone + +# ============================================================================ +# For reference: minprotobuf encoder that produced the Expected_* bytes above. +# Kept as documentation of the wire-format derivation. +# ============================================================================ + +when false: + import pkg/libp2p/protobuf/minprotobuf + + proc legacyEncode(manifest: Manifest): seq[byte] = + var pbNode = initProtoBuffer() + var header = initProtoBuffer() + header.write(1, manifest.manifestVersion) + header.write(2, manifest.treeCid.data.buffer) + header.write(3, manifest.blockSize.uint32) + header.write(4, manifest.datasetSize.uint64) + header.write(5, manifest.codec.uint32) + header.write(6, manifest.hcodec.uint32) + header.write(7, manifest.version.uint32) + if manifest.filename.isSome: + header.write(8, manifest.filename.get()) + if manifest.mimetype.isSome: + header.write(9, manifest.mimetype.get()) + pbNode.write(1, header) + pbNode.finish() + pbNode.buffer