## C-compatible FFI wrapper around the SDS ReliabilityManager. ## ## Built on the `nim-ffi` package's high-level macros: `declareLibrary` emits the ## bootstrap + `sds_set_event_callback`; `{.ffiCtor.}`/`{.ffi.}`/`{.ffiDtor.}` ## generate the C entry points, marshalling parameters and return values as JSON. ## 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 JSON representation, so it is hand-written and ## dispatched to the worker thread to store the provider in a thread-local. import std/[base64, json] import ffi import sds import ./events/[ json_message_ready_event, json_message_sent_event, json_missing_dependencies_event, json_periodic_sync_event, json_repair_ready_event, ] # Bootstrap (pragmas, linker flags, libsdsNimMain, initializeLibrary) plus the # `sds_set_event_callback(ctx, callback, userData)` C export. It also declares # 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: [].} # 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 ################################################################################ ### JSON-marshalled request/response types type SdsConfig* {.ffi.} = object participantId: string ## empty disables SDS-R (see newReliabilityManager) type SdsWrapRequest* {.ffi.} = object message: seq[byte] messageId: string channelId: string type SdsWrapResponse* {.ffi.} = object message: seq[byte] type SdsUnwrapRequest* {.ffi.} = object message: seq[byte] type SdsMarkDependenciesRequest* {.ffi.} = object messageIds: seq[string] channelId: string ################################################################################ ### Constructor — creates the FFI context and the ReliabilityManager. ### ### The event closures run on the worker thread and forward JSON payloads to the ### C callback registered via sds_set_event_callback (dispatchFfiEvent reads the ### per-thread callback state, so no context handle is needed here). 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 let rm = newReliabilityManager(participantId = config.participantId.SdsParticipantID).valueOr: error "Failed creating reliability manager", error = error return err("Failed creating reliability manager: " & $error) let messageReadyCb = proc( messageId: SdsMessageID, channelId: SdsChannelID ) {.gcsafe.} = dispatchFfiEvent("message_ready"): $JsonMessageReadyEvent.new(messageId, channelId) let messageSentCb = proc( messageId: SdsMessageID, channelId: SdsChannelID ) {.gcsafe.} = dispatchFfiEvent("message_sent"): $JsonMessageSentEvent.new(messageId, channelId) let missingDependenciesCb = proc( messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID ) {.gcsafe.} = dispatchFfiEvent("missing_dependencies"): $JsonMissingDependenciesEvent.new(messageId, missingDeps, channelId) let periodicSyncCb = proc() {.gcsafe.} = dispatchFfiEvent("periodic_sync"): $JsonPeriodicSyncEvent.new() let repairReadyCb = proc(message: seq[byte], channelId: SdsChannelID) {.gcsafe.} = dispatchFfiEvent("repair_ready"): $JsonRepairReadyEvent.new(message, 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) return hintBytes return @[] await rm.setCallbacks( messageReadyCb, messageSentCb, missingDependenciesCb, periodicSyncCb, retrievalHintProvider, repairReadyCb, ) return ok(rm) ################################################################################ ### Async methods — each runs its body on the worker thread. 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)) proc sdsUnwrapReceivedMessage*( rm: ReliabilityManager, req: SdsUnwrapRequest ): Future[Result[string, string]] {.ffi.} = # The response carries nested objects (missingDeps) which the framework's # object serializer cannot emit, so the JSON is built by hand and returned as # a string. Shape matches the legacy unwrap response. let (unwrapped, missingDeps, channelId) = ( await unwrapReceivedMessage(rm, req.message) ).valueOr: return err("error processing unwrap request: " & $error) var node = newJObject() node["message"] = %*unwrapped node["channelId"] = %*channelId 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) 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("") 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("") 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("") ################################################################################ ### Destructor — runs library cleanup then tears down the FFI context. proc sdsDestroy*(rm: ReliabilityManager) {.ffiDtor.} = discard ################################################################################ ### Retrieval-hint provider (hand-written: a C function pointer cannot be passed ### as JSON). The setter dispatches a request so the provider is stored in the ### worker thread's thread-local, where sdsCreate's hint closure reads it. proc sdsNoopCallback( callerRet: cint, msg: ptr cchar, len: csize_t, userData: pointer ) {.cdecl, gcsafe, raises: [].} = discard registerReqFFI(SdsSetHintReq, ctx: ptr FFIContext[ReliabilityManager]): proc(cbPtr: pointer, udPtr: pointer): Future[Result[string, string]] {.async.} = sdsRetrievalHintCb = cbPtr sdsRetrievalHintUserData = udPtr return ok("") proc sds_set_retrieval_hint_provider( ctx: ptr FFIContext[ReliabilityManager], callback: SdsRetrievalHintProvider, userData: pointer, ): cint {.dynlib, exportc, cdecl, raises: [].} = initializeLibrary() if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)): return RET_ERR let sendRes = try: ffi_context.sendRequestToFFIThread( ctx, SdsSetHintReq.ffiNewReq( sdsNoopCallback, nil, cast[pointer](callback), userData ), ) except Exception as exc: Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg) if sendRes.isErr(): return RET_ERR return RET_OK # Emit binding metadata (no-op unless -d:ffiGenBindings). Must follow every # {.ffi.}/{.ffiCtor.}/{.ffiDtor.} annotation. genBindings()