feat: recycle pooled FFI contexts; compile-time request ids

Two foreign-host concurrency fixes for refc, both needed so a Go host can
hammer the FFI under load without corrupting Nim's per-thread GC.

1. Compile-time request ids. ffiNewReq / the method + ctor wrappers built
   the request id with `$T` at runtime, allocating a Nim GC string on the
   foreign caller's (often transient) thread. Emit a `cstring` literal of
   the type name instead — no allocation on the caller thread.

2. Recycle pooled contexts instead of destroy/recreate. Restores the
   release/v0.1 model that v0.2 dropped: a pool slot's worker + event
   threads and signal fds are built once and reused. The ffiDtor now
   requests a synchronous recycle (drain in-flight handlers, free the lib,
   clear listeners, release the slot) on the FFI thread, keeping the
   threads alive; createFFIContext reuses an initialised slot. Without this
   every create/destroy churned ~6 signal fds, so fd numbers climbed past
   FD_SETSIZE (1024) and ThreadSignalPtr.waitSync's select() failed with
   EINVAL under create/destroy load.

   Adds CtxLifecycle (Active/RecyclePending/Recycling), ctx-level
   inUse/tryClaim/release/markAsActive, requestRecycle (waits on a new
   recycleDoneSignal), freeLib (refc GC_unref / orc =destroy of ctor-owned
   libs), recycleContext, FFIEventRegistry.clearListeners, and roots the
   ctor-stored ref lib under refc (GC_ref, balanced in freeLib).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

ffi/ffi_context.nim

@@ -2,7 +2,7 @@
 
 {.passc: "-fPIC".}
 
-import std/[atomics, locks, options, tables]
+import std/[atomics, locks, options, sequtils, tables]
 import chronicles, chronos, chronos/threadsync, results
 import
   ./ffi_types,
@@ -15,8 +15,30 @@ import
 
 export ffi_events, ffi_handles
 
+type CtxLifecycle* {.pure.} = enum
+  ## State machine guarding a pooled FFI context (Atomic on FFIContext).
+  ##   Active         -> RecyclePending   when the ffiDtor requests recycle
+  ##   RecyclePending -> Recycling        FFI loop claimed it, draining handlers
+  ##   Recycling      -> Active           createFFIContext reuses the slot
+  Active
+  RecyclePending
+  Recycling
+
 type FFIContext*[T] = object
   myLib*: ptr T # main library object (Waku, LibP2P, SDS, …)
+  myLibRefd*: bool
+    # refc only: true once myLib[] (a ref) has been GC_ref'd to root it against
+    # the cycle collector. Balanced by GC_unref in freeLib.
+  myLibOwned*: bool
+    # true once a ctor stored a createShared'd lib into myLib (vs the worker's
+    # stack fallback). freeLib only frees/destroys owned libs.
+  inUse*: Atomic[bool]
+    # Whether this pooled context is claimed. The recycle handler clears it on
+    # the FFI thread so the slot returns to the pool without recreating threads.
+  lifecycle*: Atomic[CtxLifecycle]
+  recycleDoneSignal: ThreadSignalPtr
+    # fired by the recycle handler once the lib is freed and the slot released;
+    # the synchronous recycleFFIContext caller waits on it.
   ffiThread: Thread[(ptr FFIContext[T])]
   eventThread: Thread[(ptr FFIContext[T])]
   reqQueueBank: RequestQueueBank
@@ -42,6 +64,9 @@ var onFFIThread* {.threadvar.}: bool
 const git_version* {.strdefine.} = "n/a"
 
 const
+  RecycleWaitTimeout* = 5.seconds
+    ## Caller-side bound for synchronous recycle; the FFI-thread drain itself is
+    ## bounded by RecycleTimeout, so this only guards against a wedged worker.
   EventThreadTickInterval* = 1.seconds
   FFIHeartbeatStartDelay* = 10.seconds
   FFIHeartbeatStaleThreshold* = 1.seconds
@@ -73,7 +98,8 @@ proc deinitContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
   deinitHandleRegistry(ctx[].handles)
   deinitEventQueue(ctx[].eventQueue)
   when defined(gcRefc):
-    # ThreadSignalPtr.close() under refc hangs via signal-handler re-entry; leak the bounded fd.
+    # ThreadSignalPtr.close() under refc hangs via signal-handler re-entry; the
+    # recycle pool makes full destroy rare, so the leaked fd stays bounded.
     discard
   else:
     closeAndNil(ctx.reqSignal)
