refactor(ffi): event thread scaffolding + FFIContext lifecycle split (#71)

Author: gmelodie

examples/timer/rust_bindings/examples/main.rs

@@ -1,10 +1,8 @@
-//! Synchronous example: exercises the library-event listener API
-//! (typed + wildcard + remove).
+//! Synchronous example: exercises the typed per-event listener API.
 //!
 //! Run with: `cargo run --example main`
 
-use my_timer::{decode_event_payload, EchoEvent, EchoRequest, MyTimerCtx, TimerConfig};
-use std::os::raw::c_int;
+use my_timer::{EchoEvent, EchoRequest, MyTimerCtx, TimerConfig};
 use std::sync::mpsc;
 use std::time::Duration;
 
@@ -14,32 +12,12 @@ fn main() -> Result<(), String> {
         Duration::from_secs(5),
     )?;
 
-    // Typed listener: the closure is invoked on the lib's dispatch
-    // thread, so forward the payload to `main` via std mpsc and block
-    // on `recv_timeout` below. `add_on_echo_fired_listener` is generated
-    // per `{.ffiEvent.}`-declared proc and takes a typed `&EchoEvent`.
+    // Closure runs on the lib's dispatch thread; forward to `main` via mpsc and recv_timeout below.
     let (tx, rx) = mpsc::channel::<EchoEvent>();
     let typed_handle = ctx.add_on_echo_fired_listener(move |evt: &EchoEvent| {
         let _ = tx.send(evt.clone());
     });
 
-    // Wildcard listener: receives every event with the FFI return code,
-    // the wire `event_id` pre-extracted from the CBOR envelope, and the
-    // raw envelope bytes. Lift to a typed payload via
-    // `decode_event_payload::<T>` when the event_id matches one you
-    // care about — this avoids hand-rolling ciborium calls per branch.
-    let wildcard_handle = ctx.add_event_listener(|ret: c_int, event_id: &str, envelope: &[u8]| {
-        println!("wildcard: ret={}, event_id={}, bytes={}", ret, event_id, envelope.len());
-        if ret == 0 && event_id == "on_echo_fired" {
-            match decode_event_payload::<EchoEvent>(envelope) {
-                Ok(evt) => println!("  decoded: message={}, echo_count={}", evt.message, evt.echo_count),
-                Err(e) => println!("  decode failed: {}", e),
-            }
-        }
-    });
-
-    // Trigger the event — fires `on_echo_fired` once, which the
-    // dispatch thread delivers to both listeners above.
     ctx.echo(EchoRequest { message: "sync-event-demo".into(), delay_ms: 1 })?;
 
     match rx.recv_timeout(Duration::from_secs(2)) {
@@ -48,6 +26,5 @@ fn main() -> Result<(), String> {
     }
 
     ctx.remove_event_listener(typed_handle);
-    ctx.remove_event_listener(wildcard_handle);
     Ok(())
 }

examples/timer/rust_bindings/examples/tokio_main.rs

@@ -1,11 +1,9 @@
-//! Tokio (async) example: same shape as `main.rs` but exercises the
-//! async `_async` API and bridges library events into a tokio-aware
-//! channel for async consumption.
+//! Tokio (async) example: same shape as `main.rs` but exercises the async `_async` API
+//! and bridges library events into a tokio mpsc for async consumption.
 //!
 //! Run with: `cargo run --example tokio_main`
 
-use my_timer::{decode_event_payload, EchoEvent, EchoRequest, MyTimerCtx, TimerConfig};
-use std::os::raw::c_int;
+use my_timer::{EchoEvent, EchoRequest, MyTimerCtx, TimerConfig};
 use std::time::Duration;
 use tokio::sync::mpsc;
 
@@ -17,44 +15,21 @@ async fn main() -> Result<(), String> {
     )
     .await?;
 
-    // Typed listener: the handler fires on the lib's dispatch thread,
-    // which is *outside* the tokio runtime. Forwarding through a tokio
-    // `unbounded_channel` (Sender is Send + Sync, non-blocking) hands
-    // the event over to the runtime so we can `.await` it below.
+    // Handler fires on the lib's dispatch thread (outside the tokio runtime); forward via tokio mpsc to await it below.
     let (typed_tx, mut typed_rx) = mpsc::unbounded_channel::<EchoEvent>();
     let typed_handle = ctx.add_on_echo_fired_listener(move |evt: &EchoEvent| {
         let _ = typed_tx.send(evt.clone());
     });
 
-    // Wildcard listener: receives every event with the FFI return code,
-    // the wire `event_id` pre-extracted from the CBOR envelope, and the
-    // raw envelope bytes. Lift to a typed payload via
-    // `decode_event_payload::<T>` when the event_id matches one you
-    // care about — this avoids hand-rolling ciborium calls per branch.
-    let wildcard_handle = ctx.add_event_listener(|ret: c_int, event_id: &str, envelope: &[u8]| {
-        println!("wildcard: ret={}, event_id={}, bytes={}", ret, event_id, envelope.len());
-        if ret == 0 && event_id == "on_echo_fired" {
-            match decode_event_payload::<EchoEvent>(envelope) {
-                Ok(evt) => println!("  decoded: message={}, echo_count={}", evt.message, evt.echo_count),
-                Err(e) => println!("  decode failed: {}", e),
-            }
-        }
-    });
-
-    // Trigger an echo via the async API — fires `on_echo_fired` once,
-    // which the dispatch thread delivers to both listeners above.
     ctx.echo_async(EchoRequest { message: "async-event-demo".into(), delay_ms: 1 })
         .await?;
 
-    // Await the typed event with a bounded timeout so a missing event
-    // surfaces as an error instead of hanging the example forever.
     let evt = tokio::time::timeout(Duration::from_secs(2), typed_rx.recv())
         .await
         .map_err(|_| "event never arrived".to_string())?
         .ok_or_else(|| "typed channel closed".to_string())?;
     println!("typed onEchoFired: message={}, echo_count={}", evt.message, evt.echo_count);
 
     ctx.remove_event_listener(typed_handle);
-    ctx.remove_event_listener(wildcard_handle);
     Ok(())
 }

