logos-messaging/nim-sds
1
0
Fork 0
mirror of https://github.com/logos-messaging/nim-sds.git synced 2026-07-04 14:59:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge d9d70cc2fe7d82bd981b745dbc986353dfc626b9 into b12f5ee07c5b764303b51fb948b32a4ade1de3b5

Browse Source
This commit is contained in:
Ivan FB 2026-06-24 19:25:27 +00:00 committed by GitHub
commit cf3ff64c90
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 832 additions and 1039 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/ci-nix.yml vendored
View File

@ -5,7 +5,7 @@ permissions:
checks: write
on:
pull_request:
branches: [master]
branches: [master, release/*]
jobs:
build:

4
.github/workflows/ci.yml vendored
View File

@ -5,9 +5,9 @@ permissions:
checks: write
on:
pull_request:
branches: [master]
branches: [master, release/*]
push:
branches: [master]
branches: [master, release/*]
workflow_dispatch:
jobs:

6
library/events/json_base_event.nim
View File

@ -1,6 +0,0 @@
type JsonEvent* = ref object of RootObj # https://rfc.vac.dev/spec/36/#jsonsignal-type
eventType* {.requiresInit.}: string
method `$`*(jsonEvent: JsonEvent): string {.base.} =
discard
# All events should implement this

16
library/events/json_message_ready_event.nim
View File

@ -1,16 +0,0 @@
import std/json
import ./json_base_event, sds/[message]
type JsonMessageReadyEvent* = ref object of JsonEvent
messageId*: SdsMessageID
channelId*: SdsChannelID
proc new*(
T: type JsonMessageReadyEvent, messageId: SdsMessageID, channelId: SdsChannelID
): T =
return JsonMessageReadyEvent(
eventType: "message_ready", messageId: messageId, channelId: channelId
)
method `$`*(jsonMessageReady: JsonMessageReadyEvent): string =
$(%*jsonMessageReady)

16
library/events/json_message_sent_event.nim
View File

@ -1,16 +0,0 @@
import std/json
import ./json_base_event, sds/[message]
type JsonMessageSentEvent* = ref object of JsonEvent
messageId*: SdsMessageID
channelId*: SdsChannelID
proc new*(
T: type JsonMessageSentEvent, messageId: SdsMessageID, channelId: SdsChannelID
): T =
return JsonMessageSentEvent(
eventType: "message_sent", messageId: messageId, channelId: channelId
)
method `$`*(jsonMessageSent: JsonMessageSentEvent): string =
$(%*jsonMessageSent)

34
library/events/json_missing_dependencies_event.nim
View File

@ -1,34 +0,0 @@
import std/json
import ./json_base_event, sds/[message], std/base64
type JsonMissingDependenciesEvent* = ref object of JsonEvent
messageId*: SdsMessageID
missingDeps*: seq[HistoryEntry]
channelId*: SdsChannelID
proc new*(
T: type JsonMissingDependenciesEvent,
messageId: SdsMessageID,
missingDeps: seq[HistoryEntry],
channelId: SdsChannelID,
): T =
return JsonMissingDependenciesEvent(
eventType: "missing_dependencies",
messageId: messageId,
missingDeps: missingDeps,
channelId: channelId,
)
method `$`*(jsonMissingDependencies: JsonMissingDependenciesEvent): string =
var node = newJObject()
node["eventType"] = %*jsonMissingDependencies.eventType
node["messageId"] = %*jsonMissingDependencies.messageId
node["channelId"] = %*jsonMissingDependencies.channelId
var missingDepsNode = newJArray()
for dep in jsonMissingDependencies.missingDeps:
var depNode = newJObject()
depNode["messageId"] = %*dep.messageId
depNode["retrievalHint"] = %*encode(dep.retrievalHint)
missingDepsNode.add(depNode)
node["missingDeps"] = missingDepsNode
$node

10
library/events/json_periodic_sync_event.nim
View File

@ -1,10 +0,0 @@
import std/json
import ./json_base_event
type JsonPeriodicSyncEvent* = ref object of JsonEvent
proc new*(T: type JsonPeriodicSyncEvent): T =
return JsonPeriodicSyncEvent(eventType: "periodic_sync")
method `$`*(jsonPeriodicSync: JsonPeriodicSyncEvent): string =
$(%*jsonPeriodicSync)

20
library/events/json_repair_ready_event.nim
View File

@ -1,20 +0,0 @@
import std/[json, base64]
import ./json_base_event, sds/[message]
type JsonRepairReadyEvent* = ref object of JsonEvent
channelId*: SdsChannelID
message*: seq[byte]
proc new*(
T: type JsonRepairReadyEvent, message: seq[byte], channelId: SdsChannelID
): T =
return JsonRepairReadyEvent(
eventType: "repair_ready", message: message, channelId: channelId
)
method `$`*(jsonRepairReady: JsonRepairReadyEvent): string =
var node = newJObject()
node["eventType"] = %*jsonRepairReady.eventType
node["channelId"] = %*jsonRepairReady.channelId
node["message"] = %*encode(jsonRepairReady.message)
$node

71
library/libsds.h
View File

@ -1,8 +1,11 @@
// Generated manually and inspired by the one generated by the Nim Compiler.
// In order to see the header file generated by Nim just run `make libsds`
// from the root repo folder and the header should be created in
// nimcache/release/libsds/libsds.h
// C API for libsds, built on the nim-ffi 0.2 framework.
//
// Parameters and results are marshalled as CBOR (RFC 8949): each request and
// response struct in library/libsds.nim is a CBOR map keyed by the exact field
// names; binary fields (message bytes, retrieval hints) are CBOR byte strings.
// Requests are passed in as a length-delimited CBOR buffer (reqCbor/reqCborLen)
// and results are delivered to the callback as a CBOR buffer (msg/len).
#ifndef __libsds__
#define __libsds__
@ -18,51 +21,61 @@
extern "C" {
#endif
// Result/event callback. On RET_OK, `msg` points to the CBOR-encoded payload of
// length `len` (an empty/void result is the single CBOR-null byte 0xf6). On
// RET_ERR, `msg` is the raw UTF-8 error string (NOT CBOR).
typedef void (*SdsCallBack) (int callerRet, const char* msg, size_t len, void* userData);
// Synchronous provider invoked by SDS-R to fetch a retrieval hint for a
// message id. The implementation allocates `*hint` (and sets `*hintLen`) with
// libc malloc; the library takes ownership and frees it with libc free.
typedef void (*SdsRetrievalHintProvider) (const char* messageId, char** hint, size_t* hintLen, void* userData);
// --- Core API Functions ---
void* SdsNewReliabilityManager(SdsCallBack callback, void* userData);
// Create a context + ReliabilityManager. reqCbor: CBOR of
// {"config":{"participantId":"..."}} (empty participantId disables SDS-R).
// Returns the context handle, or NULL on failure. The callback also fires on
// async completion.
void* sds_create(const uint8_t* reqCbor, size_t reqCborLen, SdsCallBack callback, void* userData);
void SdsSetEventCallback(void* ctx, SdsCallBack callback, void* userData);
// Register an event listener for `eventName` (message_ready, message_sent,
// missing_dependencies, periodic_sync, repair_ready). The callback receives a
// CBOR EventEnvelope {"eventType":"<name>","payload":{...}}. Returns a listener
// id (> 0) usable with sds_remove_event_listener, or 0 if the callback is NULL.
uint64_t sds_add_event_listener(void* ctx, const char* eventName, SdsCallBack callback, void* userData);
void SdsSetRetrievalHintProvider(void* ctx, SdsRetrievalHintProvider callback, void* userData);
// Remove a previously registered event listener. Returns RET_OK on success.
int sds_remove_event_listener(void* ctx, uint64_t listenerId);
int SdsCleanupReliabilityManager(void* ctx, SdsCallBack callback, void* userData);
// Register the retrieval-hint provider used by SDS-R.
int sds_set_retrieval_hint_provider(void* ctx, SdsRetrievalHintProvider callback, void* userData);
int SdsResetReliabilityManager(void* ctx, SdsCallBack callback, void* userData);
// reqCbor: CBOR of {"req":{"message":<bytes>,"messageId":"..","channelId":".."}}
// Result CBOR: {"message":<bytes>}
int sds_wrap_outgoing_message(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
int SdsWrapOutgoingMessage(void* ctx,
void* message,
size_t messageLen,
const char* messageId,
const char* channelId,
SdsCallBack callback,
void* userData);
// reqCbor: CBOR of {"req":{"message":<bytes>}}
// Result CBOR: {"message":<bytes>,"channelId":"..","missingDeps":[{"messageId":"..","retrievalHint":<bytes>}]}
int sds_unwrap_received_message(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
int SdsUnwrapReceivedMessage(void* ctx,
void* message,
size_t messageLen,
SdsCallBack callback,
void* userData);
// reqCbor: CBOR of {"req":{"messageIds":["..",".."],"channelId":".."}}
int sds_mark_dependencies_met(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
int SdsMarkDependenciesMet(void* ctx,
char** messageIDs,
size_t count,
const char* channelId,
SdsCallBack callback,
void* userData);
// No request payload — pass reqCbor=NULL, reqCborLen=0.
int sds_reset(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
int SdsStartPeriodicTasks(void* ctx, SdsCallBack callback, void* userData);
// No request payload — pass reqCbor=NULL, reqCborLen=0.
int sds_start_periodic_tasks(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
// Tear down the context created by sds_create.
int sds_destroy(void* ctx);
#ifdef __cplusplus
}
#endif
#endif /* __libsds__ */
#endif /* __libsds__ */

627
library/libsds.nim
View File

@ -1,469 +1,282 @@
import std/[strutils, sequtils, json, base64, locks]
## C-compatible FFI wrapper around the SDS ReliabilityManager.
##
## Built on the `nim-ffi` 0.2 package's high-level macros: `declareLibrary`
## emits the bootstrap plus the per-event listener registry
## (`sds_add_event_listener` / `sds_remove_event_listener`);
## `{.ffiCtor.}`/`{.ffi.}`/`{.ffiDtor.}` generate the C entry points,
## marshalling parameters and return values as CBOR; `{.ffiEvent.}` declares
## the library-initiated events (also CBOR). Exported C names are snake_case
## (`sds_wrap_outgoing_message`, …); see `library/libsds.h`. The Go bindings
## (sds-go-bindings) must match this API.
##
## The one exception is `sds_set_retrieval_hint_provider`: it takes a C
## function pointer, which has no sensible CBOR representation, so it is
## hand-written and dispatched to the worker thread (the pointers travel as
## uint64 through the request channel) to store the provider in a thread-local.
import system/ansi_c
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,
]
# Emit the library bootstrap: the {.exported.}/{.callback.} pragmas, the
# `-fPIC`/soname linker flags, the `libsdsNimMain` import and the
# `initializeLibrary()` proc the exported entry points call on every hop.
declareLibraryBase("sds")
# C callback typedefs (mirrors libsds.h). `SdsCallBack` is structurally the
# nim-ffi `FFICallBack`; the alias keeps the exported signatures readable.
type SdsCallBack* = FFICallBack
# Bootstrap (pragmas, linker flags, libsdsNimMain, initializeLibrary) plus the
# `sds_add_event_listener` / `sds_remove_event_listener` C exports and the
# per-type `ReliabilityManagerFFIPool` used by the hand-written entry point
# below (the ffiCtor/ffiDtor macros declare it too, guarded by `when not
# declared`).
declareLibrary("sds", ReliabilityManager)
type SdsRetrievalHintProvider* = proc(
messageId: cstring, hint: ptr cstring, hintLen: ptr csize_t, userData: pointer
) {.cdecl, gcsafe, raises: [].}
# One pool per library type; the macros that would normally declare it
# (ffiCtor/ffiDtor) are not used here because we hand-write the entry points
# to preserve the exact C ABI, so we declare it explicitly.
var ReliabilityManagerFFIPool: FFIContextPool[ReliabilityManager]
# registerReqFFI inspects each request field's type via `$node`, which only
# handles plain identifiers — a bracketed `SharedSeq[byte]` makes it choke. The
# aliases give the generated request structs non-bracketed field types.
type
SdsSharedBytes = SharedSeq[byte]
SdsSharedCstrs = SharedSeq[cstring]
# The active retrieval-hint provider, stored per worker thread (one thread per
# context). Set by sds_set_retrieval_hint_provider via a dispatched request so
# the write lands on the worker thread, where the manager's hint closure reads
# it during message processing.
var sdsRetrievalHintCb {.threadvar.}: pointer
var sdsRetrievalHintUserData {.threadvar.}: pointer
################################################################################
### Retrieval-hint provider registry
###
### The retrieval-hint provider is a synchronous request/response callback
### (the C side returns bytes inline), so it does not fit the fire-and-forget
### event model. nim-ffi's FFIContext has no slot for it, so we keep a small
### per-context registry here. A fixed array of plain (non-GC) records keeps
### the lookup callable from the {.gcsafe.} hint closure running on the FFI
### thread.
### CBOR-marshalled request/response types
type RetrievalHintSlot = object
ctx: pointer
cb: pointer
userData: pointer
type SdsConfig* {.ffi.} = object
participantId: string ## empty disables SDS-R (see newReliabilityManager)
var retrievalHintSlots: array[MaxFFIContexts, RetrievalHintSlot]
var retrievalHintsLock: Lock
retrievalHintsLock.initLock()
type SdsWrapRequest* {.ffi.} = object
message: seq[byte]
messageId: string
channelId: string
proc setRetrievalHint(ctx: pointer, cb: pointer, userData: pointer) =
withLock retrievalHintsLock:
var free = -1
for i in 0 ..< MaxFFIContexts:
if retrievalHintSlots[i].ctx == ctx:
retrievalHintSlots[i] = RetrievalHintSlot(ctx: ctx, cb: cb, userData: userData)
return
if free < 0 and retrievalHintSlots[i].ctx.isNil:
free = i
if free >= 0:
retrievalHintSlots[free] = RetrievalHintSlot(ctx: ctx, cb: cb, userData: userData)
type SdsWrapResponse* {.ffi.} = object
message: seq[byte]
proc getRetrievalHint(ctx: pointer): tuple[cb: pointer, userData: pointer] {.gcsafe.} =
withLock retrievalHintsLock:
for i in 0 ..< MaxFFIContexts:
if retrievalHintSlots[i].ctx == ctx:
return (retrievalHintSlots[i].cb, retrievalHintSlots[i].userData)
return (nil, nil)
type SdsUnwrapRequest* {.ffi.} = object
message: seq[byte]
proc clearRetrievalHint(ctx: pointer) =
withLock retrievalHintsLock:
for i in 0 ..< MaxFFIContexts:
if retrievalHintSlots[i].ctx == ctx:
retrievalHintSlots[i] = RetrievalHintSlot()
return
# One missing dependency: the message id plus an optional retrieval hint. The
# hint is a raw byte string on the CBOR wire (no base64, unlike the old JSON).
type SdsMissingDep* {.ffi.} = object
messageId: string
retrievalHint: seq[byte]
type SdsUnwrapResponse* {.ffi.} = object
message: seq[byte]
channelId: string
missingDeps: seq[SdsMissingDep]
type SdsMarkDependenciesRequest* {.ffi.} = object
messageIds: seq[string]
channelId: string
################################################################################
### Shared-memory copy helpers
###
### Request payloads carrying binary/pointer data must be deep-copied into
### shared memory on the caller thread, because the FFI thread acks receipt
### before it reads the payload — the caller may free its buffer in between.
### cstring fields are deep-copied by the generated `ffiNewReq`; raw byte and
### `char**` arrays are not, so we copy them here.
### Library-initiated events (CBOR EventEnvelope via {.ffiEvent.})
proc copyToSharedSeqByte(p: pointer, len: int): SharedSeq[byte] =
if p.isNil or len <= 0:
return (cast[ptr UncheckedArray[byte]](nil), 0)
let data = allocShared(len)
copyMem(data, p, len)
return (cast[ptr UncheckedArray[byte]](data), len)
type SdsMessageReadyEvent* {.ffi.} = object
messageId: string
channelId: string
proc copyToSharedSeqCstr(p: pointer, count: int): SharedSeq[cstring] =
if p.isNil or count <= 0:
return (cast[ptr UncheckedArray[cstring]](nil), 0)
let data = cast[ptr UncheckedArray[cstring]](allocShared(sizeof(cstring) * count))
let src = cast[ptr UncheckedArray[cstring]](p)
for i in 0 ..< count:
data[i] = src[i].alloc()
return (data, count)
type SdsMessageSentEvent* {.ffi.} = object
messageId: string
channelId: string
proc freeSharedSeqCstr(s: var SharedSeq[cstring]) =
if not s.data.isNil():
for i in 0 ..< s.len:
if not s.data[i].isNil:
deallocShared(s.data[i])
deallocShared(s.data)
s.len = 0
type SdsMissingDependenciesEvent* {.ffi.} = object
messageId: string
missingDeps: seq[SdsMissingDep]
channelId: string
type SdsPeriodicSyncEvent* {.ffi.} = object
ok: bool ## carries no data; a field keeps the CBOR map non-degenerate
type SdsRepairReadyEvent* {.ffi.} = object
message: seq[byte]
channelId: string
proc onMessageReady*(evt: SdsMessageReadyEvent) {.ffiEvent: "message_ready".}
proc onMessageSent*(evt: SdsMessageSentEvent) {.ffiEvent: "message_sent".}
proc onMissingDependencies*(
evt: SdsMissingDependenciesEvent
) {.ffiEvent: "missing_dependencies".}
proc onPeriodicSync*(evt: SdsPeriodicSyncEvent) {.ffiEvent: "periodic_sync".}
proc onRepairReady*(evt: SdsRepairReadyEvent) {.ffiEvent: "repair_ready".}
proc toMissingDeps(entries: seq[HistoryEntry]): seq[SdsMissingDep] =
var deps = newSeq[SdsMissingDep](entries.len)
for i, entry in entries:
deps[i] =
SdsMissingDep(messageId: entry.messageId, retrievalHint: entry.retrievalHint)
return deps
################################################################################
### Event callbacks
### Constructor — creates the FFI context and the ReliabilityManager.
###
### These build the AppCallbacks closures handed to the ReliabilityManager.
### They run on the FFI worker thread and forward JSON event payloads to the
### C callback registered via SdsSetEventCallback (stored on the context).
### The event closures run on the worker thread and forward CBOR payloads to
### the listeners registered via sds_add_event_listener (the {.ffiEvent.} procs
### read the per-thread event queue, so no context handle is needed here).
proc onMessageReady(ctx: ptr FFIContext[ReliabilityManager]): MessageReadyCallback =
return proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} =
callEventCallback(ctx, "onMessageReady"):
$JsonMessageReadyEvent.new(messageId, channelId)
proc sdsCreate*(
config: SdsConfig
): Future[Result[ReliabilityManager, string]] {.ffiCtor.} =
# The ctor body runs on the (possibly recycled) worker thread. Drop any
# retrieval-hint provider left over from a previous owner of this thread so a
# stale C function pointer is never invoked.
sdsRetrievalHintCb = nil
sdsRetrievalHintUserData = nil
proc onMessageSent(ctx: ptr FFIContext[ReliabilityManager]): MessageSentCallback =
return proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} =
callEventCallback(ctx, "onMessageSent"):
$JsonMessageSentEvent.new(messageId, channelId)
let rm = newReliabilityManager(
participantId = config.participantId.SdsParticipantID
).valueOr:
error "Failed creating reliability manager", error = error
return err("Failed creating reliability manager: " & $error)
proc onMissingDependencies(
ctx: ptr FFIContext[ReliabilityManager]
): MissingDependenciesCallback =
return proc(
let messageReadyCb = proc(
messageId: SdsMessageID, channelId: SdsChannelID
) {.gcsafe.} =
onMessageReady(SdsMessageReadyEvent(messageId: messageId, channelId: channelId))
let messageSentCb = proc(
messageId: SdsMessageID, channelId: SdsChannelID
) {.gcsafe.} =
onMessageSent(SdsMessageSentEvent(messageId: messageId, channelId: channelId))
let missingDependenciesCb = proc(
messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID
) {.gcsafe.} =
callEventCallback(ctx, "onMissingDependencies"):
$JsonMissingDependenciesEvent.new(messageId, missingDeps, channelId)
proc onPeriodicSync(ctx: ptr FFIContext[ReliabilityManager]): PeriodicSyncCallback =
return proc() {.gcsafe.} =
callEventCallback(ctx, "onPeriodicSync"):
$JsonPeriodicSyncEvent.new()
proc onRepairReady(ctx: ptr FFIContext[ReliabilityManager]): RepairReadyCallback =
return proc(message: seq[byte], channelId: SdsChannelID) {.gcsafe.} =
callEventCallback(ctx, "onRepairReady"):
$JsonRepairReadyEvent.new(message, channelId)
proc onRetrievalHint(ctx: ptr FFIContext[ReliabilityManager]): RetrievalHintProvider =
return proc(messageId: SdsMessageID): seq[byte] {.gcsafe.} =
let (cb, userData) = getRetrievalHint(cast[pointer](ctx))
if cb.isNil():
return @[]
var hint: cstring
var hintLen: csize_t
cast[SdsRetrievalHintProvider](cb)(
messageId.cstring, addr hint, addr hintLen, userData
onMissingDependencies(
SdsMissingDependenciesEvent(
messageId: messageId,
missingDeps: toMissingDeps(missingDeps),
channelId: channelId,
)
)
let periodicSyncCb = proc() {.gcsafe.} =
onPeriodicSync(SdsPeriodicSyncEvent(ok: true))
let repairReadyCb = proc(message: seq[byte], channelId: SdsChannelID) {.gcsafe.} =
onRepairReady(SdsRepairReadyEvent(message: message, channelId: channelId))
let retrievalHintProvider = proc(messageId: SdsMessageID): seq[byte] {.gcsafe.} =
if sdsRetrievalHintCb.isNil():
return @[]
var hint: cstring
var hintLen: csize_t
cast[SdsRetrievalHintProvider](sdsRetrievalHintCb)(
messageId.cstring, addr hint, addr hintLen, sdsRetrievalHintUserData
)
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 libc malloc (Go's C.CBytes); free it
# with libc free, not Nim's deallocShared, to keep the allocator paired.
c_free(cast[pointer](hint))
return hintBytes
return @[]
await rm.setCallbacks(
messageReadyCb, messageSentCb, missingDependenciesCb, periodicSyncCb,
retrievalHintProvider, repairReadyCb,
)
return ok(rm)
################################################################################
### Request handlers (executed on the FFI worker thread)
### Async methods — each runs its body on the worker thread.
registerReqFFI(SdsCreateRmReq, ctx: ptr FFIContext[ReliabilityManager]):
proc(): Future[Result[string, string]] {.async.} =
# TODO: thread `participantId` through SdsNewReliabilityManager FFI input
# and remove this hardcoded "". Empty id silently disables SDS-R; this is
# acceptable as a temporary FFI-only fallback until sds-go-bindings and
# logos-delivery's C-side caller are updated to supply the identity.
let rm = newReliabilityManager(participantId = "".SdsParticipantID).valueOr:
error "Failed creating reliability manager", error = error
return err("Failed creating reliability manager: " & $error)
proc sdsWrapOutgoingMessage*(
rm: ReliabilityManager, req: SdsWrapRequest
): Future[Result[SdsWrapResponse, string]] {.ffi.} =
let wrapped = (
await wrapOutgoingMessage(rm, req.message, req.messageId, req.channelId)
).valueOr:
error "WRAP_MESSAGE failed", error = error
return err("error processing wrap request: " & $error)
return ok(SdsWrapResponse(message: wrapped))
await rm.setCallbacks(
onMessageReady(ctx), onMessageSent(ctx), onMissingDependencies(ctx),
onPeriodicSync(ctx), onRetrievalHint(ctx), onRepairReady(ctx),
proc sdsUnwrapReceivedMessage*(
rm: ReliabilityManager, req: SdsUnwrapRequest
): Future[Result[SdsUnwrapResponse, string]] {.ffi.} =
let (unwrapped, missingDeps, channelId) = (
await unwrapReceivedMessage(rm, req.message)
).valueOr:
return err("error processing unwrap request: " & $error)
return ok(
SdsUnwrapResponse(
message: unwrapped, channelId: channelId, missingDeps: toMissingDeps(missingDeps)
)
)
ctx.myLib[] = rm
return ok("")
proc sdsMarkDependenciesMet*(
rm: ReliabilityManager, req: SdsMarkDependenciesRequest
): Future[Result[string, string]] {.ffi.} =
(await markDependenciesMet(rm, req.messageIds, req.channelId)).isOkOr:
error "MARK_DEPENDENCIES_MET failed", error = error
return err("error processing mark-dependencies request: " & $error)
return ok("")
registerReqFFI(SdsResetRmReq, ctx: ptr FFIContext[ReliabilityManager]):
proc(): Future[Result[string, string]] {.async.} =
(await resetReliabilityManager(ctx.myLib[])).isOkOr:
error "RESET_RELIABILITY_MANAGER failed", error = error
return err("error processing RESET_RELIABILITY_MANAGER request: " & $error)
return ok("")
proc sdsReset*(rm: ReliabilityManager): Future[Result[string, string]] {.ffi.} =
(await resetReliabilityManager(rm)).isOkOr:
error "RESET failed", error = error
return err("error processing reset request: " & $error)
return ok("")
registerReqFFI(SdsStartPeriodicTasksReq, ctx: ptr FFIContext[ReliabilityManager]):
proc(): Future[Result[string, string]] {.async.} =
ctx.myLib[].startPeriodicTasks()
return ok("")
registerReqFFI(SdsWrapMessageReq, ctx: ptr FFIContext[ReliabilityManager]):
proc(
message: SdsSharedBytes, messageId: cstring, channelId: cstring
): Future[Result[string, string]] {.async.} =
var msg = message
defer:
deallocSharedSeq(msg)
let wrappedMessage = (
await wrapOutgoingMessage(ctx.myLib[], message.toSeq(), $messageId, $channelId)
).valueOr:
error "WRAP_MESSAGE failed", error = error
return err("error processing WRAP_MESSAGE request: " & $error)
# returns a comma-separated string of bytes
return ok(wrappedMessage.mapIt($it).join(","))
registerReqFFI(SdsUnwrapMessageReq, ctx: ptr FFIContext[ReliabilityManager]):
proc(message: SdsSharedBytes): Future[Result[string, string]] {.async.} =
var msg = message
defer:
deallocSharedSeq(msg)
let (unwrappedMessage, missingDeps, extractedChannelId) = (
await unwrapReceivedMessage(ctx.myLib[], message.toSeq())
).valueOr:
return err("error processing UNWRAP_MESSAGE request: " & $error)
# return the result as a json string
var node = newJObject()
node["message"] = %*unwrappedMessage
node["channelId"] = %*extractedChannelId
var missingDepsNode = newJArray()
for dep in missingDeps:
var depNode = newJObject()
depNode["messageId"] = %*dep.messageId
depNode["retrievalHint"] = %*encode(dep.retrievalHint)
missingDepsNode.add(depNode)
node["missingDeps"] = missingDepsNode
return ok($node)
registerReqFFI(SdsMarkDepsReq, ctx: ptr FFIContext[ReliabilityManager]):
proc(
messageIds: SdsSharedCstrs, channelId: cstring
): Future[Result[string, string]] {.async.} =
var ids = messageIds
defer:
freeSharedSeqCstr(ids)
let messageIdSeq = ids.toSeq().mapIt($it)
(await markDependenciesMet(ctx.myLib[], messageIdSeq, $channelId)).isOkOr:
error "MARK_DEPENDENCIES_MET failed", error = error
return err("error processing MARK_DEPENDENCIES_MET request: " & $error)
return ok("")
proc sdsStartPeriodicTasks*(
rm: ReliabilityManager
): Future[Result[string, string]] {.ffi.} =
# The empty await forces the macro down its async path so the body runs on the
# worker thread — startPeriodicTasks schedules futures on that thread's loop.
await sleepAsync(chronos.milliseconds(0))
rm.startPeriodicTasks()
return ok("")
################################################################################
### Dispatch helper
###
### Sends a request to the FFI worker thread and returns RET_OK/RET_ERR,
### reporting any failure through the callback. The try/except keeps the
### exported entry points `raises: []` (sendRequestToFFIThread can raise),
### which `processReq` alone would not guarantee.
### Destructor — runs library cleanup then tears down the FFI context.
template dispatchReq(
ctx: untyped, callback: FFICallBack, userData: pointer, reqExpr: untyped
) =
let sendRes =
try:
ffi_context.sendRequestToFFIThread(ctx, reqExpr)
except Exception as exc:
Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg)
if sendRes.isErr():
let m = "libsds error: " & sendRes.error
callback(RET_ERR, unsafeAddr m[0], cast[csize_t](m.len), userData)
return RET_ERR
return RET_OK
proc sdsDestroy*(rm: ReliabilityManager) {.ffiDtor.} =
discard
################################################################################
### Exported C entry points (called from the application thread)
###
### Signatures must match library/libsds.h exactly. Each one validates the
### context against the pool (rejecting nil/dangling pointers at the boundary),
### checks the callback, deep-copies any pointer payloads into shared memory,
### then dispatches a request to the FFI worker thread.
### Retrieval-hint provider (hand-written: a C function pointer cannot be passed
### as CBOR). The setter dispatches a request — the provider/userData pointers
### travel as uint64 — so the provider is stored in the worker thread's
### thread-local, where sdsCreate's hint closure reads it.
proc SdsNewReliabilityManager(
callback: FFICallBack, userData: pointer
): pointer {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
proc sdsNoopCallback(
callerRet: cint, msg: ptr cchar, len: csize_t, userData: pointer
) {.cdecl, gcsafe, raises: [].} =
discard
if isNil(callback):
echo "error: missing callback in SdsNewReliabilityManager"
return nil
registerReqFFI(SdsSetHintReq, ctx: ptr FFIContext[ReliabilityManager]):
proc(cbPtr: uint64, udPtr: uint64): Future[Result[string, string]] {.async.} =
sdsRetrievalHintCb = cast[pointer](cbPtr)
sdsRetrievalHintUserData = cast[pointer](udPtr)
return ok("")
let ctx = ReliabilityManagerFFIPool.createFFIContext().valueOr:
let msg = "Error creating SDS FFI context: " & $error
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return nil
let sendRes =
try:
ffi_context.sendRequestToFFIThread(ctx, SdsCreateRmReq.ffiNewReq(callback, userData))
except Exception as exc:
Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg)
if sendRes.isErr():
let msg = "error creating reliability manager: " & sendRes.error
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
discard ReliabilityManagerFFIPool.destroyFFIContext(ctx)
return nil
return cast[pointer](ctx)
proc SdsSetEventCallback(
ctx: ptr FFIContext[ReliabilityManager], callback: FFICallBack, userData: pointer
) {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
echo "error: invalid context in SdsSetEventCallback"
return
ctx[].callbackState.callback = cast[pointer](callback)
ctx[].callbackState.userData = userData
proc SdsSetRetrievalHintProvider(
proc sds_set_retrieval_hint_provider(
ctx: ptr FFIContext[ReliabilityManager],
callback: SdsRetrievalHintProvider,
userData: pointer,
) {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
echo "error: invalid context in SdsSetRetrievalHintProvider"
return
setRetrievalHint(cast[pointer](ctx), cast[pointer](callback), userData)
proc SdsCleanupReliabilityManager(
ctx: ptr FFIContext[ReliabilityManager], callback: FFICallBack, userData: pointer
): cint {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
return RET_ERR
if isNil(callback):
return RET_MISSING_CALLBACK
clearRetrievalHint(cast[pointer](ctx))
let res = ReliabilityManagerFFIPool.destroyFFIContext(ctx)
if res.isErr():
let msg = "error cleaning up reliability manager: " & res.error
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
let sendRes =
try:
ffi_context.sendRequestToFFIThread(
ctx,
SdsSetHintReq.ffiNewReq(
sdsNoopCallback, nil, cast[uint64](cast[pointer](callback)),
cast[uint64](userData),
),
)
except Exception as exc:
Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg)
if sendRes.isErr():
return RET_ERR
callback(RET_OK, nil, 0, userData)
return RET_OK
proc SdsResetReliabilityManager(
ctx: ptr FFIContext[ReliabilityManager], callback: FFICallBack, userData: pointer
): cint {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
return RET_ERR
if isNil(callback):
return RET_MISSING_CALLBACK
dispatchReq(ctx, callback, userData, SdsResetRmReq.ffiNewReq(callback, userData))
proc SdsWrapOutgoingMessage(
ctx: ptr FFIContext[ReliabilityManager],
message: pointer,
messageLen: csize_t,
messageId: cstring,
channelId: cstring,
callback: FFICallBack,
userData: pointer,
): cint {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
return RET_ERR
if isNil(callback):
return RET_MISSING_CALLBACK
if message == nil and messageLen > 0:
let msg = "libsds error: message pointer is NULL but length > 0"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return RET_ERR
if messageId == nil:
let msg = "libsds error: message ID pointer is NULL"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return RET_ERR
if channelId == nil:
let msg = "libsds error: channel ID pointer is NULL"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return RET_ERR
if $channelId == "":
let msg = "libsds error: channel ID is empty string"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return RET_ERR
let sharedMsg = copyToSharedSeqByte(message, messageLen.int)
dispatchReq(
ctx, callback, userData,
SdsWrapMessageReq.ffiNewReq(callback, userData, sharedMsg, messageId, channelId),
)
proc SdsUnwrapReceivedMessage(
ctx: ptr FFIContext[ReliabilityManager],
message: pointer,
messageLen: csize_t,
callback: FFICallBack,
userData: pointer,
): cint {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
return RET_ERR
if isNil(callback):
return RET_MISSING_CALLBACK
if message == nil and messageLen > 0:
let msg = "libsds error: message pointer is NULL but length > 0"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return RET_ERR
let sharedMsg = copyToSharedSeqByte(message, messageLen.int)
dispatchReq(ctx, callback, userData, SdsUnwrapMessageReq.ffiNewReq(callback, userData, sharedMsg))
proc SdsMarkDependenciesMet(
ctx: ptr FFIContext[ReliabilityManager],
messageIds: pointer,
count: csize_t,
channelId: cstring,
callback: FFICallBack,
userData: pointer,
): cint {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
return RET_ERR
if isNil(callback):
return RET_MISSING_CALLBACK
if messageIds == nil and count > 0:
let msg = "libsds error: MessageIDs pointer is NULL but count > 0"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return RET_ERR
if channelId == nil:
let msg = "libsds error: channel ID pointer is NULL"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return RET_ERR
if $channelId == "":
let msg = "libsds error: channel ID is empty string"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
return RET_ERR
let sharedIds = copyToSharedSeqCstr(messageIds, count.int)
dispatchReq(
ctx, callback, userData,
SdsMarkDepsReq.ffiNewReq(callback, userData, sharedIds, channelId),
)
proc SdsStartPeriodicTasks(
ctx: ptr FFIContext[ReliabilityManager], callback: FFICallBack, userData: pointer
): cint {.dynlib, exportc, cdecl, raises: [].} =
initializeLibrary()
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
return RET_ERR
if isNil(callback):
return RET_MISSING_CALLBACK
dispatchReq(ctx, callback, userData, SdsStartPeriodicTasksReq.ffiNewReq(callback, userData))
# Emit binding metadata (no-op unless -d:ffiGenBindings). Must follow every
# {.ffi.}/{.ffiCtor.}/{.ffiDtor.} annotation.
genBindings()

305
nimble.lock
View File

@ -12,27 +12,37 @@
}
},
"bearssl": {
"version": "0.2.6",
"vcsRevision": "11e798b62b8e6beabe958e048e9e24c7e0f9ee63",
"version": "0.2.8",
"vcsRevision": "22c6a76ce015bc07e011562bdcfc51d9446c1e82",
"url": "https://github.com/status-im/nim-bearssl",
"downloadMethod": "git",
"dependencies": [
"unittest2"
],
"checksums": {
"sha1": "7e068f119664cf47ad0cfb74ef4c56fb6b616523"
"sha1": "da4dd7ae96d536bdaf42dca9c85d7aed024b6a86"
}
},
"bearssl_pkey_decoder": {
"version": "0.1.0",
"vcsRevision": "21dd3710df9345ed2ad8bf8f882761e07863b8e0",
"url": "https://github.com/vacp2p/bearssl_pkey_decoder",
"testutils": {
"version": "0.8.1",
"vcsRevision": "6ce5e5e2301ccbc04b09d27ff78741ff4d352b4d",
"url": "https://github.com/status-im/nim-testutils",
"downloadMethod": "git",
"dependencies": [
"bearssl"
"unittest2"
],
"checksums": {
"sha1": "21b42e2e6ddca6c875d3fc50f36a5115abf51714"
"sha1": "96a11cf8b84fa9bd12d4a553afa1cc4b7f9df4e3"
}
},
"npeg": {
"version": "1.3.0",
"vcsRevision": "409f6796d0e880b3f0222c964d1da7de6e450811",
"url": "https://github.com/zevv/npeg",
"downloadMethod": "git",
"dependencies": [],
"checksums": {
"sha1": "64f15c85a059c889cb11c5fe72372677c50da621"
}
},
"results": {
@ -46,8 +56,8 @@
}
},
"stew": {
"version": "0.4.2",
"vcsRevision": "b66168735d6f3841c5239c3169d3fe5fe98b1257",
"version": "0.5.0",
"vcsRevision": "4382b18f04b3c43c8409bfcd6b62063773b2bbaa",
"url": "https://github.com/status-im/nim-stew",
"downloadMethod": "git",
"dependencies": [
@ -55,12 +65,56 @@
"unittest2"
],
"checksums": {
"sha1": "928e82cb8d2f554e8f10feb2349ee9c32fee3a8c"
"sha1": "db22942939773ab7d5a0f2b2668c237240c67dd6"
}
},
"httputils": {
"version": "0.4.1",
"vcsRevision": "f142cb2e8bd812dd002a6493b6082827bb248592",
"url": "https://github.com/status-im/nim-http-utils",
"downloadMethod": "git",
"dependencies": [
"stew",
"results",
"unittest2"
],
"checksums": {
"sha1": "016774ab31c3afff9a423f7d80584905ee59c570"
}
},
"chronos": {
"version": "4.2.2",
"vcsRevision": "45f43a9ad8bd8bcf5903b42f365c1c879bd54240",
"url": "https://github.com/status-im/nim-chronos",
"downloadMethod": "git",
"dependencies": [
"results",
"stew",
"bearssl",
"httputils",
"unittest2"
],
"checksums": {
"sha1": "3a4c9477df8cef20a04e4f1b54a2d74fdfc2a3d0"
}
},
"metrics": {
"version": "0.2.1",
"vcsRevision": "a1296caf3ebb5f30f51a5feae7749a30df2824c2",
"url": "https://github.com/status-im/nim-metrics",
"downloadMethod": "git",
"dependencies": [
"chronos",
"results",
"stew"
],
"checksums": {
"sha1": "84bb09873d7677c06046f391c7b473cd2fcff8a2"
}
},
"faststreams": {
"version": "0.5.0",
"vcsRevision": "ce27581a3e881f782f482cb66dc5b07a02bd615e",
"version": "0.5.1",
"vcsRevision": "50889cd16ec8771106cdd0eeea460039e8571e06",
"url": "https://github.com/status-im/nim-faststreams",
"downloadMethod": "git",
"dependencies": [
@ -68,12 +122,12 @@
"unittest2"
],
"checksums": {
"sha1": "ee61e507b805ae1df7ec936f03f2d101b0d72383"
"sha1": "969ceb3666e807db8fe5c8df63466749822367a9"
}
},
"serialization": {
"version": "0.5.2",
"vcsRevision": "b0f2fa32960ea532a184394b0f27be37bd80248b",
"version": "0.5.3",
"vcsRevision": "4092500cea76154576539371709ae801afbd2a9d",
"url": "https://github.com/status-im/nim-serialization",
"downloadMethod": "git",
"dependencies": [
@ -82,7 +136,37 @@
"stew"
],
"checksums": {
"sha1": "fa35c1bb76a0a02a2379fe86eaae0957c7527cb8"
"sha1": "c087d26c50da40436599163888532660d6f9e631"
}
},
"protobuf_serialization": {
"version": "0.5.2",
"vcsRevision": "cec5f1da897c0b3e6d3a1f2da6a36b4bbdc3a1a8",
"url": "https://github.com/status-im/nim-protobuf-serialization",
"downloadMethod": "git",
"dependencies": [
"stew",
"faststreams",
"serialization",
"npeg",
"unittest2"
],
"checksums": {
"sha1": "48db6535e6c85825c7761820388a0e50c8f3eab3"
}
},
"cbor_serialization": {
"version": "0.3.0",
"vcsRevision": "1664160e04d153573373afddc552b9cbf6fbe4dc",
"url": "https://github.com/vacp2p/nim-cbor-serialization",
"downloadMethod": "git",
"dependencies": [
"serialization",
"stew",
"results"
],
"checksums": {
"sha1": "ab126eae09a6e39c72972a6a0b83cb06a2ffe8f0"
}
},
"json_serialization": {
@ -100,18 +184,6 @@
"sha1": "8b3115354104858a0ac9019356fb29720529c2bd"
}
},
"testutils": {
"version": "0.8.0",
"vcsRevision": "e4d37dc1652d5c63afb89907efb5a5e812261797",
"url": "https://github.com/status-im/nim-testutils",
"downloadMethod": "git",
"dependencies": [
"unittest2"
],
"checksums": {
"sha1": "d1678f50aa47d113b4e77d41eec2190830b523fa"
}
},
"chronicles": {
"version": "0.12.2",
"vcsRevision": "27ec507429a4eb81edc20f28292ee8ec420be05b",
@ -127,170 +199,6 @@
"sha1": "02febb20d088120b2836d3306cfa21f434f88f65"
}
},
"httputils": {
"version": "0.4.0",
"vcsRevision": "c53852d9e24205b6363bba517fa8ee7bde823691",
"url": "https://github.com/status-im/nim-http-utils",
"downloadMethod": "git",
"dependencies": [
"stew",
"results",
"unittest2"
],
"checksums": {
"sha1": "298bc5b6fe4e5aa9c3b7e2ebfa17191675020f10"
}
},
"chronos": {
"version": "4.0.4",
"vcsRevision": "0646c444fce7c7ed08ef6f2c9a7abfd172ffe655",
"url": "https://github.com/status-im/nim-chronos",
"downloadMethod": "git",
"dependencies": [
"results",
"stew",
"bearssl",
"httputils",
"unittest2"
],
"checksums": {
"sha1": "455802a90204d8ad6b31d53f2efff8ebfe4c834a"
}
},
"dnsclient": {
"version": "0.3.4",
"vcsRevision": "23214235d4784d24aceed99bbfe153379ea557c8",
"url": "https://github.com/ba0f3/dnsclient.nim",
"downloadMethod": "git",
"dependencies": [],
"checksums": {
"sha1": "65262c7e533ff49d6aca5539da4bc6c6ce132f40"
}
},
"jwt": {
"version": "0.2",
"vcsRevision": "18f8378de52b241f321c1f9ea905456e89b95c6f",
"url": "https://github.com/vacp2p/nim-jwt.git",
"downloadMethod": "git",
"dependencies": [
"bearssl",
"bearssl_pkey_decoder"
],
"checksums": {
"sha1": "bcfd6fc9c5e10a52b87117219b7ab5c98136bc8e"
}
},
"nimcrypto": {
"version": "0.7.3",
"vcsRevision": "b3dbc9c4d08e58c5b7bfad6dc7ef2ee52f2f4c08",
"url": "https://github.com/cheatfate/nimcrypto",
"downloadMethod": "git",
"dependencies": [],
"checksums": {
"sha1": "f72b90fe3f4da09efa482de4f8729e7ee4abea2f"
}
},
"metrics": {
"version": "0.1.2",
"vcsRevision": "11d0cddfb0e711aa2a8c75d1892ae24a64c299fc",
"url": "https://github.com/status-im/nim-metrics",
"downloadMethod": "git",
"dependencies": [
"chronos",
"results",
"stew"
],
"checksums": {
"sha1": "5cdac99d85d3c146d170e85064c88fb28f377842"
}
},
"secp256k1": {
"version": "0.6.0.3.2",
"vcsRevision": "d8f1288b7c72f00be5fc2c5ea72bf5cae1eafb15",
"url": "https://github.com/status-im/nim-secp256k1",
"downloadMethod": "git",
"dependencies": [
"stew",
"results",
"nimcrypto"
],
"checksums": {
"sha1": "6618ef9de17121846a8c1d0317026b0ce8584e10"
}
},
"zlib": {
"version": "0.1.0",
"vcsRevision": "e680f269fb01af2c34a2ba879ff281795a5258fe",
"url": "https://github.com/status-im/nim-zlib",
"downloadMethod": "git",
"dependencies": [
"stew",
"results"
],
"checksums": {
"sha1": "bbde4f5a97a84b450fef7d107461e5f35cf2b47f"
}
},
"websock": {
"version": "0.2.1",
"vcsRevision": "35ae76f1559e835c80f9c1a3943bf995d3dd9eb5",
"url": "https://github.com/status-im/nim-websock",
"downloadMethod": "git",
"dependencies": [
"chronos",
"httputils",
"chronicles",
"stew",
"nimcrypto",
"bearssl",
"results",
"zlib"
],
"checksums": {
"sha1": "1cb5efa10cd389bc01d0707c242ae010c76a03cd"
}
},
"lsquic": {
"version": "0.0.1",
"vcsRevision": "4fb03ee7bfb39aecb3316889fdcb60bec3d0936f",
"url": "https://github.com/vacp2p/nim-lsquic",
"downloadMethod": "git",
"dependencies": [
"zlib",
"stew",
"chronos",
"nimcrypto",
"unittest2",
"chronicles"
],
"checksums": {
"sha1": "f465fa994346490d0924d162f53d9b5aec62f948"
}
},
"libp2p": {
"version": "1.15.2",
"vcsRevision": "ca48c3718246bb411ff0e354a70cb82d9a28de0d",
"url": "https://github.com/vacp2p/nim-libp2p",
"downloadMethod": "git",
"dependencies": [
"nimcrypto",
"dnsclient",
"bearssl",
"chronicles",
"chronos",
"metrics",
"secp256k1",
"stew",
"websock",
"unittest2",
"results",
"lsquic",
"jwt"
],
"checksums": {
"sha1": "3b2cdc7e00261eb4210ca3d44ec3bd64c2b7bbba"
}
},
"stint": {
"version": "0.8.2",
"vcsRevision": "470b7892561b5179ab20bd389a69217d6213fe58",
@ -315,17 +223,18 @@
}
},
"ffi": {
"version": "0.1.4",
"vcsRevision": "fb25f069d2dfae2b543d79d2c1a81f197de22a2b",
"version": "0.2.0",
"vcsRevision": "a66c53a34b8c44cbb952294585942ca4434a9321",
"url": "https://github.com/logos-messaging/nim-ffi",
"downloadMethod": "git",
"dependencies": [
"chronos",
"chronicles",
"taskpools"
"taskpools",
"cbor_serialization"
],
"checksums": {
"sha1": "4a5d4020a40106fa2a698d5fe975b9a8ba961f91"
"sha1": "edc4f13560feef402582e9fff98788976b166ce2"
}
}
},

1
nix/default.nix
View File

@ -81,6 +81,7 @@ in stdenv.mkDerivation {
version = "${version}-${revision}";
env = {
SDS_NIX_DEPS = "1";
NIMFLAGS = "-d:disableMarchNative";
ANDROID_SDK_ROOT = optionalString isAndroidBuild pkgs.androidPkgs.sdk;
ANDROID_NDK_ROOT = optionalString isAndroidBuild pkgs.androidPkgs.ndk;

154
nix/deps.nix
View File

@ -12,15 +12,22 @@
bearssl = pkgs.fetchgit {
url = "https://github.com/status-im/nim-bearssl";
rev = "11e798b62b8e6beabe958e048e9e24c7e0f9ee63";
sha256 = "0qx36iiawrhmx9qjqcyfvz0134ph9dy8ryq3ch8d31gq6ir7aw84";
rev = "22c6a76ce015bc07e011562bdcfc51d9446c1e82";
sha256 = "1cvdd7lfrpa6asmc39al3g4py5nqhpqmvypc36r5qyv7p5arc8a3";
fetchSubmodules = true;
};
bearssl_pkey_decoder = pkgs.fetchgit {
url = "https://github.com/vacp2p/bearssl_pkey_decoder";
rev = "21dd3710df9345ed2ad8bf8f882761e07863b8e0";
sha256 = "0bl3f147zmkazbhdkr4cj1nipf9rqiw3g4hh1j424k9hpl55zdpg";
testutils = pkgs.fetchgit {
url = "https://github.com/status-im/nim-testutils";
rev = "6ce5e5e2301ccbc04b09d27ff78741ff4d352b4d";
sha256 = "1vbkr6i5yxhc2ai3b7rbglhmyc98f99x874fqdp6a152a6kqgwxy";
fetchSubmodules = true;
};
npeg = pkgs.fetchgit {
url = "https://github.com/zevv/npeg";
rev = "409f6796d0e880b3f0222c964d1da7de6e450811";
sha256 = "1h2f5znbpa3svk7wsw2axn8f7f59d23xq85z148kiv6fqh0ffwbm";
fetchSubmodules = true;
};
@ -33,22 +40,57 @@
stew = pkgs.fetchgit {
url = "https://github.com/status-im/nim-stew";
rev = "b66168735d6f3841c5239c3169d3fe5fe98b1257";
sha256 = "10n71vfa6klzd9dmal96jy0hiqk04gaj8wc9g91z6fclryf0yq92";
rev = "4382b18f04b3c43c8409bfcd6b62063773b2bbaa";
sha256 = "0mx9g5m636h3sk5pllcpylk51brf7lx91izx3gc23k3ih3hrxyk2";
fetchSubmodules = true;
};
httputils = pkgs.fetchgit {
url = "https://github.com/status-im/nim-http-utils";
rev = "f142cb2e8bd812dd002a6493b6082827bb248592";
sha256 = "03msj4zdxraz4qx9cidb17g7v0asazxv91nng6xxbzjxz0qaqxw6";
fetchSubmodules = true;
};
chronos = pkgs.fetchgit {
url = "https://github.com/status-im/nim-chronos";
rev = "45f43a9ad8bd8bcf5903b42f365c1c879bd54240";
sha256 = "1v1n59zfzznp97pvwgs9kf136bqmv4x2s2y9f24msspa7qv27w39";
fetchSubmodules = true;
};
metrics = pkgs.fetchgit {
url = "https://github.com/status-im/nim-metrics";
rev = "a1296caf3ebb5f30f51a5feae7749a30df2824c2";
sha256 = "02vxqy20g8012ks939ac25ksc25k727q84si0p2cmihy5bw1a3qm";
fetchSubmodules = true;
};
faststreams = pkgs.fetchgit {
url = "https://github.com/status-im/nim-faststreams";
rev = "ce27581a3e881f782f482cb66dc5b07a02bd615e";
sha256 = "0y6bw2scnmr8cxj4fg18w7f34l2bh9qwg5nhlgd84m9fpr5bqarn";
rev = "50889cd16ec8771106cdd0eeea460039e8571e06";
sha256 = "1hd4bhvw5lzwg924i8dif5mi61h0ayiplq38djvrdbfsjdhw2zvw";
fetchSubmodules = true;
};
serialization = pkgs.fetchgit {
url = "https://github.com/status-im/nim-serialization";
rev = "b0f2fa32960ea532a184394b0f27be37bd80248b";
sha256 = "0wip1fjx7ka39ck1g1xvmyarzq1p5dlngpqil6zff8k8z5skiz27";
rev = "4092500cea76154576539371709ae801afbd2a9d";
sha256 = "04pz6d6p3nd1y2khbb667fcd6p2jk4bxv65iaffzq06bqqhalcwc";
fetchSubmodules = true;
};
protobuf_serialization = pkgs.fetchgit {
url = "https://github.com/status-im/nim-protobuf-serialization";
rev = "cec5f1da897c0b3e6d3a1f2da6a36b4bbdc3a1a8";
sha256 = "0dmrm8l1293fcmyzm4kmhwplyrd1clrjs8hpf9s9bpbyjw7vf927";
fetchSubmodules = true;
};
cbor_serialization = pkgs.fetchgit {
url = "https://github.com/vacp2p/nim-cbor-serialization";
rev = "1664160e04d153573373afddc552b9cbf6fbe4dc";
sha256 = "0c1rj4fk0fcqvsf0yqhxvm8h10aww75gi4yfsjhlczh88ypywii2";
fetchSubmodules = true;
};
@ -59,13 +101,6 @@
fetchSubmodules = true;
};
testutils = pkgs.fetchgit {
url = "https://github.com/status-im/nim-testutils";
rev = "e4d37dc1652d5c63afb89907efb5a5e812261797";
sha256 = "0nv0a9jm5b1rn3y52cxvyj8xz3jg235mp0xbirfp2cda0icgy1si";
fetchSubmodules = true;
};
chronicles = pkgs.fetchgit {
url = "https://github.com/status-im/nim-chronicles";
rev = "27ec507429a4eb81edc20f28292ee8ec420be05b";
@ -73,83 +108,6 @@
fetchSubmodules = true;
};
httputils = pkgs.fetchgit {
url = "https://github.com/status-im/nim-http-utils";
rev = "c53852d9e24205b6363bba517fa8ee7bde823691";
sha256 = "1b332smfyp2yvhvfjrfqy4kvh9pc5w6hqh17f1yclz5z1j5xdpf1";
fetchSubmodules = true;
};
chronos = pkgs.fetchgit {
url = "https://github.com/status-im/nim-chronos";
rev = "0646c444fce7c7ed08ef6f2c9a7abfd172ffe655";
sha256 = "1r499jl0lhnjq7hgddwgjl0gh3y1mprnqkhk0h6yh3cwgsmr5ym9";
fetchSubmodules = true;
};
dnsclient = pkgs.fetchgit {
url = "https://github.com/ba0f3/dnsclient.nim";
rev = "23214235d4784d24aceed99bbfe153379ea557c8";
sha256 = "03mf3lw5c0m5nq9ppa49nylrl8ibkv2zzlc0wyhqg7w09kz6hks6";
fetchSubmodules = true;
};
jwt = pkgs.fetchgit {
url = "https://github.com/vacp2p/nim-jwt.git";
rev = "18f8378de52b241f321c1f9ea905456e89b95c6f";
sha256 = "1986czmszdxj6g9yr7xn1fx8y2y9mwpb3f1bn9nc6973qawsdm0p";
fetchSubmodules = true;
};
nimcrypto = pkgs.fetchgit {
url = "https://github.com/cheatfate/nimcrypto";
rev = "b3dbc9c4d08e58c5b7bfad6dc7ef2ee52f2f4c08";
sha256 = "1v4rz42lwcazs6isi3kmjylkisr84mh0kgmlqycx4i885dn3g0l4";
fetchSubmodules = true;
};
metrics = pkgs.fetchgit {
url = "https://github.com/status-im/nim-metrics";
rev = "11d0cddfb0e711aa2a8c75d1892ae24a64c299fc";
sha256 = "1jrf2cf7v3iqjsk6grzvivxic1shhaxnvab6d35rxs2kcy6b5dv0";
fetchSubmodules = true;
};
secp256k1 = pkgs.fetchgit {
url = "https://github.com/status-im/nim-secp256k1";
rev = "d8f1288b7c72f00be5fc2c5ea72bf5cae1eafb15";
sha256 = "1qjrmwbngb73f6r1fznvig53nyal7wj41d1cmqfksrmivk2sgrn2";
fetchSubmodules = true;
};
zlib = pkgs.fetchgit {
url = "https://github.com/status-im/nim-zlib";
rev = "e680f269fb01af2c34a2ba879ff281795a5258fe";
sha256 = "1xw9f1gjsgqihdg7kdkbaq1wankgnx2vn9l3ihc6nqk2jzv5bvk5";
fetchSubmodules = true;
};
websock = pkgs.fetchgit {
url = "https://github.com/status-im/nim-websock";
rev = "35ae76f1559e835c80f9c1a3943bf995d3dd9eb5";
sha256 = "1j6dklzb6b6bv2aiglbiyflja2vdpmyxfirv98f49y62mykq0yrw";
fetchSubmodules = true;
};
lsquic = pkgs.fetchgit {
url = "https://github.com/vacp2p/nim-lsquic";
rev = "4fb03ee7bfb39aecb3316889fdcb60bec3d0936f";
sha256 = "0qdhcd4hyp185szc9sv3jvwdwc9zp3j0syy7glxv13k9bchfmkfg";
fetchSubmodules = true;
};
libp2p = pkgs.fetchgit {
url = "https://github.com/vacp2p/nim-libp2p";
rev = "ca48c3718246bb411ff0e354a70cb82d9a28de0d";
sha256 = "07qfjjrq6w7bj9dbchvcrpla47jidngbrgmigbhl7fh3cfkdabc9";
fetchSubmodules = true;
};
stint = pkgs.fetchgit {
url = "https://github.com/status-im/nim-stint";
rev = "470b7892561b5179ab20bd389a69217d6213fe58";
@ -166,8 +124,8 @@
ffi = pkgs.fetchgit {
url = "https://github.com/logos-messaging/nim-ffi";
rev = "fb25f069d2dfae2b543d79d2c1a81f197de22a2b";
sha256 = "0zkjnrm2yjlw27q99kv2x8ll61mbz4nr0cvmyq0csydh43c08k0p";
rev = "a66c53a34b8c44cbb952294585942ca4434a9321";
sha256 = "1vnfzi9a9fhpspr963z5in2g6n4lm4xcgrbvzm3fgfiqc5i6l6sa";
fetchSubmodules = true;
};

28
sds.nimble
View File

@ -1,7 +1,7 @@
import strutils, os
# Package
version = "0.3.0"
version = "0.4.0"
author = "Logos Messaging Team"
description = "E2E Scalable Data Sync API"
license = "MIT"
@ -10,15 +10,22 @@ srcDir = "sds"
# Dependencies
requires "nim >= 2.2.4"
requires "chronos >= 4.0.4"
requires "libp2p >= 1.15.2"
requires "protobuf_serialization >= 0.5.0"
requires "chronicles"
requires "stew"
requires "stint"
requires "metrics"
requires "results"
# Only library/ (the FFI wrapper) uses nim-ffi, not core sds/. Keep the floor
# low so core-only consumers aren't forced up; nimble.lock pins library/'s version.
requires "https://github.com/logos-messaging/nim-ffi >= 0.1.3"
# nim-ffi isn't in the nimble registry, so a plain `nimble` build fetches it by
# URL. The Nix build pre-populates deps offline (can't clone) and sets
# SDS_NIX_DEPS to resolve it by name from the installed pkgs2 instead.
# This branch adopts the nim-ffi 0.2 CBOR ABI, so it pins v0.2.0-rc.1 and
# overrides master's relaxed >= 0.1.3 floor (which exists for core-only consumers).
when existsEnv("SDS_NIX_DEPS"):
requires "ffi >= 0.2.0"
else:
requires "https://github.com/logos-messaging/nim-ffi#a66c53a34b8c44cbb952294585942ca4434a9321"
requires "https://github.com/vacp2p/nim-cbor-serialization#v0.3.0"
proc buildLibrary(
outLibNameAndExt: string,
@ -32,16 +39,16 @@ proc buildLibrary(
if `type` == "static":
exec "nim c" & " --out:build/" & outLibNameAndExt &
" --threads:on --app:staticlib --opt:size --noMain --mm:refc --header --nimMainPrefix:libsds " &
" --threads:on --app:staticlib --opt:size --noMain --mm:refc --header --nimMainPrefix:libsds -d:noSignalHandler " &
extra_params & " " & srcDir & name & ".nim"
else:
when defined(windows):
exec "nim c" & " --out:build/" & outLibNameAndExt &
" --threads:on --app:lib --opt:size --noMain --mm:refc --header --nimMainPrefix:libsds " &
" --threads:on --app:lib --opt:size --noMain --mm:refc --header --nimMainPrefix:libsds -d:noSignalHandler " &
extra_params & " " & srcDir & name & ".nim"
else:
exec "nim c" & " --out:build/" & outLibNameAndExt &
" --threads:on --app:lib --opt:size --noMain --mm:refc --header --nimMainPrefix:libsds " &
" --threads:on --app:lib --opt:size --noMain --mm:refc --header --nimMainPrefix:libsds -d:noSignalHandler " &
extra_params & " " & srcDir & name & ".nim"
proc getMyCpu(): string =
@ -161,8 +168,8 @@ proc buildMobileIOS(srcDir = ".", sdkPath = "") =
# Use unique symbol prefix to avoid conflicts with other Nim libraries
exec "nim c" & " --nimcache:" & nimcacheDir & " --os:ios --cpu:" & cpu &
" --compileOnly:on" & " --noMain --mm:refc" & " --threads:on --opt:size --header" &
" --nimMainPrefix:libsds" & " --cc:clang" & " -d:useMalloc" & " " & srcDir &
"/libsds.nim"
" --nimMainPrefix:libsds" & " --cc:clang" & " -d:useMalloc" & " -d:noSignalHandler" &
" " & srcDir & "/libsds.nim"
# 2) Compile all generated C files to object files with hidden visibility
# This prevents symbol conflicts with other Nim libraries (e.g., libnim_status_client)
@ -259,6 +266,7 @@ proc buildMobileAndroid(srcDir = ".", extra_params = "") =
exec "nim c" &
" --out:" & outDir & "/libsds.so" &
" --threads:on --app:lib --opt:size --noMain --mm:refc --nimMainPrefix:libsds" &
" -d:noSignalHandler" &
" --cc:clang" &
" --clang.exe:\"" & ndkClang & "\"" &
" --clang.linkerexe:\"" & ndkClang & "\"" &

321
sds/protobuf.nim
View File

@ -1,189 +1,222 @@
import libp2p/protobuf/minprotobuf
## SDS network wire codec.
##
## Messages are described as annotated protobuf types and (de)serialised with
## `nim-protobuf-serialization`'s type-driven `Protobuf.encode/decode`. The
## domain types (`SdsMessage`, `HistoryEntry`) keep their distinct/`requiresInit`
## shape; small `*PB` mirrors carry the field annotations and a trivial
## conversion bridges the two. The mirror string-ish fields are `seq[byte]`
## (not `pstring`) so message/channel/sender ids stay opaque bytes — no UTF-8
## validation — and the distinct `SdsParticipantID` needs no special support.
##
## Singular fields use the proto3 `optional` label (`Opt[T]`), which is the
## recommended form for forward-compatibility; presence is exposed but the
## actual validity of mandatory identifiers is checked at the application layer
## after decoding (proto3 has no `required`).
{.push raises: [].}
import endians
import protobuf_serialization
import protobuf_serialization/pkg/results
import ./types/[sds_message_id, history_entry, sds_message, reliability_error]
import ./protobufutil
import ./bloom
import ./sds_utils
proc encodeHistoryEntry*(entry: HistoryEntry): ProtoBuffer =
var entryPb = initProtoBuffer()
entryPb.write(1, entry.messageId)
if entry.retrievalHint.len > 0:
entryPb.write(2, entry.retrievalHint)
if entry.senderId.len > 0:
entryPb.write(3, entry.senderId.string)
entryPb.finish()
entryPb
# ---------------------------------------------------------------------------
# Wire types
# ---------------------------------------------------------------------------
proc decodeHistoryEntry*(entryPb: ProtoBuffer): ProtobufResult[HistoryEntry] =
var entry = HistoryEntry.init("")
if not ?entryPb.getField(1, entry.messageId):
return err(ProtobufError.missingRequiredField("HistoryEntry.messageId"))
discard entryPb.getField(2, entry.retrievalHint)
var senderIdStr: string
if entryPb.getField(3, senderIdStr).valueOr(false):
entry.senderId = senderIdStr.SdsParticipantID
ok(entry)
type
HistoryEntryPB* {.proto3.} = object
messageId* {.fieldNumber: 1.}: Opt[seq[byte]]
retrievalHint* {.fieldNumber: 2.}: Opt[seq[byte]]
senderId* {.fieldNumber: 3.}: Opt[seq[byte]]
proc encode*(msg: SdsMessage): ProtoBuffer =
var pb = initProtoBuffer()
SdsMessagePB* {.proto3.} = object
messageId* {.fieldNumber: 1.}: Opt[seq[byte]]
lamportTimestamp* {.fieldNumber: 2, pint.}: Opt[int64]
causalHistory* {.fieldNumber: 3.}: seq[HistoryEntryPB]
channelId* {.fieldNumber: 4.}: Opt[seq[byte]]
content* {.fieldNumber: 5.}: Opt[seq[byte]]
bloomFilter* {.fieldNumber: 6.}: Opt[seq[byte]]
senderId* {.fieldNumber: 7.}: Opt[seq[byte]]
repairRequest* {.fieldNumber: 13.}: seq[HistoryEntryPB]
pb.write(1, msg.messageId)
pb.write(2, uint64(msg.lamportTimestamp))
BloomFilterPB {.proto3.} = object
data {.fieldNumber: 1.}: Opt[seq[byte]]
capacity {.fieldNumber: 2, pint.}: Opt[uint64]
errorRate {.fieldNumber: 3, pint.}: Opt[uint64]
kHashes {.fieldNumber: 4, pint.}: Opt[uint64]
mBits {.fieldNumber: 5, pint.}: Opt[uint64]
for entry in msg.causalHistory:
let entryPb = encodeHistoryEntry(entry)
pb.write(3, entryPb.buffer)
# ---------------------------------------------------------------------------
# string <-> bytes (opaque, no UTF-8 validation) and optional-bytes helper
# ---------------------------------------------------------------------------
pb.write(4, msg.channelId)
pb.write(5, msg.content)
pb.write(6, msg.bloomFilter)
func toBytes(s: string): seq[byte] =
var b = newSeq[byte](s.len)
if s.len > 0:
copyMem(addr b[0], unsafeAddr s[0], s.len)
return b
if msg.senderId.len > 0:
pb.write(7, msg.senderId.string)
func toStr(b: seq[byte]): string =
var s = newString(b.len)
if b.len > 0:
copyMem(addr s[0], unsafeAddr b[0], b.len)
return s
for entry in msg.repairRequest:
let entryPb = encodeHistoryEntry(entry)
pb.write(13, entryPb.buffer)
func optBytes(b: seq[byte]): Opt[seq[byte]] =
## Present only when non-empty, so empty optionals stay off the wire.
if b.len > 0:
return Opt.some(b)
return Opt.none(seq[byte])
pb.finish()
# ---------------------------------------------------------------------------
# Domain <-> wire conversion
# ---------------------------------------------------------------------------
func toPB*(e: HistoryEntry): HistoryEntryPB =
return HistoryEntryPB(
messageId: optBytes(e.messageId.toBytes),
retrievalHint: optBytes(e.retrievalHint),
senderId: optBytes(e.senderId.string.toBytes),
)
func fromPB*(e: HistoryEntryPB): HistoryEntry =
return HistoryEntry(
messageId: e.messageId.valueOr(@[]).toStr,
retrievalHint: e.retrievalHint.valueOr(@[]),
senderId: e.senderId.valueOr(@[]).toStr.SdsParticipantID,
)
func toPB*(m: SdsMessage): SdsMessagePB =
var pb = SdsMessagePB(
messageId: optBytes(m.messageId.toBytes),
lamportTimestamp: Opt.some(m.lamportTimestamp),
channelId: optBytes(m.channelId.toBytes),
content: optBytes(m.content),
bloomFilter: optBytes(m.bloomFilter),
senderId: optBytes(m.senderId.string.toBytes),
)
for e in m.causalHistory:
pb.causalHistory.add(e.toPB)
for e in m.repairRequest:
pb.repairRequest.add(e.toPB)
return pb
proc decode*(T: type SdsMessage, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var msg = SdsMessage.init("", 0, @[], "", @[], @[])
func fromPB*(pb: SdsMessagePB): SdsMessage =
var causal: seq[HistoryEntry]
for e in pb.causalHistory:
causal.add(e.fromPB)
var repair: seq[HistoryEntry]
for e in pb.repairRequest:
repair.add(e.fromPB)
return SdsMessage.init(
messageId = pb.messageId.valueOr(@[]).toStr,
lamportTimestamp = pb.lamportTimestamp.valueOr(0'i64),
causalHistory = causal,
channelId = pb.channelId.valueOr(@[]).toStr,
content = pb.content.valueOr(@[]),
bloomFilter = pb.bloomFilter.valueOr(@[]),
senderId = pb.senderId.valueOr(@[]).toStr.SdsParticipantID,
repairRequest = repair,
)
if not ?pb.getField(1, msg.messageId):
return err(ProtobufError.missingRequiredField("messageId"))
var timestamp: uint64
if not ?pb.getField(2, timestamp):
return err(ProtobufError.missingRequiredField("lamportTimestamp"))
msg.lamportTimestamp = int64(timestamp)
# Handle both old and new causal history formats
var historyBuffers: seq[seq[byte]]
if pb.getRepeatedField(3, historyBuffers).isOk():
# New format: repeated HistoryEntry
for histBuffer in historyBuffers:
let entryPb = initProtoBuffer(histBuffer)
let entry = ?decodeHistoryEntry(entryPb)
msg.causalHistory.add(entry)
else:
# Try old format: repeated string
var causalHistory: seq[SdsMessageID]
let histResult = pb.getRepeatedField(3, causalHistory)
if histResult.isOk():
msg.causalHistory = toCausalHistory(causalHistory)
if not ?pb.getField(4, msg.channelId):
return err(ProtobufError.missingRequiredField("channelId"))
if not ?pb.getField(5, msg.content):
return err(ProtobufError.missingRequiredField("content"))
if not ?pb.getField(6, msg.bloomFilter):
msg.bloomFilter = @[] # Empty if not present
# SDS-R: decode senderId (field 7, optional)
var msgSenderIdStr: string
if pb.getField(7, msgSenderIdStr).valueOr(false):
msg.senderId = msgSenderIdStr.SdsParticipantID
# SDS-R: decode repair request (field 13, optional)
var repairBuffers: seq[seq[byte]]
if pb.getRepeatedField(13, repairBuffers).isOk():
for repairBuffer in repairBuffers:
let entryPb = initProtoBuffer(repairBuffer)
let entry = ?decodeHistoryEntry(entryPb)
msg.repairRequest.add(entry)
return ok(msg)
proc extractChannelId*(data: seq[byte]): Result[SdsChannelID, ReliabilityError] =
## For extraction of channel ID without full message deserialization
try:
let pb = initProtoBuffer(data)
var channelId: SdsChannelID
let fieldOk = pb.getField(4, channelId).valueOr:
return err(ReliabilityError.reDeserializationError)
if not fieldOk:
return err(ReliabilityError.reDeserializationError)
return ok(channelId)
except:
return err(ReliabilityError.reDeserializationError)
# ---------------------------------------------------------------------------
# Message (de)serialisation
# ---------------------------------------------------------------------------
proc serializeMessage*(msg: SdsMessage): Result[seq[byte], ReliabilityError] =
let pb = encode(msg)
return ok(pb.buffer)
try:
return ok(Protobuf.encode(msg.toPB))
except CatchableError:
return err(ReliabilityError.reSerializationError)
proc deserializeMessage*(data: seq[byte]): Result[SdsMessage, ReliabilityError] =
let msg = SdsMessage.decode(data).valueOr:
## proto3 has no required fields, so the mandatory identifiers are validated
## by hand after decoding. `content`/`bloomFilter`/`lamportTimestamp` may
## legitimately be empty/zero (e.g. periodic sync messages).
try:
let msg = Protobuf.decode(data, SdsMessagePB).fromPB
if msg.messageId.len == 0 or msg.channelId.len == 0:
return err(ReliabilityError.reDeserializationError)
for e in msg.causalHistory:
if e.messageId.len == 0:
return err(ReliabilityError.reDeserializationError)
for e in msg.repairRequest:
if e.messageId.len == 0:
return err(ReliabilityError.reDeserializationError)
return ok(msg)
except CatchableError:
return err(ReliabilityError.reDeserializationError)
return ok(msg)
proc extractChannelId*(data: seq[byte]): Result[SdsChannelID, ReliabilityError] =
## Channel ID without retaining the rest of the decoded message.
try:
return ok(Protobuf.decode(data, SdsMessagePB).channelId.valueOr(@[]).toStr)
except CatchableError:
return err(ReliabilityError.reDeserializationError)
# Single `HistoryEntry` (de)serialisation, used by the snapshot codec for the
# repair-buffer entries it embeds. Kept here so all `Protobuf.decode` calls live
# in this module.
proc serializeHistoryEntry*(e: HistoryEntry): Result[seq[byte], ReliabilityError] =
try:
return ok(Protobuf.encode(e.toPB))
except CatchableError:
return err(ReliabilityError.reSerializationError)
proc deserializeHistoryEntry*(data: seq[byte]): Result[HistoryEntry, ReliabilityError] =
try:
return ok(Protobuf.decode(data, HistoryEntryPB).fromPB)
except CatchableError:
return err(ReliabilityError.reDeserializationError)
# ---------------------------------------------------------------------------
# Bloom filter (de)serialisation
# ---------------------------------------------------------------------------
proc serializeBloomFilter*(filter: BloomFilter): Result[seq[byte], ReliabilityError] =
var pb = initProtoBuffer()
try:
var bytes = newSeq[byte](filter.intArray.len * sizeof(int))
for i, val in filter.intArray:
var leVal: int
littleEndian64(addr leVal, unsafeAddr val)
let start = i * sizeof(int)
copyMem(addr bytes[start], addr leVal, sizeof(int))
copyMem(addr bytes[i * sizeof(int)], addr leVal, sizeof(int))
pb.write(1, bytes)
pb.write(2, uint64(filter.capacity))
pb.write(3, uint64(filter.errorRate * 1_000_000))
pb.write(4, uint64(filter.kHashes))
pb.write(5, uint64(filter.mBits))
except:
let pb = BloomFilterPB(
data: optBytes(bytes),
capacity: Opt.some(uint64(filter.capacity)),
errorRate: Opt.some(uint64(filter.errorRate * 1_000_000)),
kHashes: Opt.some(uint64(filter.kHashes)),
mBits: Opt.some(uint64(filter.mBits)),
)
return ok(Protobuf.encode(pb))
except CatchableError:
return err(ReliabilityError.reSerializationError)
pb.finish()
return ok(pb.buffer)
proc deserializeBloomFilter*(data: seq[byte]): Result[BloomFilter, ReliabilityError] =
if data.len == 0:
return err(ReliabilityError.reDeserializationError)
let pb = initProtoBuffer(data)
var bytes: seq[byte]
var cap, errRate, kHashes, mBits: uint64
try:
let
field1_Ok = pb.getField(1, bytes).valueOr:
return err(ReliabilityError.reDeserializationError)
field2_Ok = pb.getField(2, cap).valueOr:
return err(ReliabilityError.reDeserializationError)
field3_Ok = pb.getField(3, errRate).valueOr:
return err(ReliabilityError.reDeserializationError)
field4_Ok = pb.getField(4, kHashes).valueOr:
return err(ReliabilityError.reDeserializationError)
field5_Ok = pb.getField(5, mBits).valueOr:
return err(ReliabilityError.reDeserializationError)
if not field1_Ok or not field2_Ok or not field3_Ok or not field4_Ok or not field5_Ok:
return err(ReliabilityError.reDeserializationError)
var intArray = newSeq[int](bytes.len div sizeof(int))
let pb = Protobuf.decode(data, BloomFilterPB)
let rawData = pb.data.valueOr(@[])
var intArray = newSeq[int](rawData.len div sizeof(int))
for i in 0 ..< intArray.len:
var leVal: int
let start = i * sizeof(int)
copyMem(addr leVal, unsafeAddr bytes[start], sizeof(int))
copyMem(addr leVal, unsafeAddr rawData[i * sizeof(int)], sizeof(int))
littleEndian64(addr intArray[i], addr leVal)
return ok(
BloomFilter.init(
capacity = int(cap),
errorRate = float(errRate) / 1_000_000,
kHashes = int(kHashes),
mBits = int(mBits),
capacity = int(pb.capacity.valueOr(0'u64)),
errorRate = float(pb.errorRate.valueOr(0'u64)) / 1_000_000,
kHashes = int(pb.kHashes.valueOr(0'u64)),
mBits = int(pb.mBits.valueOr(0'u64)),
intArray = intArray,
)
)
except:
except CatchableError:
return err(ReliabilityError.reDeserializationError)
{.pop.}

168
sds/protobufutil.nim
View File

@ -1,19 +1,175 @@
# adapted from https://github.com/waku-org/nwaku/blob/master/waku/common/protobuf.nim
# Minimal hand-rolled protobuf field codec, a thin shim over
# `nim-protobuf-serialization`'s low-level wire `codec` module.
#
# `sds/protobuf.nim` and `sds/snapshot_codec.nim` build messages by hand at the
# field level — including a backward-compatible decode path the type-driven
# `Protobuf.encode/decode` API cannot express, and required-field / always-write
# semantics its default-value omission would break — so this exposes a small
# accumulating `ProtoBuffer` with `write`/`getField`/`getRepeatedField`/`finish`:
# * unsigned integers encode as plain varints (last value wins on decode);
# * strings and byte seqs encode length-delimited, with no UTF-8 validation
# (strings are treated as opaque bytes — message ids may be binary);
# * a field whose stored wire type differs from the requested one is skipped,
# as `protoc` does; only a malformed buffer yields an error.
#
# On construction from bytes the buffer is parsed once, in a single forward pass
# with the library's reader, into per-field value lists; the `getField` accessors
# are then plain lookups rather than re-scanning the buffer for every field.
{.push raises: [].}
import libp2p/protobuf/minprotobuf
import libp2p/varint
import std/tables
import results
import faststreams/inputs
import protobuf_serialization/codec except ProtobufError
import ./types/protobuf_error
export minprotobuf, varint, protobuf_error
export results, protobuf_error
converter toProtobufError*(err: minprotobuf.ProtoError): ProtobufError =
type ProtoBuffer* = object ## Accumulating protobuf field buffer.
buffer*: seq[byte]
## Reads are served from these parse-once indexes (populated by `init(data)`),
## keyed by field number; values are kept in wire order so last-wins / repeated
## semantics fall out directly.
varints: Table[int, seq[uint64]]
lengthDelims: Table[int, seq[seq[byte]]]
parseOk: bool
converter toProtobufError*(err: ProtoError): ProtobufError =
case err
of minprotobuf.ProtoError.RequiredFieldMissing:
of ProtoError.RequiredFieldMissing:
return ProtobufError(kind: ProtobufErrorKind.MissingRequiredField, field: "unknown")
else:
return ProtobufError(kind: ProtobufErrorKind.DecodeFailure, error: err)
proc missingRequiredField*(T: type ProtobufError, field: string): T =
return ProtobufError.init(field)
# ---------------------------------------------------------------------------
# Construction
# ---------------------------------------------------------------------------
proc init*(T: type ProtoBuffer): T =
return T(buffer: @[], parseOk: true)
proc init*(T: type ProtoBuffer, data: seq[byte]): T =
## Parse `data` once into per-field value lists. A malformed buffer leaves
## `parseOk = false`, which every accessor reports as a decode error.
var pb = T(buffer: data, parseOk: true)
var sh = memoryInput(data)
try:
let stream = sh.s
while stream.readable:
let hdr = readHeader(stream)
case hdr.kind
of WireKind.Varint:
pb.varints.mgetOrPut(hdr.number, @[]).add(uint64(readValue(stream, puint64)))
of WireKind.LengthDelim:
pb.lengthDelims.mgetOrPut(hdr.number, @[]).add(seq[byte](readValue(stream, pbytes)))
of WireKind.Fixed64:
skipValue(stream, fixed64)
of WireKind.Fixed32:
skipValue(stream, fixed32)
except CatchableError:
pb.parseOk = false
return pb
proc finish*(pb: var ProtoBuffer) =
## No length prefix is used, so finishing only asserts the invariant that a
## top-level buffer is never empty.
doAssert(pb.buffer.len > 0)
# ---------------------------------------------------------------------------
# Writing
# ---------------------------------------------------------------------------
proc writeVarint(pb: var ProtoBuffer, field: int, value: uint64) =
pb.buffer.add(toBytes(FieldHeader.init(field, WireKind.Varint)))
pb.buffer.add(toBytes(puint64(value)))
proc write*(pb: var ProtoBuffer, field: int, value: uint64) =
pb.writeVarint(field, value)
proc write*(pb: var ProtoBuffer, field: int, value: uint32) =
pb.writeVarint(field, uint64(value))
proc writeLengthDelim(pb: var ProtoBuffer, field: int, data: openArray[byte]) =
pb.buffer.add(toBytes(FieldHeader.init(field, WireKind.LengthDelim)))
pb.buffer.add(toBytes(puint64(uint64(data.len))))
if data.len > 0:
pb.buffer.add(data)
proc write*(pb: var ProtoBuffer, field: int, value: openArray[byte]) =
pb.writeLengthDelim(field, value)
proc write*(pb: var ProtoBuffer, field: int, value: string) =
pb.writeLengthDelim(field, value.toOpenArrayByte(0, value.high))
# ---------------------------------------------------------------------------
# Reading
# ---------------------------------------------------------------------------
proc bytesToString(b: seq[byte]): string =
## Copy raw bytes into a string without UTF-8 validation — protobuf strings
## are opaque bytes here, and message ids may not be valid UTF-8.
var s = newString(b.len)
if b.len > 0:
copyMem(addr s[0], unsafeAddr b[0], b.len)
return s
proc getField*(pb: ProtoBuffer, field: int, output: var uint64): ProtoResult[bool] =
if not pb.parseOk:
return err(ProtoError.VarintDecode)
let values = pb.varints.getOrDefault(field)
if values.len > 0:
output = values[^1]
return ok(true)
return ok(false)
proc getField*(pb: ProtoBuffer, field: int, output: var uint32): ProtoResult[bool] =
if not pb.parseOk:
return err(ProtoError.VarintDecode)
let values = pb.varints.getOrDefault(field)
if values.len > 0:
output = uint32(values[^1])
return ok(true)
return ok(false)
proc getField*(pb: ProtoBuffer, field: int, output: var seq[byte]): ProtoResult[bool] =
if not pb.parseOk:
return err(ProtoError.VarintDecode)
let values = pb.lengthDelims.getOrDefault(field)
if values.len > 0:
output = values[^1]
return ok(true)
return ok(false)
proc getField*(pb: ProtoBuffer, field: int, output: var string): ProtoResult[bool] =
if not pb.parseOk:
return err(ProtoError.VarintDecode)
let values = pb.lengthDelims.getOrDefault(field)
if values.len > 0:
output = bytesToString(values[^1])
return ok(true)
return ok(false)
proc getRepeatedField*(
pb: ProtoBuffer, field: int, output: var seq[seq[byte]]
): ProtoResult[bool] =
if not pb.parseOk:
return err(ProtoError.VarintDecode)
output = pb.lengthDelims.getOrDefault(field)
return ok(output.len > 0)
proc getRepeatedField*(
pb: ProtoBuffer, field: int, output: var seq[string]
): ProtoResult[bool] =
if not pb.parseOk:
return err(ProtoError.VarintDecode)
let values = pb.lengthDelims.getOrDefault(field)
output.setLen(0)
for v in values:
output.add(bytesToString(v))
return ok(output.len > 0)
{.pop.}

71
sds/snapshot_codec.nim
View File

@ -13,7 +13,6 @@
{.push raises: [].}
import std/[sets, times]
import libp2p/protobuf/minprotobuf
import ./types/[
channel_meta, history_update, sds_message, sds_message_id, history_entry,
unacknowledged_message, incoming_message, repair_entry, reliability_error,
@ -44,20 +43,19 @@ proc fromUnixMs(ms: int64): Time =
# ---------------------------------------------------------------------------
proc encodeUnacked(u: UnacknowledgedMessage): ProtoBuffer =
var pb = initProtoBuffer()
let msgPb = wire.encode(u.message)
pb.write(1, msgPb.buffer)
var pb = ProtoBuffer.init()
pb.write(1, wire.serializeMessage(u.message).get())
pb.write(2, uint64(u.sendTime.toUnixMs))
pb.write(3, uint32(u.resendAttempts))
pb.finish()
pb
proc decodeUnacked(buf: seq[byte]): ProtobufResult[UnacknowledgedMessage] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var msgBytes: seq[byte]
if not ?pb.getField(1, msgBytes):
return err(ProtobufError.missingRequiredField("UnacknowledgedMessage.message"))
let msg = SdsMessage.decode(msgBytes).valueOr:
let msg = wire.deserializeMessage(msgBytes).valueOr:
return err(ProtobufError.missingRequiredField("UnacknowledgedMessage.message"))
var sendMs: uint64
if not ?pb.getField(2, sendMs):
@ -77,20 +75,19 @@ proc decodeUnacked(buf: seq[byte]): ProtobufResult[UnacknowledgedMessage] =
# ---------------------------------------------------------------------------
proc encodeIncoming(m: IncomingMessage): ProtoBuffer =
var pb = initProtoBuffer()
let msgPb = wire.encode(m.message)
pb.write(1, msgPb.buffer)
var pb = ProtoBuffer.init()
pb.write(1, wire.serializeMessage(m.message).get())
for dep in m.missingDeps:
pb.write(2, dep) # SdsMessageID is string
pb.finish()
pb
proc decodeIncoming(buf: seq[byte]): ProtobufResult[IncomingMessage] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var msgBytes: seq[byte]
if not ?pb.getField(1, msgBytes):
return err(ProtobufError.missingRequiredField("IncomingMessage.message"))
let msg = SdsMessage.decode(msgBytes).valueOr:
let msg = wire.deserializeMessage(msgBytes).valueOr:
return err(ProtobufError.missingRequiredField("IncomingMessage.message"))
var deps: seq[SdsMessageID]
discard pb.getRepeatedField(2, deps)
@ -104,20 +101,19 @@ proc decodeIncoming(buf: seq[byte]): ProtobufResult[IncomingMessage] =
# ---------------------------------------------------------------------------
proc encodeOutRepairEntry(e: OutgoingRepairEntry): ProtoBuffer =
var pb = initProtoBuffer()
let histPb = wire.encodeHistoryEntry(e.outHistEntry)
pb.write(1, histPb.buffer)
var pb = ProtoBuffer.init()
pb.write(1, wire.serializeHistoryEntry(e.outHistEntry).get())
pb.write(2, uint64(e.minTimeRepairReq.toUnixMs))
pb.finish()
pb
proc decodeOutRepairEntry(buf: seq[byte]): ProtobufResult[OutgoingRepairEntry] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var histBytes: seq[byte]
if not ?pb.getField(1, histBytes):
return err(ProtobufError.missingRequiredField("OutgoingRepairEntry.outHistEntry"))
let histPb = initProtoBuffer(histBytes)
let entry = ?wire.decodeHistoryEntry(histPb)
let entry = wire.deserializeHistoryEntry(histBytes).valueOr:
return err(ProtobufError.missingRequiredField("HistoryEntry"))
var ms: uint64
if not ?pb.getField(2, ms):
return err(ProtobufError.missingRequiredField("OutgoingRepairEntry.minTimeRepairReq"))
@ -128,7 +124,7 @@ proc decodeOutRepairEntry(buf: seq[byte]): ProtobufResult[OutgoingRepairEntry] =
)
proc encodeOutRepairKV(kv: OutgoingRepairKV): ProtoBuffer =
var pb = initProtoBuffer()
var pb = ProtoBuffer.init()
pb.write(1, kv.messageId)
let entryPb = encodeOutRepairEntry(kv.entry)
pb.write(2, entryPb.buffer)
@ -136,7 +132,7 @@ proc encodeOutRepairKV(kv: OutgoingRepairKV): ProtoBuffer =
pb
proc decodeOutRepairKV(buf: seq[byte]): ProtobufResult[OutgoingRepairKV] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var msgId: SdsMessageID
if not ?pb.getField(1, msgId):
return err(ProtobufError.missingRequiredField("OutgoingRepairKV.messageId"))
@ -151,21 +147,20 @@ proc decodeOutRepairKV(buf: seq[byte]): ProtobufResult[OutgoingRepairKV] =
# ---------------------------------------------------------------------------
proc encodeInRepairEntry(e: IncomingRepairEntry): ProtoBuffer =
var pb = initProtoBuffer()
let histPb = wire.encodeHistoryEntry(e.inHistEntry)
pb.write(1, histPb.buffer)
var pb = ProtoBuffer.init()
pb.write(1, wire.serializeHistoryEntry(e.inHistEntry).get())
pb.write(2, e.cachedMessage)
pb.write(3, uint64(e.minTimeRepairResp.toUnixMs))
pb.finish()
pb
proc decodeInRepairEntry(buf: seq[byte]): ProtobufResult[IncomingRepairEntry] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var histBytes: seq[byte]
if not ?pb.getField(1, histBytes):
return err(ProtobufError.missingRequiredField("IncomingRepairEntry.inHistEntry"))
let histPb = initProtoBuffer(histBytes)
let entry = ?wire.decodeHistoryEntry(histPb)
let entry = wire.deserializeHistoryEntry(histBytes).valueOr:
return err(ProtobufError.missingRequiredField("HistoryEntry"))
var cached: seq[byte]
if not ?pb.getField(2, cached):
return err(ProtobufError.missingRequiredField("IncomingRepairEntry.cachedMessage"))
@ -181,7 +176,7 @@ proc decodeInRepairEntry(buf: seq[byte]): ProtobufResult[IncomingRepairEntry] =
)
proc encodeInRepairKV(kv: IncomingRepairKV): ProtoBuffer =
var pb = initProtoBuffer()
var pb = ProtoBuffer.init()
pb.write(1, kv.messageId)
let entryPb = encodeInRepairEntry(kv.entry)
pb.write(2, entryPb.buffer)
@ -189,7 +184,7 @@ proc encodeInRepairKV(kv: IncomingRepairKV): ProtoBuffer =
pb
proc decodeInRepairKV(buf: seq[byte]): ProtobufResult[IncomingRepairKV] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var msgId: SdsMessageID
if not ?pb.getField(1, msgId):
return err(ProtobufError.missingRequiredField("IncomingRepairKV.messageId"))
@ -204,7 +199,7 @@ proc decodeInRepairKV(buf: seq[byte]): ProtobufResult[IncomingRepairKV] =
# ---------------------------------------------------------------------------
proc encode*(meta: ChannelMeta): ProtoBuffer =
var pb = initProtoBuffer()
var pb = ProtoBuffer.init()
pb.write(1, meta.schemaVersion)
pb.write(2, uint64(meta.lamportTimestamp))
for u in meta.outgoingBuffer:
@ -223,7 +218,7 @@ proc encode*(meta: ChannelMeta): ProtoBuffer =
pb
proc decode*(T: type ChannelMeta, buf: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var meta = ChannelMeta.init()
var ver: uint32
@ -271,17 +266,16 @@ proc deserializeChannelMeta*(
# ---------------------------------------------------------------------------
proc encode*(d: ChannelData): ProtoBuffer =
var pb = initProtoBuffer()
var pb = ProtoBuffer.init()
let metaPb = encode(d.meta)
pb.write(1, metaPb.buffer)
for m in d.messageHistory:
let msgPb = wire.encode(m)
pb.write(2, msgPb.buffer)
pb.write(2, wire.serializeMessage(m).get())
pb.finish()
pb
proc decode*(T: type ChannelData, buf: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var d = ChannelData.init()
var metaBytes: seq[byte]
if not ?pb.getField(1, metaBytes):
@ -290,7 +284,7 @@ proc decode*(T: type ChannelData, buf: seq[byte]): ProtobufResult[T] =
var histBufs: seq[seq[byte]]
discard pb.getRepeatedField(2, histBufs)
for b in histBufs:
let m = SdsMessage.decode(b).valueOr:
let m = wire.deserializeMessage(b).valueOr:
return err(ProtobufError.missingRequiredField("ChannelData.messageHistory[i]"))
d.messageHistory.add(m)
ok(d)
@ -300,22 +294,21 @@ proc decode*(T: type ChannelData, buf: seq[byte]): ProtobufResult[T] =
# ---------------------------------------------------------------------------
proc encode*(u: HistoryUpdate): ProtoBuffer =
var pb = initProtoBuffer()
var pb = ProtoBuffer.init()
for m in u.append:
let msgPb = wire.encode(m)
pb.write(1, msgPb.buffer)
pb.write(1, wire.serializeMessage(m).get())
for id in u.evict:
pb.write(2, id)
pb.finish()
pb
proc decode*(T: type HistoryUpdate, buf: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buf)
let pb = ProtoBuffer.init(buf)
var u = HistoryUpdate.init()
var appBufs: seq[seq[byte]]
discard pb.getRepeatedField(1, appBufs)
for b in appBufs:
let m = SdsMessage.decode(b).valueOr:
let m = wire.deserializeMessage(b).valueOr:
return err(ProtobufError.missingRequiredField("HistoryUpdate.append[i]"))
u.append.add(m)
var ev: seq[SdsMessageID]

17
sds/types/protobuf_error.nim
View File

@ -1,7 +1,18 @@
import results
import libp2p/protobuf/minprotobuf
type
ProtoError* {.pure.} = enum
## Low-level protobuf wire decode errors surfaced by the field codec in
## `sds/protobufutil.nim`.
VarintDecode
MessageIncomplete
BufferOverflow
BadWireType
IncorrectBlob
RequiredFieldMissing
ProtoResult*[T] = Result[T, ProtoError]
ProtobufErrorKind* {.pure.} = enum
DecodeFailure
MissingRequiredField
@ -9,13 +20,13 @@ type
ProtobufError* = object
case kind*: ProtobufErrorKind
of DecodeFailure:
error*: minprotobuf.ProtoError
error*: ProtoError
of MissingRequiredField:
field*: string
ProtobufResult*[T] = Result[T, ProtobufError]
proc init*(T: type ProtobufError, error: minprotobuf.ProtoError): T =
proc init*(T: type ProtobufError, error: ProtoError): T =
return T(kind: ProtobufErrorKind.DecodeFailure, error: error)
proc init*(T: type ProtobufError, field: string): T =