Skip to content

Commit 05595e2

Browse files
Ivansete-statusclaude
Ivansete-status
and
claude
committed
feat(ffi): target nim-ffi master (v0.2.0) — CBOR + event registry
Port the {.ffi.} wrapper to nim-ffi 0.2.0 (master). 0.2.0 is a breaking redesign over 0.1.4: events move to a per-context multi-listener registry (sds_add_event_listener / sds_remove_event_listener) fired via {.ffiEvent.} emitters, and request/response/event marshalling switches from JSON to CBOR. - libsds.nim: typed {.ffiEvent.} payloads replace the JSON event modules; CBOR handles nesting, so unwrap returns a typed response again. The retrieval-hint provider (a C function pointer, not CBOR-encodable) passes its address as a uint64 via a {.ffi.} method that stores it in a worker-thread threadvar. - pin nim-ffi to master HEAD by commit. The lock version is kept a clean semver ("0.2.0") on purpose: nimble's `#`-prefixed special version in the lock breaks `nimble setup -l` (an unquoted `#` truncates the git path), so only vcsRevision carries the commit. - add the new transitive dep cbor_serialization to the lock and nix/deps.nix. - regenerate libsds.h for the CBOR/registry ABI. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent 252a0d5 commit 05595e2

5 files changed

Lines changed: 188 additions & 122 deletions

File tree

‎library/libsds.h‎

Lines changed: 49 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11

2-
// C API for libsds, built on the nim-ffi framework.
2+
// C API for libsds, built on the nim-ffi framework (v0.2.0+).
33
//
4-
// Parameters and results are marshalled as JSON: each request/response struct
5-
// in library/libsds.nim is a JSON object, passed in via the `*Json` cstring
6-
// argument and returned to the callback as a JSON string. Binary fields
7-
// (message bytes) are JSON arrays of byte values.
4+
// Requests, responses and events are marshalled as CBOR. Request payloads are
5+
// passed as a (reqCbor, reqCborLen) byte buffer; results and events are
6+
// delivered to the callback as a CBOR buffer (msg, len). Each request/response
7+
// struct and event payload is defined in library/libsds.nim. Events are
8+
// wrapped in a CBOR envelope { eventType: <wire name>, payload: <struct> }.
89
#ifndef __libsds__
910
#define __libsds__
1011