@@ -81,6 +107,7 @@ proc deinitContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
     closeAndNil(ctx.threadExitSignal)
     closeAndNil(ctx.eventQueueSignal)
     closeAndNil(ctx.eventThreadExitSignal)
+    closeAndNil(ctx.recycleDoneSignal)
   ok()
 
 template newSignalOrErr(field: untyped, name: string) =
@@ -95,6 +122,10 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
   ctx.threadExitSignal = nil
   ctx.eventQueueSignal = nil
   ctx.eventThreadExitSignal = nil
+  ctx.recycleDoneSignal = nil
+  ctx.myLibOwned = false
+  ctx.myLibRefd = false
+  ctx.lifecycle.store(CtxLifecycle.Active)
   initRequestQueue(ctx[].reqQueueBank)
   initEventRegistry(ctx[].eventRegistry)
   initHandleRegistry(ctx[].handles)
@@ -117,6 +148,7 @@ proc initContextResources*[T](ctx: ptr FFIContext[T]): Result[void, string] =
   newSignalOrErr(ctx.threadExitSignal, "threadExitSignal")
   newSignalOrErr(ctx.eventQueueSignal, "eventQueueSignal")
   newSignalOrErr(ctx.eventThreadExitSignal, "eventThreadExitSignal")
+  newSignalOrErr(ctx.recycleDoneSignal, "recycleDoneSignal")
 
   ctx.registeredRequests = addr ffi_types.registeredRequests
 
@@ -167,6 +199,40 @@ proc signalStop*[T](ctx: ptr FFIContext[T]): Result[void, string] =
     error "failed to signal eventQueueSignal in signalStop", error = error
   ok()
 
+proc tryClaim*[T](ctx: ptr FFIContext[T]): bool =
+  ## Atomically claim a free pooled context (false -> true).
+  var expected = false
+  ctx.inUse.compareExchange(expected, true)
+
+proc release*[T](ctx: ptr FFIContext[T]) =
+  ctx.inUse.store(false)
+
+proc isInUse*[T](ctx: ptr FFIContext[T]): bool =
+  ctx.inUse.load()
+
+proc markAsActive*[T](ctx: ptr FFIContext[T]) =
+  ## Reused context: its worker threads are still alive; re-arm for requests.
+  ctx.lifecycle.store(CtxLifecycle.Active)
+
+proc requestRecycle*[T](ctx: ptr FFIContext[T]): Result[void, string] =
+  ## Ask the FFI thread to drain, free the lib and release the slot, WITHOUT
+  ## stopping its worker/event threads, so the next createFFIContext reuses them.
+  ## Synchronous: waits on recycleDoneSignal. No fd churn -> no select() limit.
+  var expected = CtxLifecycle.Active
+  if not ctx.lifecycle.compareExchange(expected, CtxLifecycle.RecyclePending):
+    return err("requestRecycle: context is not Active (already recycling)")
+
+  let fired = ctx.reqSignal.fireSync().valueOr:
+    return err("requestRecycle: failed to signal the FFI thread: " & $error)
+  if not fired:
+    return err("requestRecycle: failed to signal the FFI thread in time")
+
+  let done = ctx.recycleDoneSignal.waitSync(RecycleWaitTimeout).valueOr:
+    return err("requestRecycle: failed waiting for recycle: " & $error)
+  if not done:
+    return err("requestRecycle: recycle did not complete in time")
+  ok()
+
 ## Per-thread exit wait before stopAndJoinThreads leaks ctx rather than hanging; async
 ## `{.ffiDtor.}` teardown can outlast the default. Override `-d:ffiThreadExitTimeoutMs=<ms>`.
 const ThreadExitTimeoutMs* {.intdefine: "ffiThreadExitTimeoutMs".} = 1500

ffi/ffi_context_pool.nim

@@ -13,37 +13,43 @@ type
     StaticCtxReady
 
   FFIContextPool*[T] = object
-    ## Fixed pool of FFI contexts, plus the one `{.ffiStatic.}` context.
-    # Each live context holds 5 ThreadSignalPtrs — one fd each on Linux, two (a
-    # socketpair) elsewhere. Under refc a destroyed context cannot close them
-    # (see `deinitContextResources`), so churn leaks fds unbounded.
-    slots: array[MaxFFIContexts, FFIContext[T]]
-    inUse: array[MaxFFIContexts, Atomic[bool]]
+    ## Fixed pool of FFI contexts, plus the one `{.ffiStatic.}` context. Each
+    ## slot's worker + event threads and signal fds are built once (on first
+    ## use) and reused across create/recycle cycles — recycle keeps them alive,
+    ## so repeated create/destroy does not churn fds. Bounds ThreadSignalPtr fds
+    ## at MaxFFIContexts * (signals per ctx).
+    contexts: array[MaxFFIContexts, FFIContext[T]]
+    initialized: array[MaxFFIContexts, Atomic[bool]]
     staticCtx: Atomic[pointer]
     staticState: Atomic[StaticCtxState]
 
