From 444ce79f2b0c7bc4821bcdaca1ad1ee11262e969 Mon Sep 17 00:00:00 2001 From: Gabriel Cruz Date: Wed, 1 Jul 2026 12:27:38 -0300 Subject: [PATCH] feat: configuratble request timeout --- CHANGELOG.md | 10 ++++ ffi/codegen/meta.nim | 28 ++++++++++ ffi/ffi_context.nim | 14 +++++ ffi/ffi_thread.nim | 47 ++++++++++++++++- ffi/ffi_thread_request.nim | 31 +++++++++--- ffi/ffi_types.nim | 6 +++ ffi/internal/ffi_macro.nim | 71 +++++++++++++++++++++++--- tests/unit/test_abi_format.nim | 31 ++++++++++++ tests/unit/test_ffi_context.nim | 90 +++++++++++++++++++++++++++++++++ 9 files changed, 314 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 04e811d..405b201 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,16 @@ All notable changes to this project are documented in this file. `nimble genbindings_c` / `genbindings_c_echo` / `check_bindings_c` / `test_c_e2e` tasks, a `tests/e2e/c` ctest harness, and a `tests/unit/test_c_codegen.nim` unit suite. +- Configurable per-request handler timeout with a finite default: each + `FFIContext` now carries a `defaultRequestTimeout` (5s) applied to every + handler, replacing the previous unbounded wait so a wedged handler can no + longer hang a foreign caller forever. On trip the caller is unblocked with an + `ffi request timed out after ms` err; the handler is left running (not + cancelled, since a hard-cancel mid-call into the underlying library can leave + it partial), and the callback still fires exactly once. Override per proc with + a `"timeout = "` spec (e.g. `{.ffi: "timeout = 30000".}`), parsed like the + `abi = ...` spec; runtime-only, codegen ignores it + ([#93](https://github.com/logos-messaging/nim-ffi/issues/93)). - Per-interaction ABI-format annotations: `declareLibrary` now takes an optional `defaultABIFormat` (`"cbor"` default, or `"c"`) that every `{.ffi.}` / `{.ffiCtor.}` / `{.ffiDtor.}` / `{.ffiRaw.}` / `{.ffiEvent.}` diff --git a/ffi/codegen/meta.nim b/ffi/codegen/meta.nim index d62b14b..6379b2f 100644 --- a/ffi/codegen/meta.nim +++ b/ffi/codegen/meta.nim @@ -67,6 +67,34 @@ proc abiCodegenImplemented*(fmt: ABIFormat): bool = ## seam a future PR flips once the `c` dispatch path is wired. fmt == ABIFormat.Cbor +proc specKey*(spec: string): string = + ## Lowercased key of a `key = value` annotation spec (the text before `=`), + ## used to route a spec to its parser. `"timeout = 30000"` → `"timeout"`. + spec.split('=')[0].strip().toLowerAscii() + +proc parseTimeoutSpec*(spec: string): tuple[ok: bool, ms: int, err: string] = + ## Parse a `"timeout = "` override (whitespace/case tolerant). + ## The value must be a positive integer number of milliseconds. On bad + ## grammar or value, returns `ok = false` with a human-readable `err`. + let parts = spec.split('=') + if parts.len != 2 or specKey(spec) != "timeout": + return + (false, 0, "invalid timeout override '" & spec & "'; expected `timeout = `") + let raw = parts[1].strip() + let ms = + try: + parseInt(raw) + except ValueError: + return ( + false, + 0, + "invalid timeout value '" & raw & + "'; expected a positive integer of milliseconds", + ) + if ms <= 0: + return (false, 0, "timeout must be a positive number of milliseconds, got: " & raw) + (true, ms, "") + proc parseABIFormatName*(name: string): tuple[ok: bool, fmt: ABIFormat] = ## Bare format name (`"c"`/`"cbor"`, case-insensitive) → `ABIFormat`; ## `ok` is false otherwise. diff --git a/ffi/ffi_context.nim b/ffi/ffi_context.nim index 422769f..8f80631 100644 --- a/ffi/ffi_context.nim +++ b/ffi/ffi_context.nim @@ -39,6 +39,15 @@ type FFIContext*[T] = object eventQueueStuck*: Atomic[bool] # sticky overflow flag running: Atomic[bool] # To control when the threads are running registeredRequests: ptr Table[cstring, FFIRequestProc] + requestTimeouts: ptr Table[cstring, int] + # Per-proc timeout overrides (ms). Points at the compile-time-filled global, + # like registeredRequests, so the FFI thread reads it GC-safely via ctx. + defaultRequestTimeout*: Duration + # Deadline applied to each handler unless a `{.ffi: "timeout = ".}` + # override raises it. On trip the caller is unblocked with a timeout err and + # the handler is left running (see processRequest). Set `InfiniteDuration` + # to opt out. Written on the owning thread before the first request; read on + # the FFI thread. var onFFIThread* {.threadvar.}: bool # Re-entrant dispatch guard for `sendRequestToFFIThread`. @@ -49,6 +58,9 @@ const EventThreadTickInterval* = 1.seconds FFIHeartbeatStartDelay* = 10.seconds # grace window for library startup FFIHeartbeatStaleThreshold* = 1.seconds + DefaultRequestTimeout* = 5.seconds + # A guess (issue #93): finite so a wedged handler can't hang a caller + # forever, generous enough to clear normal handlers. Overridable per proc. include ./event_thread include ./ffi_thread @@ -105,6 +117,7 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] = initEventQueue(ctx[].eventQueue) ctx.ffiHeartbeat.store(0) ctx.eventQueueStuck.store(false) + ctx.defaultRequestTimeout = DefaultRequestTimeout var success = false defer: @@ -120,6 +133,7 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] = newSignalOrErr(ctx.eventThreadExitSignal, "eventThreadExitSignal") ctx.registeredRequests = addr ffi_types.registeredRequests + ctx.requestTimeouts = addr ffi_types.requestTimeoutsMs ctx.running.store(true) diff --git a/ffi/ffi_thread.nim b/ffi/ffi_thread.nim index 3ec3c3a..9676918 100644 --- a/ffi/ffi_thread.nim +++ b/ffi/ffi_thread.nim @@ -44,6 +44,45 @@ proc sendRequestToFFIThread*( ok() +func resolveRequestTimeout[T](reqIdCs: cstring, ctx: ptr FFIContext[T]): Duration = + ## Per-proc `{.ffi: "timeout = ".}` override if one was registered for this + ## request type, otherwise the context-wide default. + let ms = ctx[].requestTimeouts[].getOrDefault(reqIdCs, 0) + if ms > 0: ms.milliseconds else: ctx.defaultRequestTimeout + +proc reportTimeoutIfTripped( + retFut: Future[Result[seq[byte], string]], + request: ptr FFIThreadRequest, + deadline: Duration, + reqId: string, +) {.async.} = + ## Waits for the handler or its deadline, whichever comes first. On a trip we + ## deliberately do NOT cancel the handler: a hard-cancel mid-call into the + ## underlying library (Waku/libp2p) can leave it partially applied, so we + ## unblock the caller with a timeout err now and let the handler run to + ## completion. `respondOnce` keeps the two paths from answering twice. + if deadline == InfiniteDuration: + return + # Handlers that already completed (e.g. a sync body) skip the timer entirely, + # keeping the per-request cost off the fast path. + if retFut.finished(): + return + let timer = sleepAsync(deadline) + # `race` returns the first to finish WITHOUT cancelling the loser, so the + # handler keeps running when the timer wins. + discard await race(retFut, timer) + if not timer.finished(): + await timer.cancelAndWait() + if retFut.finished(): + return + warn "ffi request timed out; caller unblocked, handler left running", + reqId = reqId, timeoutMs = deadline.milliseconds + request.respondOnce( + Result[seq[byte], string].err( + "ffi request timed out after " & $deadline.milliseconds & "ms" + ) + ) + proc processRequest[T]( request: ptr FFIThreadRequest, ctx: ptr FFIContext[T] ) {.async.} = @@ -59,9 +98,15 @@ proc processRequest[T]( else: ctx[].registeredRequests[][reqIdCs](cast[pointer](request), ctx) - # CatchableError covers CancelledError from the shutdown drain; handleRes must still run. + # CatchableError covers CancelledError from the shutdown drain; handleRes must + # still run, so the timeout race and the handler await share one try — a cancel + # mid-race must not skip the response-and-free below. let res = try: + # May answer the caller early with a timeout err; the handler keeps running. + await reportTimeoutIfTripped( + retFut, request, resolveRequestTimeout(reqIdCs, ctx), reqId + ) await retFut except CatchableError as e: Result[seq[byte], string].err( diff --git a/ffi/ffi_thread_request.nim b/ffi/ffi_thread_request.nim index b6b8daf..cf3e6bf 100644 --- a/ffi/ffi_thread_request.nim +++ b/ffi/ffi_thread_request.nim @@ -30,6 +30,9 @@ type FFIThreadRequest* = object ## Intrusive ingress-queue link (see `ffi_request_queue.nim`). Touched only ## under the queue's lock; the request doubles as its own node, so no ## separate node alloc lands on the per-thread ORC MemRegion. + responded*: bool + ## De-duplicates the callback across the timeout and completion paths. Both + ## run on the FFI thread, so a plain flag suffices — no cross-thread race. proc allocBaseRequest( callback: FFICallBack, userData: pointer, reqId: cstring @@ -44,6 +47,7 @@ proc allocBaseRequest( ret[].data = nil ret[].dataLen = 0 ret[].next = nil + ret[].responded = false return ret proc copySharedPayload(req: ptr FFIThreadRequest, data: ptr byte, dataLen: int) = @@ -125,12 +129,9 @@ proc deleteRequest*(request: ptr FFIThreadRequest) = c_free(cast[pointer](request[].reqId)) c_free(request) -proc handleRes*(res: Result[seq[byte], string], request: ptr FFIThreadRequest) = - ## Fires the registered callback exactly once and frees the request. - ## Success payload is CBOR bytes; error payload is the raw UTF-8 error string. - defer: - deleteRequest(request) - +proc fireCallback(res: Result[seq[byte], string], request: ptr FFIThreadRequest) = + ## Delivers one response to the foreign callback. Success payload is CBOR + ## bytes; error payload is the raw UTF-8 error string. if res.isErr(): foreignThreadGc: let msg = if res.error.len > 0: res.error else: EmptyErrorMarker @@ -155,5 +156,23 @@ proc handleRes*(res: Result[seq[byte], string], request: ptr FFIThreadRequest) = RET_OK, cast[ptr cchar](addr sentinel), 1.csize_t, request[].userData ) +proc respondOnce*(request: ptr FFIThreadRequest, res: Result[seq[byte], string]) = + ## Fires the callback the first time it's called for `request` and no-ops + ## after — the timeout path and the handler-completion path both call it, but + ## the foreign side must be answered exactly once. Does NOT free the request: + ## freeing stays with `handleRes` so the handler always owns the buffer until + ## it finishes. + if request[].responded: + return + request[].responded = true + fireCallback(res, request) + +proc handleRes*(res: Result[seq[byte], string], request: ptr FFIThreadRequest) = + ## Terminal step of every request: delivers the response (unless a timeout + ## already did) and frees the request exactly once. + defer: + deleteRequest(request) + respondOnce(request, res) + proc nilProcess*(reqId: cstring): Future[Result[seq[byte], string]] {.async.} = return err("This request type is not implemented: " & $reqId) diff --git a/ffi/ffi_types.nim b/ffi/ffi_types.nim index 0029ec2..a3add13 100644 --- a/ffi/ffi_types.nim +++ b/ffi/ffi_types.nim @@ -37,5 +37,11 @@ template foreignThreadGc*(body: untyped) = ## The value is a proc that handles the request asynchronously. var registeredRequests*: Table[cstring, FFIRequestProc] +## Per-request handler-timeout overrides in milliseconds, keyed by the same Req +## type name as `registeredRequests`. Populated at compile time from a +## `{.ffi: "timeout = ".}` spec; an absent key means "use the context's +## `defaultRequestTimeout`". Like `registeredRequests`, never mutated at run time. +var requestTimeoutsMs*: Table[cstring, int] + ### End of FFI utils ################################################################################ diff --git a/ffi/internal/ffi_macro.nim b/ffi/internal/ffi_macro.nim index 0d3b133..66e341f 100644 --- a/ffi/internal/ffi_macro.nim +++ b/ffi/internal/ffi_macro.nim @@ -33,6 +33,50 @@ proc resolveABIFormat(abiSpecs: seq[NimNode]): ABIFormat {.compileTime.} = fmt = parsed.fmt fmt +proc resolveFFISpecs( + specs: seq[NimNode] +): tuple[abi: ABIFormat, timeoutMs: int] {.compileTime.} = + ## Resolve an annotation's `"abi = ..."` and `"timeout = ..."` string specs + ## (last of each wins), inheriting the library-default ABI when absent. + ## `timeoutMs == 0` means "no per-proc override" (use the context default). + var abi = currentDefaultABIFormat + var timeoutMs = 0 + for spec in specs: + if spec.kind notin {nnkStrLit, nnkRStrLit, nnkTripleStrLit}: + error( + "FFI override must be a string literal like \"abi = c\" or " & + "\"timeout = 30000\", got: " & spec.repr + ) + case specKey($spec) + of "abi": + let parsed = parseAbiSpec($spec) + if not parsed.ok: + error(parsed.err) + abi = parsed.fmt + of "timeout": + let parsed = parseTimeoutSpec($spec) + if not parsed.ok: + error(parsed.err) + timeoutMs = parsed.ms + else: + error( + "unknown FFI override '" & $spec & "'; expected `abi = ...` or `timeout = ...`" + ) + (abi, timeoutMs) + +proc registerRequestTimeout( + reqTypeName: NimNode, timeoutMs: int +): NimNode {.compileTime.} = + ## Top-level assignment that records a per-proc handler timeout at module init, + ## keyed by the same Req type name the dispatcher registry uses. Empty when no + ## override was given. + if timeoutMs <= 0: + return newStmtList() + newAssignment( + newTree(nnkBracketExpr, ident("requestTimeoutsMs"), newLit($reqTypeName)), + newLit(timeoutMs), + ) + proc gateABIFormat(fmt: ABIFormat, where: string) {.compileTime.} = ## Abort if the selected ABI's codegen isn't wired yet (only `Cbor` is), so a ## `c` request fails loudly instead of emitting CBOR mislabeled as C. @@ -604,7 +648,8 @@ macro ffiRaw*(args: varargs[untyped]): untyped = requireLibraryDeclared("`.ffiRaw.`") let prc = args[^1] - gateABIFormat(resolveABIFormat(args[0 ..^ 2]), "`.ffiRaw.` proc") + let (rawAbiFormat, rawTimeoutMs) = resolveFFISpecs(args[0 ..^ 2]) + gateABIFormat(rawAbiFormat, "`.ffiRaw.` proc") let procName = prc[0] let formalParams = prc[3] @@ -677,7 +722,8 @@ macro ffiRaw*(args: varargs[untyped]): untyped = registerReqFFI(`reqName`, `paramIdent`: `paramType`): `anonymousProcNode` - let stmts = newStmtList(registerReq, ffiProc) + let stmts = + newStmtList(registerReq, ffiProc, registerRequestTimeout(reqName, rawTimeoutMs)) when defined(ffiDumpMacros): echo stmts.repr @@ -747,14 +793,16 @@ macro ffi*(args: varargs[untyped]): untyped = ## proc mylib_send*(w: MyLib, cfg: SendConfig): Future[Result[string, string]] {.ffi.} = ## return ok("done") - # Annotated node is the last vararg; leading args are `"abi = ..."` specs. + # Annotated node is the last vararg; leading args are override specs. let prc = args[^1] - let abiFormat = resolveABIFormat(args[0 ..^ 2]) + let (abiFormat, timeoutMs) = resolveFFISpecs(args[0 ..^ 2]) # A value type stands alone (no library required). Its `c` companion is # emitted later by `genBindings()`, since a type-pragma macro can only return # a TypeDef; `cbor` rides the generic overloads. Both abis are valid here. if prc.kind == nnkTypeDef: + if timeoutMs > 0: + error("`.ffi.` on a type takes no `timeout` override (it applies to procs)") gateFFITypeABIFormat(abiFormat, "`.ffi.` type") var cleanTypeDef = prc.copyNimTree() if cleanTypeDef[0].kind == nnkPragmaExpr: @@ -988,7 +1036,9 @@ macro ffi*(args: varargs[untyped]): untyped = ) ) - return newStmtList(helperProc, registerReq, ffiProc) + return newStmtList( + helperProc, registerReq, ffiProc, registerRequestTimeout(reqTypeName, timeoutMs) + ) let stmts = asyncPath() @@ -1238,7 +1288,7 @@ macro ffiCtor*(args: varargs[untyped]): untyped = requireLibraryDeclared("`.ffiCtor.`") let prc = args[^1] - let abiFormat = resolveABIFormat(args[0 ..^ 2]) + let (abiFormat, timeoutMs) = resolveFFISpecs(args[0 ..^ 2]) gateABIFormat(abiFormat, "`.ffiCtor.` proc") let procName = prc[0] @@ -1408,7 +1458,14 @@ macro ffiCtor*(args: varargs[untyped]): untyped = var `poolIdent`: FFIContextPool[`libTypeName`] let stmts = newStmtList( - typeDef, ffiNewReqProc, helperProc, processProc, addToReg, poolDecl, ffiProc + typeDef, + ffiNewReqProc, + helperProc, + processProc, + addToReg, + poolDecl, + ffiProc, + registerRequestTimeout(reqTypeName, timeoutMs), ) when defined(ffiDumpMacros): diff --git a/tests/unit/test_abi_format.nim b/tests/unit/test_abi_format.nim index ec2ecac..d41df1f 100644 --- a/tests/unit/test_abi_format.nim +++ b/tests/unit/test_abi_format.nim @@ -38,6 +38,13 @@ proc abitest_echo*( ): Future[Result[int, string]] {.ffi: "abi = cbor".} = return ok(n) +# Per-proc handler-timeout override (issue #93): parsed like the abi spec and +# recorded in `requestTimeoutsMs`, keyed by the generated Req type name. +proc abitest_slow*( + lib: AbiLib, n: int +): Future[Result[int, string]] {.ffi: "timeout = 30000".} = + return ok(n) + # Event with an explicit ABI override passed after the wire name. proc abitest_pinged*(p: Pinged) {.ffiEvent("on_pinged", "abi = cbor").} @@ -91,6 +98,30 @@ suite "ABI format parsing": check parseAbiSpec("abi = bson").ok == false # unknown format check "bson" in parseAbiSpec("abi = bson").err +suite "handler-timeout spec parsing (issue #93)": + test "specKey extracts the lowercased, trimmed key": + check specKey("timeout = 30000") == "timeout" + check specKey(" ABI = c ") == "abi" + check specKey("bare") == "bare" + + test "parseTimeoutSpec accepts `timeout = `, flexible spacing": + check parseTimeoutSpec("timeout = 30000") == (true, 30000, "") + check parseTimeoutSpec("TIMEOUT=100").ms == 100 + check parseTimeoutSpec(" timeout = 5 ").ms == 5 + + test "parseTimeoutSpec rejects malformed specs and non-positive values": + check parseTimeoutSpec("30000").ok == false # missing `timeout =` + check parseTimeoutSpec("abi = c").ok == false # wrong key + check parseTimeoutSpec("timeout = 1 = 2").ok == false # too many `=` + check parseTimeoutSpec("timeout = abc").ok == false # not an integer + check parseTimeoutSpec("timeout = 0").ok == false # must be positive + check parseTimeoutSpec("timeout = -5").ok == false # must be positive + + test "a `timeout` override is recorded; a plain proc has no entry": + # Populated at module init from the annotations above. + check requestTimeoutsMs["AbitestSlowReq".cstring] == 30000 + check not requestTimeoutsMs.hasKey("AbitestPingReq".cstring) + suite "ABI proc-dispatch readiness (why c is still gated on procs)": test "cbor proc-dispatch is wired; c proc-dispatch is gated": # This predicate is what the proc-form macros consult: `cbor` is wired diff --git a/tests/unit/test_ffi_context.nim b/tests/unit/test_ffi_context.nim index 0b76845..c2463fd 100644 --- a/tests/unit/test_ffi_context.nim +++ b/tests/unit/test_ffi_context.nim @@ -11,6 +11,7 @@ type CallbackData = object lock: Lock cond: Cond called: bool + callCount: int retCode: cint msg: array[1024, byte] msgLen: int @@ -34,6 +35,7 @@ proc testCallback( copyMem(addr d[].msg[0], msg, n) d[].msgLen = n d[].called = true + inc d[].callCount signal(d[].cond) release(d[].lock) @@ -610,3 +612,91 @@ suite "reentrancy guard (PR #23 review, item 6)": let nestedMsg = gReentrantNestedRes.recv() check nestedMsg.startsWith("err:") check "reentrant ffi call" in nestedMsg + +# Per-proc handler timeout (issue #93): a `{.ffi: "timeout = ".}` override +# bounds how long a handler may run before the caller is unblocked with an err. +# The handler is NOT cancelled — it keeps running — so the callback must still +# fire exactly once. + +type TimeoutConfig {.ffi.} = object + dummy: int + +proc testlib_slow_timeout*( + lib: SimpleLib, cfg: TimeoutConfig +): Future[Result[string, string]] {.ffi: "timeout = 100".} = + await sleepAsync(500.milliseconds) + return ok("slow-timeout-done") + +proc testlib_under_deadline*( + lib: SimpleLib, cfg: TimeoutConfig +): Future[Result[string, string]] {.ffi: "timeout = 1000".} = + await sleepAsync(50.milliseconds) + return ok("under-deadline-done") + +proc createSimpleCtx(): ptr FFIContext[SimpleLib] = + ## Spins up a SimpleLib context via the ctor and returns it (nil on failure). + var ctorD: CallbackData + initCallbackData(ctorD) + defer: + deinitCallbackData(ctorD) + var cfg = cborEncode(TestlibCreateCtorReq(config: SimpleConfig(initialValue: 1))) + let ctorRet = + testlib_create(encodedPtr(cfg), cfg.len.csize_t, testCallback, addr ctorD) + if ctorRet.isNil(): + return nil + waitCallback(ctorD) + if ctorD.retCode != RET_OK: + return nil + let ctxAddr = ctorAddrFromCbor(callbackBytes(ctorD)) + if ctxAddr == 0: + return nil + cast[ptr FFIContext[SimpleLib]](ctxAddr) + +suite "per-proc request timeout (issue #93)": + test "handler past its deadline yields a timeout err, fired exactly once": + let ctx = createSimpleCtx() + check not ctx.isNil() + defer: + check SimpleLibFFIPool.destroyFFIContext(ctx).isOk() + + var d: CallbackData + initCallbackData(d) + defer: + deinitCallbackData(d) + + var reqBytes = cborEncode(TestlibSlowTimeoutReq(cfg: TimeoutConfig(dummy: 0))) + let ret = testlib_slow_timeout( + ctx, testCallback, addr d, encodedPtr(reqBytes), reqBytes.len.csize_t + ) + check ret == RET_OK + + waitCallback(d) + check d.retCode == RET_ERR + check "timed out" in callbackErr(d) + + # The handler (500 ms) is still running past the 100 ms deadline; once it + # finishes it must NOT deliver a second callback. + os.sleep(700) + check d.callCount == 1 + + test "handler finishing under its deadline returns normally": + let ctx = createSimpleCtx() + check not ctx.isNil() + defer: + check SimpleLibFFIPool.destroyFFIContext(ctx).isOk() + + var d: CallbackData + initCallbackData(d) + defer: + deinitCallbackData(d) + + var reqBytes = cborEncode(TestlibUnderDeadlineReq(cfg: TimeoutConfig(dummy: 0))) + let ret = testlib_under_deadline( + ctx, testCallback, addr d, encodedPtr(reqBytes), reqBytes.len.csize_t + ) + check ret == RET_OK + + waitCallback(d) + check d.retCode == RET_OK + check cborDecode(callbackBytes(d), string).value == "under-deadline-done" + check d.callCount == 1