@@ -20,48 +21,67 @@
2021
extern "C" {
2122
#endif
2223

23-
// Result/event callback. `msg` is the (JSON) payload of length `len`.
24+
// Result/event callback. `msg` is the CBOR payload of length `len`.
2425
// callerRet is one of the RET_* codes above.
2526
typedef void (*SdsCallBack) (int callerRet, const char* msg, size_t len, void* userData);
2627

2728
// Synchronous provider invoked by SDS-R to fetch a retrieval hint for a
2829
// message id. The implementation allocates `*hint` (and sets `*hintLen`); the
29-
// library takes ownership and frees it with deallocShared.
30+
// library takes ownership and frees it with deallocShared. Registered via
31+
// sds_set_retrieval_hint_provider (see below).
3032
typedef void (*SdsRetrievalHintProvider) (const char* messageId, char** hint, size_t* hintLen, void* userData);
3133

3234

33-
// --- Core API Functions ---
35+
// --- Lifecycle -------------------------------------------------------------
3436

37+
// Create a context + ReliabilityManager. reqCbor encodes SdsConfig
38+
// { participantId: tstr } (empty participantId disables SDS-R). Returns the
39+
// context handle, or NULL on failure; the callback also fires on completion.
40+
void* sds_create(const uint8_t* reqCbor, size_t reqCborLen, SdsCallBack callback, void* userData);
3541

36-
// Create a context + ReliabilityManager. configJson: {"participantId":"..."}
37-
// (empty participantId disables SDS-R). Returns the context handle, or NULL on
38-
// failure. The callback also fires on async completion.
39-
void* sds_create(const char* configJson, SdsCallBack callback, void* userData);
42+
// Tear down the context created by sds_create. Blocks until the worker and
43+
// watchdog threads have joined.
44+
int sds_destroy(void* ctx);
4045

41-
// Register the event callback (message_ready, message_sent,
42-
// missing_dependencies, periodic_sync, repair_ready). Payloads are JSON.
43-
void sds_set_event_callback(void* ctx, SdsCallBack callback, void* userData);
4446

45-
// Register the retrieval-hint provider used by SDS-R.
46-
int sds_set_retrieval_hint_provider(void* ctx, SdsRetrievalHintProvider callback, void* userData);
47+
// --- Events ----------------------------------------------------------------
48+
// Subscribe `callback` to an event by wire name and receive a stable listener
49+
// id (non-zero). Event wire names: "message_ready", "message_sent",
50+
// "missing_dependencies", "periodic_sync", "repair_ready". Subscribe to each
51+
// event separately. Payloads arrive as CBOR { eventType, payload }.
52+
uint64_t sds_add_event_listener(void* ctx, const char* eventName, SdsCallBack callback, void* userData);
4753

48-
// reqJson: {"message":[..bytes..],"messageId":"..","channelId":".."}
49-
// Result JSON: {"message":[..bytes..]}
50-
int sds_wrap_outgoing_message(void* ctx, SdsCallBack callback, void* userData, const char* reqJson);
54+
// Remove a listener by id. Returns 0 on success, non-zero if not found.
55+
int sds_remove_event_listener(void* ctx, uint64_t listenerId);
5156

52-
// reqJson: {"message":[..bytes..]}
53-
// Result JSON: {"message":[..],"channelId":"..","missingDeps":[{"messageId":"..","retrievalHint":"<base64>"}]}
54-
int sds_unwrap_received_message(void* ctx, SdsCallBack callback, void* userData, const char* reqJson);
57+
// Register the SDS-R retrieval-hint provider. reqCbor encodes
58+
// SdsHintProviderRequest { callbackAddr: uint, userDataAddr: uint } — the
59+
// SdsRetrievalHintProvider function pointer and its user-data as integer
60+
// addresses.
61+
int sds_set_retrieval_hint_provider(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
5562

56-
// reqJson: {"messageIds":["..",".."],"channelId":".."}
57-
int sds_mark_dependencies_met(void* ctx, SdsCallBack callback, void* userData, const char* reqJson);
5863

59-
int sds_reset(void* ctx, SdsCallBack callback, void* userData);
64+
// --- Core API Functions ----------------------------------------------------
65+
// Each takes a CBOR-encoded request buffer; the result is delivered to
66+
// `callback` as CBOR.
6067

61-
int sds_start_periodic_tasks(void* ctx, SdsCallBack callback, void* userData);
68+
// reqCbor: SdsWrapRequest { message: bytes, messageId: tstr, channelId: tstr }
69+
// result: SdsWrapResponse { message: bytes }
70+
int sds_wrap_outgoing_message(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
6271

63-
// Tear down the context created by sds_create.
64-
int sds_destroy(void* ctx, SdsCallBack callback, void* userData);
72+
// reqCbor: SdsUnwrapRequest { message: bytes }
73+
// result: SdsUnwrapResponse { message: bytes, channelId: tstr,
74+
// missingDeps: [{ messageId: tstr, retrievalHint: bytes }] }
75+
int sds_unwrap_received_message(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
76+
77+
// reqCbor: SdsMarkDependenciesRequest { messageIds: [tstr], channelId: tstr }
78+
int sds_mark_dependencies_met(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
79+
80+
// reqCbor: empty/unit payload (no fields).
81+
int sds_reset(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
82+
83+
// reqCbor: empty/unit payload (no fields).
84+
int sds_start_periodic_tasks(void* ctx, SdsCallBack callback, void* userData, const uint8_t* reqCbor, size_t reqCborLen);
6585

6686

6787
#ifdef __cplusplus

‎library/libsds.nim‎

Lines changed: 110 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,35 @@
11
## C-compatible FFI wrapper around the SDS ReliabilityManager.
22
##
3-
## Built on the `nim-ffi` package's high-level macros: `declareLibrary` emits the
4-
## bootstrap + `sds_set_event_callback`; `{.ffiCtor.}`/`{.ffi.}`/`{.ffiDtor.}`
5-
## generate the C entry points, marshalling parameters and return values as JSON.
6-
## Exported C names are snake_case (`sds_wrap_outgoing_message`, …); see
7-
## `library/libsds.h`. The Go bindings (sds-go-bindings) must match this API.
3+
## Built on nim-ffi (v0.2.0+): `declareLibrary` emits the bootstrap plus the
4+
## event-listener ABI (`sds_add_event_listener` / `sds_remove_event_listener`);
5+
## `{.ffiCtor.}`/`{.ffi.}`/`{.ffiDtor.}` generate the C entry points; and
6+
## `{.ffiEvent.}` declares library-initiated events. Requests, responses and
7+
## events are marshalled as CBOR (see library/libsds.h). Exported C names are
8+
## snake_case. The Go bindings (sds-go-bindings) must match this API.
89
##
9-
## The one exception is `sds_set_retrieval_hint_provider`: it takes a C function
10-
## pointer, which has no sensible JSON representation, so it is hand-written and
11-
## dispatched to the worker thread to store the provider in a thread-local.
10+
## The one hand-written export is `sds_set_retrieval_hint_provider`: it takes a
11+
## C function pointer (no CBOR representation), so it dispatches a request that
12+
## stores the provider in a worker-thread thread-local.
1213

13-
import std/[base64, json, sequtils]
14+
import std/[sequtils]
1415
import ffi
1516
import sds
16-
import ./events/[
17-
json_message_ready_event, json_message_sent_event, json_missing_dependencies_event,
18-
json_periodic_sync_event, json_repair_ready_event,
19-
]
2017

21-
# Bootstrap (pragmas, linker flags, libsdsNimMain, initializeLibrary) plus the
22-
# `sds_set_event_callback(ctx, callback, userData)` C export.
18+
# Bootstrap + sds_add_event_listener / sds_remove_event_listener.
2319
declareLibrary("sds", ReliabilityManager)
2420

2521
type SdsRetrievalHintProvider* = proc(
2622
messageId: cstring, hint: ptr cstring, hintLen: ptr csize_t, userData: pointer
2723
) {.cdecl, gcsafe, raises: [].}
2824

29-
# The active retrieval-hint provider, stored per worker thread (one thread per
30-
# context). Set by sds_set_retrieval_hint_provider via a dispatched request so
31-
# the write lands on the worker thread, where the manager's hint closure reads
32-
# it during message processing.
25+
# Active retrieval-hint provider, per worker thread (one thread per context).
26+
# Set by sds_set_retrieval_hint_provider through a dispatched request so the
27+
# write lands on the worker thread, where the manager's hint closure reads it.
3328
var sdsRetrievalHintCb {.threadvar.}: pointer
3429
var sdsRetrievalHintUserData {.threadvar.}: pointer
3530

3631
################################################################################
37-
### JSON-marshalled request/response types
32+
### CBOR-marshalled request/response types
3833

3934
type SdsConfig* {.ffi.} = object
4035
participantId: string ## empty disables SDS-R (see newReliabilityManager)
@@ -50,16 +45,59 @@ type SdsWrapResponse* {.ffi.} = object
5045
type SdsUnwrapRequest* {.ffi.} = object
5146
message: seq[byte]
5247

48+
type SdsMissingDep* {.ffi.} = object
49+
messageId: string
50+
retrievalHint: seq[byte]
51+
52+
type SdsUnwrapResponse* {.ffi.} = object
53+
message: seq[byte]
54+
channelId: string
55+
missingDeps: seq[SdsMissingDep]
56+
5357
type SdsMarkDependenciesRequest* {.ffi.} = object
5458
messageIds: seq[string]
5559
channelId: string
5660

61+
################################################################################
62+
### Library-initiated events
63+
###
64+
### Each {.ffiEvent.} proc is an emitter: calling it from a worker-thread
65+
### handler dispatches a CBOR EventEnvelope to every listener subscribed (via
66+
### sds_add_event_listener) to the matching wire name.
67+
68+
type SdsMessageReadyPayload* {.ffi.} = object
69+
messageId: string
70+
channelId: string
71+
72+
type SdsMessageSentPayload* {.ffi.} = object
73+
messageId: string
74+
channelId: string
75+
76+
type SdsMissingDependenciesPayload* {.ffi.} = object
77+
messageId: string
78+
channelId: string
79+
missingDeps: seq[SdsMissingDep]
80+
81+
type SdsPeriodicSyncPayload* {.ffi.} = object
82+
placeholder: bool ## events need a payload type; periodic sync carries no data
83+
84+
type SdsRepairReadyPayload* {.ffi.} = object
85+
message: seq[byte]
86+
channelId: string
87+
88+
proc emitMessageReady*(p: SdsMessageReadyPayload) {.ffiEvent: "message_ready".}
89+
proc emitMessageSent*(p: SdsMessageSentPayload) {.ffiEvent: "message_sent".}
90+
proc emitMissingDependencies*(
91+
p: SdsMissingDependenciesPayload
92+
) {.ffiEvent: "missing_dependencies".}
93+
proc emitPeriodicSync*(p: SdsPeriodicSyncPayload) {.ffiEvent: "periodic_sync".}
94+
proc emitRepairReady*(p: SdsRepairReadyPayload) {.ffiEvent: "repair_ready".}
95+
5796
################################################################################
5897
### Constructor — creates the FFI context and the ReliabilityManager.
5998
###
60-
### The AppCallbacks closures run on the worker thread and forward events to the
61-
### C callback registered via sds_set_event_callback (dispatchFfiEvent reads the
62-
### per-thread callback state, so no context handle is needed here).
99+
### The AppCallbacks closures run on the worker thread; they build typed
100+
### payloads and fire the {.ffiEvent.} emitters, which reach the C listeners.
63101

64102
proc sdsCreate*(
65103
config: SdsConfig
@@ -71,28 +109,39 @@ proc sdsCreate*(
71109
let messageReadyCb = proc(
72110
messageId: SdsMessageID, channelId: SdsChannelID
73111
) {.gcsafe.} =
74-
dispatchFfiEvent("message_ready"):
75-
$JsonMessageReadyEvent.new(messageId, channelId)
112+
{.cast(gcsafe).}:
113+
emitMessageReady(
114+
SdsMessageReadyPayload(messageId: $messageId, channelId: $channelId)
115+
)
76116

77117
let messageSentCb = proc(
78118
messageId: SdsMessageID, channelId: SdsChannelID
79119
) {.gcsafe.} =
80-
dispatchFfiEvent("message_sent"):
81-
$JsonMessageSentEvent.new(messageId, channelId)
120+
{.cast(gcsafe).}:
121+
emitMessageSent(
122+
SdsMessageSentPayload(messageId: $messageId, channelId: $channelId)
123+
)
82124

83125
let missingDependenciesCb = proc(
84126
messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID
85127
) {.gcsafe.} =
86-
dispatchFfiEvent("missing_dependencies"):
87-
$JsonMissingDependenciesEvent.new(messageId, missingDeps, channelId)
128+
{.cast(gcsafe).}:
129+
let deps = missingDeps.mapIt(
130+
SdsMissingDep(messageId: $it.messageId, retrievalHint: it.retrievalHint)
131+
)
132+
emitMissingDependencies(
133+
SdsMissingDependenciesPayload(
134+
messageId: $messageId, channelId: $channelId, missingDeps: deps
135+
)
136+
)
88137

89138
let periodicSyncCb = proc() {.gcsafe.} =
90-
dispatchFfiEvent("periodic_sync"):
91-
$JsonPeriodicSyncEvent.new()
139+
{.cast(gcsafe).}:
140+
emitPeriodicSync(SdsPeriodicSyncPayload(placeholder: false))
92141

93142
let repairReadyCb = proc(message: seq[byte], channelId: SdsChannelID) {.gcsafe.} =
94-
dispatchFfiEvent("repair_ready"):
95-
$JsonRepairReadyEvent.new(message, channelId)
143+
{.cast(gcsafe).}:
144+
emitRepairReady(SdsRepairReadyPayload(message: message, channelId: $channelId))
96145

97146
let retrievalHintProvider = proc(messageId: SdsMessageID): seq[byte] {.gcsafe.} =
98147
if sdsRetrievalHintCb.isNil():
@@ -133,26 +182,18 @@ proc sdsWrapOutgoingMessage*(
133182

134183
proc sdsUnwrapReceivedMessage*(
135184
rm: ReliabilityManager, req: SdsUnwrapRequest
136-
): Future[Result[string, string]] {.ffi.} =
137-
# The response carries nested objects (missingDeps) which the framework's
138-
# object serializer cannot emit, so the JSON is built by hand and returned as
139-
# a string. Shape matches the legacy unwrap response.
185+
): Future[Result[SdsUnwrapResponse, string]] {.ffi.} =
140186
let (unwrapped, missingDeps, channelId) = (
141187
await unwrapReceivedMessage(rm, req.message)
142188
).valueOr:
143189
return err("error processing unwrap request: " & $error)
144190

145-
var node = newJObject()
146-
node["message"] = %*unwrapped
147-
node["channelId"] = %*channelId
148-
var missingDepsNode = newJArray()
149-
for dep in missingDeps:
150-
var depNode = newJObject()
151-
depNode["messageId"] = %*dep.messageId
152-
depNode["retrievalHint"] = %*encode(dep.retrievalHint)
153-
missingDepsNode.add(depNode)
154-
node["missingDeps"] = missingDepsNode
155-
return ok($node)
191+
let deps = missingDeps.mapIt(
192+
SdsMissingDep(messageId: $it.messageId, retrievalHint: it.retrievalHint)
193+
)
194+
return ok(
195+
SdsUnwrapResponse(message: unwrapped, channelId: $channelId, missingDeps: deps)
196+
)
156197

157198
proc sdsMarkDependenciesMet*(
158199
rm: ReliabilityManager, req: SdsMarkDependenciesRequest
@@ -185,44 +226,27 @@ proc sdsDestroy*(rm: ReliabilityManager) {.ffiDtor.} =
185226
discard
186227

187228
################################################################################
188-
### Retrieval-hint provider (hand-written: a C function pointer cannot be passed
189-
### as JSON). The setter dispatches a request so the provider is stored in the
190-
### worker thread's thread-local, where sdsCreate's hint closure reads it.
191-
192-
proc sdsNoopCallback(
193-
callerRet: cint, msg: ptr cchar, len: csize_t, userData: pointer
194-
) {.cdecl, gcsafe, raises: [].} =
195-
discard
196-
197-
registerReqFFI(SdsSetHintReq, ctx: ptr FFIContext[ReliabilityManager]):
198-
proc(cbPtr: pointer, udPtr: pointer): Future[Result[string, string]] {.async.} =
199-
sdsRetrievalHintCb = cbPtr
200-
sdsRetrievalHintUserData = udPtr
201-
return ok("")
202-
203-
proc sds_set_retrieval_hint_provider(
204-
ctx: ptr FFIContext[ReliabilityManager],
205-
callback: SdsRetrievalHintProvider,
206-
userData: pointer,
207-
): cint {.dynlib, exportc, cdecl, raises: [].} =
208-
initializeLibrary()
209-
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
210-
return RET_ERR
211-
212-
let sendRes =
213-
try:
214-
ffi_context.sendRequestToFFIThread(
215-
ctx,
216-
SdsSetHintReq.ffiNewReq(
217-
sdsNoopCallback, nil, cast[pointer](callback), userData
218-
),
219-
)
220-
except Exception as exc:
221-
Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg)
222-
if sendRes.isErr():
223-
return RET_ERR
224-
return RET_OK
229+
### Retrieval-hint provider.
230+
###
231+
### The provider is a C function pointer, which has no CBOR representation, so
232+
### it is passed as integer addresses. The body runs on the worker thread (the
233+
### empty await forces the async path) and stores the pointers in the
234+
### thread-local that sdsCreate's hint closure reads. The caller passes the
235+
### function pointer and user-data as uint64 addresses.
236+
237+
type SdsHintProviderRequest* {.ffi.} = object
238+
callbackAddr: uint64
239+
userDataAddr: uint64
240+
241+
proc sdsSetRetrievalHintProvider*(
242+
rm: ReliabilityManager, req: SdsHintProviderRequest
243+
): Future[Result[string, string]] {.ffi.} =
244+
discard rm
245+
await sleepAsync(chronos.milliseconds(0))
246+
sdsRetrievalHintCb = cast[pointer](req.callbackAddr)
247+
sdsRetrievalHintUserData = cast[pointer](req.userDataAddr)
248+
return ok("")
225249

226250
# Emit binding metadata (no-op unless -d:ffiGenBindings). Must follow every
227-
# {.ffi.}/{.ffiCtor.}/{.ffiDtor.} annotation.
251+
# {.ffi.}/{.ffiCtor.}/{.ffiDtor.}/{.ffiEvent.} annotation.
228252
genBindings()

0 commit comments

Comments
 (0)