mirror of
https://github.com/logos-messaging/nim-sds.git
synced 2026-07-22 23:59:27 +00:00
fix: keep SDS causalHistory wire-compatible with pre-SDS-R nodes (#86)
This commit is contained in:
parent
3aab2fbee0
commit
01fdcfcff9
4
.github/workflows/ci.yml
vendored
4
.github/workflows/ci.yml
vendored
@ -145,7 +145,9 @@ jobs:
|
||||
|
||||
- uses: maxim-lobanov/setup-xcode@v1
|
||||
with:
|
||||
xcode-version: '16.2'
|
||||
# 'latest-stable' tracks whatever the macos-latest runner ships, so
|
||||
# this survives Xcode rotations (the runner dropped 16.2 for 26.x).
|
||||
xcode-version: 'latest-stable'
|
||||
|
||||
- uses: jiro4989/setup-nim-action@v2
|
||||
with:
|
||||
|
||||
@ -13,10 +13,16 @@
|
||||
import std/[base64, json]
|
||||
import ffi
|
||||
import sds
|
||||
import ./events/[
|
||||
json_message_ready_event, json_message_sent_event, json_missing_dependencies_event,
|
||||
json_periodic_sync_event, json_repair_ready_event,
|
||||
]
|
||||
import
|
||||
./events/[
|
||||
json_message_ready_event, json_message_sent_event, json_missing_dependencies_event,
|
||||
json_periodic_sync_event, json_repair_ready_event,
|
||||
]
|
||||
|
||||
# Frees memory the retrieval-hint provider allocated with malloc (the Go
|
||||
# binding's C.CBytes). Must match that allocator — Nim's deallocShared here
|
||||
# corrupts Nim's own heap.
|
||||
proc c_free(p: pointer) {.importc: "free", header: "<stdlib.h>".}
|
||||
|
||||
# Bootstrap (pragmas, linker flags, libsdsNimMain, initializeLibrary) plus the
|
||||
# `sds_set_event_callback(ctx, callback, userData)` C export. It also declares
|
||||
@ -114,7 +120,10 @@ proc sdsCreate*(
|
||||
if not hint.isNil() and hintLen > 0:
|
||||
var hintBytes = newSeq[byte](hintLen)
|
||||
copyMem(addr hintBytes[0], hint, hintLen)
|
||||
deallocShared(hint)
|
||||
# The provider allocates `hint` with malloc (the Go binding's C.CBytes)
|
||||
# and hands us ownership; free it with libc free. Nim's deallocShared here
|
||||
# corrupts Nim's allocator (SIGSEGV in deallocBigChunk/avltree).
|
||||
c_free(hint)
|
||||
return hintBytes
|
||||
return @[]
|
||||
|
||||
@ -219,9 +228,7 @@ proc sds_set_retrieval_hint_provider(
|
||||
try:
|
||||
ffi_context.sendRequestToFFIThread(
|
||||
ctx,
|
||||
SdsSetHintReq.ffiNewReq(
|
||||
sdsNoopCallback, nil, cast[pointer](callback), userData
|
||||
),
|
||||
SdsSetHintReq.ffiNewReq(sdsNoopCallback, nil, cast[pointer](callback), userData),
|
||||
)
|
||||
except Exception as exc:
|
||||
Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg)
|
||||
|
||||
@ -69,6 +69,7 @@ task test, "Run the test suite":
|
||||
exec "nim c -r --outdir:build tests/test_reliability.nim"
|
||||
exec "nim c -r --outdir:build tests/test_persistence.nim"
|
||||
exec "nim c -r --outdir:build tests/test_snapshot_codec.nim"
|
||||
exec "nim c -r --outdir:build tests/test_wire_compat.nim"
|
||||
|
||||
task libsdsDynamicWindows, "Generate bindings":
|
||||
let outLibNameAndExt = "libsds.dll"
|
||||
|
||||
@ -15,6 +15,7 @@
|
||||
|
||||
{.push raises: [].}
|
||||
|
||||
import std/tables
|
||||
import endians
|
||||
import protobuf_serialization
|
||||
import protobuf_serialization/pkg/results
|
||||
@ -34,11 +35,17 @@ type
|
||||
SdsMessagePB* {.proto3.} = object
|
||||
messageId* {.fieldNumber: 1.}: Opt[seq[byte]]
|
||||
lamportTimestamp* {.fieldNumber: 2, pint.}: Opt[int64]
|
||||
causalHistory* {.fieldNumber: 3.}: seq[HistoryEntryPB]
|
||||
# Field 3 is a repeated string/bytes of message IDs, wire-compatible with
|
||||
# pre-SDS-R nodes (v0.2.x/v0.3.x). Richer per-entry SDS-R metadata
|
||||
# (senderId, retrievalHint) rides in the additive field 8, keyed by message
|
||||
# ID; older nodes skip it as an unknown field while still reading the causal
|
||||
# history from field 3.
|
||||
causalHistory* {.fieldNumber: 3.}: seq[seq[byte]]
|
||||
channelId* {.fieldNumber: 4.}: Opt[seq[byte]]
|
||||
content* {.fieldNumber: 5.}: Opt[seq[byte]]
|
||||
bloomFilter* {.fieldNumber: 6.}: Opt[seq[byte]]
|
||||
senderId* {.fieldNumber: 7.}: Opt[seq[byte]]
|
||||
causalHistoryMeta* {.fieldNumber: 8.}: seq[HistoryEntryPB]
|
||||
repairRequest* {.fieldNumber: 13.}: seq[HistoryEntryPB]
|
||||
|
||||
BloomFilterPB {.proto3.} = object
|
||||
@ -98,15 +105,26 @@ func toPB*(m: SdsMessage): SdsMessagePB =
|
||||
senderId: optBytes(m.senderId.string.toBytes),
|
||||
)
|
||||
for e in m.causalHistory:
|
||||
pb.causalHistory.add(e.toPB)
|
||||
pb.causalHistory.add(e.messageId.toBytes)
|
||||
if e.retrievalHint.len > 0 or e.senderId.len > 0:
|
||||
pb.causalHistoryMeta.add(e.toPB)
|
||||
for e in m.repairRequest:
|
||||
pb.repairRequest.add(e.toPB)
|
||||
return pb
|
||||
|
||||
func fromPB*(pb: SdsMessagePB): SdsMessage =
|
||||
# Rebuild causal history from field 3 (authoritative ID list) and enrich each
|
||||
# entry with the additive field-8 metadata (senderId/retrievalHint) keyed by
|
||||
# message ID. Messages from pre-SDS-R nodes carry no field 8, so entries
|
||||
# decode to bare IDs.
|
||||
var meta = initTable[SdsMessageID, HistoryEntry]()
|
||||
for e in pb.causalHistoryMeta:
|
||||
let he = e.fromPB
|
||||
meta[he.messageId] = he
|
||||
var causal: seq[HistoryEntry]
|
||||
for e in pb.causalHistory:
|
||||
causal.add(e.fromPB)
|
||||
for idBytes in pb.causalHistory:
|
||||
let id = idBytes.toStr
|
||||
causal.add(meta.getOrDefault(id, HistoryEntry.init(id)))
|
||||
var repair: seq[HistoryEntry]
|
||||
for e in pb.repairRequest:
|
||||
repair.add(e.fromPB)
|
||||
|
||||
95
tests/test_wire_compat.nim
Normal file
95
tests/test_wire_compat.nim
Normal file
@ -0,0 +1,95 @@
|
||||
## Verifies the SDS message wire stays backward-compatible with pre-SDS-R nodes
|
||||
## (v0.2.x/v0.3.x): causalHistory is a repeated string of message IDs in field 3,
|
||||
## and the richer per-entry metadata (senderId, retrievalHint) rides in the
|
||||
## additive field 8, which older nodes skip.
|
||||
|
||||
import std/unittest
|
||||
import protobuf_serialization
|
||||
import protobuf_serialization/pkg/results
|
||||
import ../sds/protobuf
|
||||
import ../sds/types/[sds_message, history_entry, sds_message_id]
|
||||
|
||||
converter toParticipantID(s: string): SdsParticipantID =
|
||||
s.SdsParticipantID
|
||||
|
||||
func toStr(b: seq[byte]): string =
|
||||
result = newString(b.len)
|
||||
if b.len > 0:
|
||||
copyMem(addr result[0], unsafeAddr b[0], b.len)
|
||||
|
||||
func toBytes(s: string): seq[byte] =
|
||||
result = newSeq[byte](s.len)
|
||||
if s.len > 0:
|
||||
copyMem(addr result[0], unsafeAddr s[0], s.len)
|
||||
|
||||
# How a pre-SDS-R node sees the message: field 3 is a repeated string of message
|
||||
# IDs; it knows nothing of the additive field 8.
|
||||
type OldSdsMessagePB {.proto3.} = object
|
||||
messageId {.fieldNumber: 1.}: Opt[seq[byte]]
|
||||
causalHistory {.fieldNumber: 3.}: seq[seq[byte]]
|
||||
channelId {.fieldNumber: 4.}: Opt[seq[byte]]
|
||||
content {.fieldNumber: 5.}: Opt[seq[byte]]
|
||||
|
||||
suite "SDS v0.4 wire backward-compatibility":
|
||||
test "old node reads causal history IDs from a v0.4-encoded message":
|
||||
let msg = SdsMessage.init(
|
||||
messageId = "0xdeadbeef",
|
||||
lamportTimestamp = 7,
|
||||
causalHistory = @[
|
||||
HistoryEntry.init("0xaaa", @[byte 1, 2, 3], "0xsender".SdsParticipantID),
|
||||
HistoryEntry.init("0xbbb"),
|
||||
],
|
||||
channelId = "0xchannel",
|
||||
content = @[byte 9, 9, 9],
|
||||
bloomFilter = @[],
|
||||
senderId = "0xmsgsender".SdsParticipantID,
|
||||
repairRequest = @[],
|
||||
)
|
||||
|
||||
let wire = serializeMessage(msg).get()
|
||||
|
||||
# An old node recovers the EXACT message IDs from field 3 and never sees the
|
||||
# senderId/hint that ride in field 8.
|
||||
let old = Protobuf.decode(wire, OldSdsMessagePB)
|
||||
check old.causalHistory.len == 2
|
||||
check old.causalHistory[0].toStr == "0xaaa"
|
||||
check old.causalHistory[1].toStr == "0xbbb"
|
||||
check old.channelId.get(@[]).toStr == "0xchannel"
|
||||
check old.content.get(@[]) == @[byte 9, 9, 9]
|
||||
|
||||
test "v0.4 round-trips message IDs, hints and senderIds":
|
||||
let msg = SdsMessage.init(
|
||||
messageId = "0xfeed",
|
||||
lamportTimestamp = 3,
|
||||
causalHistory = @[
|
||||
HistoryEntry.init("0xaaa", @[byte 1, 2, 3], "0xsender".SdsParticipantID),
|
||||
HistoryEntry.init("0xbbb"),
|
||||
],
|
||||
channelId = "0xchannel",
|
||||
content = @[byte 1],
|
||||
bloomFilter = @[],
|
||||
)
|
||||
|
||||
let decoded = deserializeMessage(serializeMessage(msg).get()).get()
|
||||
check decoded.causalHistory.len == 2
|
||||
check decoded.causalHistory[0].messageId == "0xaaa"
|
||||
check decoded.causalHistory[0].retrievalHint == @[byte 1, 2, 3]
|
||||
check decoded.causalHistory[0].senderId.string == "0xsender"
|
||||
check decoded.causalHistory[1].messageId == "0xbbb"
|
||||
check decoded.causalHistory[1].retrievalHint.len == 0
|
||||
check decoded.causalHistory[1].senderId.string.len == 0
|
||||
|
||||
test "v0.4 decodes a legacy message (field 3 IDs only, no field 8)":
|
||||
# Build a message the way a pre-SDS-R node would: field 3 = bare ID strings.
|
||||
let legacy = OldSdsMessagePB(
|
||||
messageId: Opt.some("0xlegacy".toBytes),
|
||||
causalHistory: @["0xaaa".toBytes, "0xbbb".toBytes],
|
||||
channelId: Opt.some("0xchannel".toBytes),
|
||||
content: Opt.some(@[byte 5]),
|
||||
)
|
||||
let decoded = deserializeMessage(Protobuf.encode(legacy)).get()
|
||||
check decoded.causalHistory.len == 2
|
||||
check decoded.causalHistory[0].messageId == "0xaaa"
|
||||
check decoded.causalHistory[0].retrievalHint.len == 0
|
||||
check decoded.causalHistory[0].senderId.string.len == 0
|
||||
check decoded.causalHistory[1].messageId == "0xbbb"
|
||||
Loading…
x
Reference in New Issue
Block a user