-proc acquireSlot[T](pool: var FFIContextPool[T]): Result[ptr FFIContext[T], string] =
-  for i in 0 ..< MaxFFIContexts:
-    var expected = false
-    if pool.inUse[i].compareExchange(expected, true):
-      return ok(pool.slots[i].addr)
-  err("FFI context pool exhausted (max " & $MaxFFIContexts & " contexts)")
-
 proc releaseSlot[T](pool: var FFIContextPool[T], ctx: ptr FFIContext[T]) =
+  ## Full-teardown release: the slot must be rebuilt before it serves again.
   for i in 0 ..< MaxFFIContexts:
-    if pool.slots[i].addr == ctx:
-      pool.inUse[i].store(false)
-      return
+    if pool.contexts[i].addr == ctx:
+      pool.initialized[i].store(false)
+      break
+  ctx.release()
 
 proc createFFIContext*[T](
     pool: var FFIContextPool[T]
 ): Result[ptr FFIContext[T], string] =
-  let ctx = pool.acquireSlot().valueOr:
-    return err("createFFIContext: acquireSlot failed: " & $error)
-  initContextResources(ctx).isOkOr:
-    pool.releaseSlot(ctx)
-    return err("createFFIContext: initContextResources failed: " & $error)
-  ok(ctx)
+  ## Acquires a context from the fixed pool. A slot's worker is built once on
+  ## first use and reused (markAsActive) on every later acquisition.
+  for i in 0 ..< MaxFFIContexts:
+    let ctx = pool.contexts[i].addr
+    if not ctx.tryClaim():
+      continue
+    if pool.initialized[i].load():
+      # Reused slot: a prior recycle drained and released it; worker still alive.
+      ctx.markAsActive()
+      return ok(ctx)
+    initContextResources(ctx).isOkOr:
+      ctx.release()
+      return err("createFFIContext: initContextResources failed: " & $error)
+    pool.initialized[i].store(true)
+    return ok(ctx)
+  err("FFI context pool exhausted (max " & $MaxFFIContexts & " contexts)")
 
 proc isStaticCtx[T](pool: var FFIContextPool[T], ctx: ptr FFIContext[T]): bool =
   ## True while `ctx` is the pool's static context, including mid-teardown.
@@ -51,16 +57,29 @@ proc isStaticCtx[T](pool: var FFIContextPool[T], ctx: ptr FFIContext[T]): bool =
   # pointer covers `Destroying` too.
   pool.staticCtx.load() == cast[pointer](ctx)
 
+proc recycleFFIContext*[T](
+    pool: var FFIContextPool[T], ctx: ptr FFIContext[T]
+): Result[void, string] =
+  ## Normal teardown: drains in-flight handlers, frees the lib and returns the
+  ## slot to the pool WITHOUT stopping its threads, so a later createFFIContext
+  ## reuses them. Synchronous (waits for the FFI thread to finish draining).
+  # Recycling it would release the slot while `staticState` still points at it.
+  if pool.isStaticCtx(ctx):
+    return err("recycleFFIContext(pool): the {.ffiStatic.} context outlives every ctx")
+  ctx.requestRecycle()
+
 proc destroyFFIContext*[T](
     pool: var FFIContextPool[T], ctx: ptr FFIContext[T]
 ): Result[void, string] =
-  ## On thread-exit timeout the slot is leaked; closing live-thread resources is unsafe.
+  ## Full teardown: stops/joins the threads and frees resources, marking the slot
+  ## uninitialised so a later createFFIContext rebuilds it; normal cleanup uses
+  ## recycleFFIContext. On thread-exit timeout the slot is leaked; closing
+  ## live-thread resources is unsafe.
   # Destroying it would release the slot while `staticState` still points at it.
   if pool.isStaticCtx(ctx):
     return err("destroyFFIContext(pool): the {.ffiStatic.} context outlives every ctx")
   ctx.stopAndJoinThreads().isOkOr:
     return err("destroyFFIContext(pool): " & $error)
