feat: move user event code to a dedicated event thread (#69)

Author: gmelodie

CHANGELOG.md

@@ -2,6 +2,26 @@
 
 All notable changes to this project are documented in this file.
 
+## [Unreleased]
+
+### Changed
+- User event callbacks now run on a dedicated event thread fed by a
+  bounded SPSC queue (default capacity 1024), so a slow listener can no
+  longer block the FFI thread or concurrent `add_event_listener` /
+  `remove_event_listener` calls
+  ([#6](https://github.com/logos-messaging/nim-ffi/issues/6)).
+- Replaced the dedicated watchdog thread with a heartbeat check that
+  runs on the event thread. The FFI thread advances an atomic heartbeat
+  each loop iteration; if it stalls for more than 1s past the start-up
+  grace window, the event thread emits the `not_responding` event.
+
+### Added
+- Queue-overflow handling: when the bounded event queue is full, the
+  library sets a sticky "stuck" flag, logs an error, fires
+  `not_responding` from the event thread, and rejects subsequent
+  `sendRequestToFFIThread` calls with `event queue stuck - library
+  cannot accept new requests`.
+
 ## [0.2.0] - 2026-06-04
 
 Major release introducing the CBOR-based wire format, CBOR-backed FFI events

ffi/event_thread.nim

@@ -5,7 +5,7 @@
 ## so each thread's machinery is readable on its own.
 ##
 ## Responsibilities:
-## - Drain queued events into listener callbacks (queue producer lands in PR #69).
+## - Drain queued events into listener callbacks.
 ## - Watch `ctx.ffiHeartbeat` and emit `NotRespondingEvent` / `RespondingEvent`
 ##   on FFI-thread stall and recovery transitions.
 
@@ -37,14 +37,13 @@ proc dispatchToListeners[T](
         )
 
 proc emitLivenessEvent[T, P](ctx: ptr FFIContext[T], name: string, payload: P) =
-  ## Encodes a zero-field liveness event (`NotRespondingEvent`,
-  ## `RespondingEvent`) and dispatches it directly to listeners, bypassing
-  ## the event queue (which may itself be wedged). Runs on the event thread.
+  ## Encodes a liveness event and dispatches directly to listeners (bypassing
+  ## the queue, which may be wedged). Runs on the event thread.
   let event =
     try:
       EventEnvelope[P](eventType: name, payload: payload).cborEncode()
-    except CatchableError as exc:
-      chronicles.error "liveness event encode failed", name = name, err = exc.msg
+    except CatchableError as e:
+      chronicles.error "liveness event encode failed", name = name, err = e.msg
       return
   let dataPtr: pointer =
     if event.len > 0: cast[pointer](unsafeAddr event[0])
@@ -55,18 +54,14 @@ proc onNotResponding*(ctx: ptr FFIContext) =
   emitLivenessEvent(ctx, NotRespondingEventName, NotRespondingEvent())
 
 proc onResponding*(ctx: ptr FFIContext) =
-  ## Fired once when the FFI thread's heartbeat starts advancing again
-  ## after a `NotRespondingEvent`. Lets consumers clear any "library
-  ## hung" UI state without polling.
+  ## Fired once when the heartbeat resumes after a NotRespondingEvent.
+  ## Lets consumers clear any "library hung" UI state without polling.
   emitLivenessEvent(ctx, RespondingEventName, RespondingEvent())
 
 proc dispatchQueuedEvent[T](ctx: ptr FFIContext[T], qe: QueuedEvent) =
   ## Frees `qe`'s c_malloc buffers on exit.
   defer:
-    if not qe.name.isNil():
-      c_free(cast[pointer](qe.name))
-    if not qe.data.isNil():
-      c_free(qe.data)
+    freeEventBuffers(qe.name, qe.data)
   ctx.dispatchToListeners($qe.name, qe.data, qe.dataLen)
 
 proc drainEventQueue[T](ctx: ptr FFIContext[T]) =
@@ -92,10 +87,8 @@ proc init(T: type HeartbeatMonitor, ctx: ptr FFIContext): T =
   )
 
 proc check[T](hb: var HeartbeatMonitor, ctx: ptr FFIContext[T]) =
-  ## Fires `onNotResponding` once the FFI thread's heartbeat counter stops
-  ## advancing past the stale threshold, and fires `onResponding` once it
-  ## starts advancing again. Both transitions latch so each is emitted at
-  ## most once per stall episode.
+  ## Fires onNotResponding / onResponding on heartbeat stall / recovery.
+  ## Both transitions latch — each fires at most once per stall episode.
   if Moment.now() - hb.startedAt <= FFIHeartbeatStartDelay:
     return
   let cur = ctx.ffiHeartbeat.load()
@@ -112,14 +105,19 @@ proc check[T](hb: var HeartbeatMonitor, ctx: ptr FFIContext[T]) =
 
 proc eventRun[T](ctx: ptr FFIContext[T]) {.async.} =
   var hb = HeartbeatMonitor.init(ctx)
+  var notifiedStuck = false # latched forever — eventQueueStuck is sticky terminal.
 
   while ctx.running.load():
-    # Wake on enqueue or tick — whichever first. The enqueue path lands in PR #69;
-    # until then the wait always times out and we fall through to the heartbeat check.
+    # Wake on enqueue or tick — whichever first.
     discard await ctx.eventQueueSignal.wait().withTimeout(EventThreadTickInterval)
 
     ctx.drainEventQueue()
 
+    # Fire after drain so reg.lock is free — FFI-thread would deadlock here.
+    if not notifiedStuck and ctx.eventQueueStuck.load():
+      onNotResponding(ctx)
+      notifiedStuck = true
+
     if not ctx.running.load():
       break
     hb.check(ctx)
@@ -134,5 +132,5 @@ proc eventThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
 
   try:
     waitFor eventRun(ctx)
-  except CatchableError as exc:
-    error "event thread exited with exception", error = exc.msg
+  except CatchableError as e:
+    error "event thread exited with exception", error = e.msg