feat: configuratble request timeout

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 <n>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 = <ms>"` 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.}`

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 = <milliseconds>"` 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 = <ms>`")
+  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.

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 = <ms>".}`
+    # 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)
 

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 = <ms>".}` 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(

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)

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 = <ms>".}` 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
 ################################################################################

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):

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 = <ms>`, 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

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 = <ms>".}` 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