-  # Required: next acquisition would otherwise re-init a live lock (UB).
   let deinitRes = ctx.deinitContextResources()
   pool.releaseSlot(ctx)
   deinitRes.isOkOr:
@@ -118,6 +137,6 @@ proc isValidCtx*[T](pool: var FFIContextPool[T], ctx: pointer): bool =
   if ctx.isNil():
     return false
   for i in 0 ..< MaxFFIContexts:
-    if cast[pointer](pool.slots[i].addr) == ctx:
-      return pool.inUse[i].load()
+    if cast[pointer](pool.contexts[i].addr) == ctx:
+      return cast[ptr FFIContext[T]](ctx).isInUse()
   false

ffi/ffi_events.nim

@@ -35,6 +35,13 @@ proc deinitEventRegistry*(reg: var FFIEventRegistry) =
   reg.byEvent = default(Table[string, seq[FFIEventListener]])
   reg.nextId = 0'u64
 
+proc clearListeners*(reg: var FFIEventRegistry) {.raises: [].} =
+  ## Drops all listeners (used when a context is recycled for reuse) without
+  ## touching the lock — the event thread keeps using it across recycles.
+  withLock reg.lock:
+    reg.byEvent.clear()
+    reg.nextId = 0'u64
+
 proc addEventListener*(
     reg: var FFIEventRegistry,
     eventName: string,

ffi/ffi_thread.nim

@@ -16,6 +16,10 @@ proc sendRequestToFFIThread*(
       "reentrant ffi call: a handler invoked sendRequestToFFIThread on its own context"
     )
 
+  if ctx.lifecycle.load() != CtxLifecycle.Active:
+    deleteRequest(ffiRequest)
+    return err("FFI context is not accepting requests (being recycled)")
+
   # Wake only when the push found the queue empty: waking per submit kills scaling, and a skipped wake just waits the consumer's 100ms poll.
   let shouldWake = ctx.reqQueueBank.pushRequest(ffiRequest)
 
@@ -82,6 +86,56 @@ proc processRequest[T](
   except Exception as e:
     error "Unexpected exception in handleRes", error = e.msg
 
+const RecycleTimeout = 1500.milliseconds
+  ## Bounds how long the recycle handler waits for in-flight handlers before it
+  ## cancels them, so a wedged handler cannot block reuse forever.
+
+proc freeLib[T](ctx: ptr FFIContext[T]) {.gcsafe.} =
+  ## Releases the library object the ctor stored in ctx.myLib. Only owned libs
+  ## (createShared'd by a ctor) are freed; the worker's stack fallback is not.
+  if not ctx.myLibOwned or ctx.myLib.isNil():
+    ctx.myLib = nil
+    return
+  when not defined(gcRefc):
+    try:
+      {.cast(gcsafe).}:
+        `=destroy`(ctx.myLib[])
+    except Exception:
+      discard
+  else:
+    when T is ref:
+      if ctx.myLibRefd:
+        GC_unref(ctx.myLib[])
+        ctx.myLibRefd = false
+  freeShared(ctx.myLib)
+  ctx.myLib = nil
+  ctx.myLibOwned = false
+
+proc recycleContext[T](
+    ctx: ptr FFIContext[T], ongoing: ptr seq[Future[void]]
+) {.async.} =
+  ## Drain in-flight handlers, free the lib, clear listeners and release the
+  ## slot — all WITHOUT stopping the worker/event threads, so the next
+  ## createFFIContext reuses them (no fd churn). Then fire recycleDoneSignal.
+  ongoing[].keepItIf(not it.finished())
+  var drained = ongoing[].len == 0
+  if not drained:
+    drained = await allFutures(ongoing[]).withTimeout(RecycleTimeout)
+  if not drained:
+    for fut in ongoing[]:
+      if not fut.finished():
+        fut.cancelSoon()
+    drained = await allFutures(ongoing[]).withTimeout(RecycleTimeout)
+
+  freeLib(ctx)
+  clearListeners(ctx[].eventRegistry)
+  ongoing[].setLen(0)
+  ctx.release()
+
+  let fireRes = ctx.recycleDoneSignal.fireSync()
+  if fireRes.isErr():
+    error "failed to fire recycleDoneSignal", err = fireRes.error
+
 var ffiEventQueueSignalPtr {.threadvar.}: ThreadSignalPtr
   # Stashed so the hook has no closure env.
 
@@ -152,6 +206,13 @@ proc ffiThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
     while ctx.running.load():
       ctx.proveAlive()
 
+      # Recycle requested by the ffiDtor: drain + free lib + release the slot,
+      # keeping this thread alive for the next createFFIContext to reuse.
+      var expected = CtxLifecycle.RecyclePending
+      if ctx.lifecycle.compareExchange(expected, CtxLifecycle.Recycling):
+        await recycleContext(ctx, addr pending)
+        continue
+
       cleanFinishedRequests()
 
       # Block until a submit signals us, or at most 100ms.

ffi/internal/ffi_macro.nim

@@ -392,13 +392,18 @@ proc buildFFINewReqProc(reqTypeName, body: NimNode): NimNode =
           `reqObjIdent`.`fieldName` = `fieldName`
       )
 
+  let reqNameLit = newLit(
+    if reqTypeName.kind == nnkPostfix:
+      $reqTypeName[1]
+    else:
+      $reqTypeName
+  )
   newBody.add(
     quote do:
-      let typeStr = $T
       # Encode into shared memory, avoiding a second seq[byte] copy.
       let (sharedData, sharedLen) = cborEncodeShared(`reqObjIdent`)
       return FFIThreadRequest.initFromOwnedShared(
-        callback, userData, typeStr.cstring, sharedData, sharedLen
+        callback, userData, cstring(`reqNameLit`), sharedData, sharedLen
       )
   )
 
@@ -1077,10 +1082,15 @@ proc buildFFIProc(
       ffiBody.add(stmt)
 
     let reqPtrIdent = genSym(nskLet, "reqPtr")
+    let reqNameLit = newLit(
+      if reqTypeName.kind == nnkPostfix:
+        $reqTypeName[1]
+      else:
+        $reqTypeName
+    )
     ffiBody.add quote do:
-      let typeStr = $`reqTypeName`
       let `reqPtrIdent` = FFIThreadRequest.initFromPtr(
-        callback, userData, typeStr.cstring, reqCbor, int(reqCborLen)
+        callback, userData, cstring(`reqNameLit`), reqCbor, int(reqCborLen)
       )
     ffiBody.add buildSendAndReply(reqPtrIdent)
 
@@ -1208,11 +1218,16 @@ proc buildCtorFFINewReqProc(reqTypeName: NimNode, paramNames: seq[string]): NimN
   let retType = newTree(nnkPtrTy, ident("FFIThreadRequest"))
   formalParams = @[retType] & formalParams
 
+  let reqNameLit = newLit(
+    if reqTypeName.kind == nnkPostfix:
+      $reqTypeName[1]
+    else:
+      $reqTypeName
+  )
   var newBody = newStmtList()
   newBody.add quote do:
-    let typeStr = $T
     return FFIThreadRequest.initFromPtr(
-      callback, userData, typeStr.cstring, reqCbor, int(reqCborLen)
+      callback, userData, cstring(`reqNameLit`), reqCbor, int(reqCborLen)
     )
 
   let newReqProc = newProc(
@@ -1299,9 +1314,18 @@ proc buildCtorProcessFFIRequestProc(
       return err($error)
 
   let myLibIdent = newDotExpr(newTree(nnkDerefExpr, ctxIdent), ident("myLib"))
+  let myLibOwnedIdent = newDotExpr(newTree(nnkDerefExpr, ctxIdent), ident("myLibOwned"))
+  let myLibRefdIdent = newDotExpr(newTree(nnkDerefExpr, ctxIdent), ident("myLibRefd"))
   newBody.add quote do:
     `myLibIdent` = createShared(`libTypeName`)
     `myLibIdent`[] = `libValIdent`
+    `myLibOwnedIdent` = true
+    # Root the ref lib under refc: it lives only via this ptr in non-GC
+    # createShared memory, invisible to the cycle collector. freeLib unroots it.
+    when defined(gcRefc):
+      when `libTypeName` is ref:
+        GC_ref(`myLibIdent`[])
+        `myLibRefdIdent` = true
 
   newBody.add quote do:
     return ok($cast[uint](`ctxIdent`))
@@ -1649,7 +1673,7 @@ macro ffiDtor*(args: varargs[untyped]): untyped =
   let poolIdent = ident($libTypeName & "FFIPool")
   ffiBody.add quote do:
     let `destroyResIdent` =
-      `poolIdent`.destroyFFIContext(cast[ptr FFIContext[`libTypeName`]](ctx))
+      `poolIdent`.recycleFFIContext(cast[ptr FFIContext[`libTypeName`]](ctx))
     if `destroyResIdent`.isErr():
       return RET_ERR