ffi/cbor_serial.nim

@@ -35,9 +35,6 @@ export cbor_serialization, options, results
 const CborNullByte*: byte = 0xf6'u8
   ## CBOR encoding of `null` — used as the wire sentinel for empty OK payloads.
 
-# ---------------------------------------------------------------------------
-# Public API
-# ---------------------------------------------------------------------------
 
 proc cborEncode*[T](x: T): seq[byte] =
   ## CBOR-encode any cbor_serialization-supported type (plus `pointer` / `ptr T`

ffi/codegen/rust.nim

@@ -63,9 +63,6 @@ proc reqStructName(p: FFIProcMeta): string =
   else:
     camel & "Req"
 
-# ---------------------------------------------------------------------------
-# File generators
-# ---------------------------------------------------------------------------
 
 proc generateCargoToml*(libName: string): string =
   # `flume` is the unified callback channel (PR #23 Rust review, item 8): one

ffi/event_thread.nim

@@ -0,0 +1,138 @@
+## Event-thread body and FFI-thread liveness monitoring.
+##
+## Included from `ffi_context.nim` — inherits its imports, FFIContext type,
+## and the heartbeat-timing constants. Lives alongside `ffi_thread.nim`
+## so each thread's machinery is readable on its own.
+##
+## Responsibilities:
+## - Drain queued events into listener callbacks (queue producer lands in PR #69).
+## - Watch `ctx.ffiHeartbeat` and emit `NotRespondingEvent` / `RespondingEvent`
+##   on FFI-thread stall and recovery transitions.
+
+type
+  NotRespondingEvent* = object
+  RespondingEvent* = object
+
+const
+  NotRespondingEventName* = "not_responding"
+  RespondingEventName* = "responding"
+
+proc dispatchToListeners[T](
+    ctx: ptr FFIContext[T], eventName: string, data: pointer, dataLen: int
+) =
+  ## Holds reg.lock for the entire snapshot + invocation so concurrent
+  ## add/remove on this registry blocks until dispatch returns.
+  withLock ctx[].eventRegistry.lock:
+    let listeners = ctx[].eventRegistry.byEvent.getOrDefault(eventName)
+    if listeners.len == 0:
+      chronicles.debug "no listener registered", event = eventName
+      return
+    foreignThreadGc:
+      try:
+        notifyListeners(listeners, RET_OK, data, dataLen)
+      except Exception, CatchableError:
+        notifyListenersErr(
+          listeners,
+          "Exception dispatching " & eventName & ": " & getCurrentExceptionMsg(),
+        )
+
+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.
+  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
+      return
+  let dataPtr: pointer =
+    if event.len > 0: cast[pointer](unsafeAddr event[0])
+    else: cast[pointer](emptyListenerPayload)
+  ctx.dispatchToListeners(name, dataPtr, event.len)
+
+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.
+  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)
+  ctx.dispatchToListeners($qe.name, qe.data, qe.dataLen)
+
+proc drainEventQueue[T](ctx: ptr FFIContext[T]) =
+  while true:
+    let opt = ctx.eventQueue.tryDequeueEvent()
+    if opt.isNone():
+      break
+    ctx.dispatchQueuedEvent(opt.get())
+
+type HeartbeatMonitor = object
+  startedAt: Moment
+  lastChange: Moment
+  lastValue: int64
+  notifiedStale: bool
+
+proc init(T: type HeartbeatMonitor, ctx: ptr FFIContext): T =
+  let now = Moment.now()
+  T(
+    startedAt: now,
+    lastChange: now,
+    lastValue: ctx.ffiHeartbeat.load(),
+    notifiedStale: false,
+  )
+
+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.
+  if Moment.now() - hb.startedAt <= FFIHeartbeatStartDelay:
+    return
+  let cur = ctx.ffiHeartbeat.load()
+  if cur != hb.lastValue:
+    if hb.notifiedStale:
+      onResponding(ctx)
+    hb.lastValue = cur
+    hb.lastChange = Moment.now()
+    hb.notifiedStale = false
+  elif not hb.notifiedStale and
+      Moment.now() - hb.lastChange > FFIHeartbeatStaleThreshold:
+    onNotResponding(ctx)
+    hb.notifiedStale = true
+
+proc eventRun[T](ctx: ptr FFIContext[T]) {.async.} =
+  var hb = HeartbeatMonitor.init(ctx)
+
+  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.
+    discard await ctx.eventQueueSignal.wait().withTimeout(EventThreadTickInterval)
+
+    ctx.drainEventQueue()
+
+    if not ctx.running.load():
+      break
+    hb.check(ctx)
+
+proc eventThreadBody[T](ctx: ptr FFIContext[T]) {.thread.} =
+  ## Drains the event queue and runs the FFI-thread heartbeat check.
+  ## Owns the queued `c_malloc` payloads until dispatch returns.
+  defer:
+    let fireRes = ctx.eventThreadExitSignal.fireSync()
+    if fireRes.isErr():
+      error "failed to fire eventThreadExitSignal", err = fireRes.error
+
+  try:
+    waitFor eventRun(ctx)
+  except CatchableError as exc:
+    error "event thread exited with exception", error = exc.msg