Add Elixir/OTP implementation of AbxBus event system

[pirate]

Summary

This PR introduces a complete Elixir/OTP implementation of AbxBus, an event bus system with queue-jump, multi-bus forwarding, and lineage tracking. The implementation ports the core functionality from the Python reference implementation while leveraging Elixir's process model and OTP patterns for concurrency control.

Key Changes

• Core event bus engine (BusServer): GenServer managing event queues, concurrency enforcement (parallel/bus-serial/global-serial), and handler dispatch
• Event worker (EventWorker): Per-event process executing handlers with timeout enforcement and result tracking
• Event store (EventStore): ETS-backed registry for O(1) event lookup, parent-child indexing, and completion notification
• Lock manager (LockManager): Centralized concurrency policy enforcement including global-serial coordination and named semaphores
• Event model (Event): Base event struct with reserved metadata fields and helper functions
• Handler tracking (HandlerEntry, HandlerResult): Registration and result tracking for event handlers
• Public API (AbxBus): High-level functions for bus lifecycle, event emission, handler registration, and awaiting
• Comprehensive test suite: 10+ test files covering locking, timeouts, forwarding, parent tracking, queue-jump mechanics, and cross-runtime parity

Notable Implementation Details

• Queue-jump without re-entrant locks: Leverages BEAM's process model—parent handler blocks in receive (not holding GenServer state), allowing child events to process independently without deadlock
• Process dictionary for handler context: Automatic parent tracking via Process.put(:abx_current_event_id, ...) replaces Python's ContextVar approach
• ETS-based event registry: Provides O(1) lookups and efficient parent-child relationship tracking
• Timeout enforcement: Event-level and handler-level timeouts with proper cancellation semantics
• Forwarding support: Events can be forwarded to other buses with same event_id (not parent-child)
• Backpressure handling: History size limits with configurable drop behavior
• Idle detection: wait_until_idle/1 blocks until all pending and in-flight events complete

The implementation maintains API parity with the Python reference while adapting concurrency patterns to OTP idioms (GenServer for bus-serial, LockManager for global-serial, bare processes for parallel).

https://claude.ai/code/session_01Qf9t3FPgh38PQUfKUcUeGx

---

---

Summary by cubic

Introduces an OTP‑native event bus :abxbus with queue‑jump, multi‑bus forwarding, lineage tracking, and cross‑runtime JSON parity. Adds deterministic UUIDv5 handler IDs, result type enforcement, and populates eventbus_id on per‑handler results; find() is faster and scales via a type‑indexed ETS lookup.

• New Features

  • Recursion‑depth guard (max_handler_recursion_depth, default 3) to stop infinite emit→await loops.
  • WAL middleware (Abxbus.Middlewares.WAL) appends completed events to a JSONL‑like log; per‑bus config in ETS.
  • JSON parity helpers: Abxbus.to_json/1, from_json/2, to_json_string/1, from_json_string/2 (uses Jason if present).
  • Real‑time handler result status updates in ETS; find() returns the most recent match.
  • Deterministic handler IDs via UUIDv5 (cross‑runtime consistent), optional explicit ID, handler_registered_at as ISO datetime; eventbus_id now set on EventResult.
  • Result type enforcement for event_result_type (:string, :integer, :number, :boolean, :atom, :list, :map, :nil, :any, or struct module); mismatches recorded as :error with Abxbus.EventHandlerResultSchemaError.

• Bug Fixes

  • Retries respect retry_on_errors for exits; robust forwarding merge across buses preserves status/results and merges event_path.
  • Global locks/semaphores: synchronous global release + broadcast; monitor holders and auto‑release on :DOWN; precise per‑holder release; bus‑scoped semaphore keys.
  • Await/find: register waiters before checks; clean waiters on timeout/already‑completed; re‑read from ETS before notifying; normalize string patterns; support transitive child_of; return most‑recent past match.
  • Depth tracking cleaned only when event_pending_bus_count is 0; safe JSON atom handling; wildcard forwarding loop prevention.
  • find() performance: added a type‑indexed ETS secondary table to avoid O(n) scans; queries now scale with events per type and no longer flake under large tables.
  • Handler lookup guard fixed to include string‑pattern handlers by comparing key counts (not entry counts), preventing silent drops.

^{Written for commit 6b6499c. Summary will update on new commits.}

[cubic-dev-ai]

19 issues found across 26 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="abx_bus/test/cross_runtime_features_test.exs">

<violation number="1" location="abx_bus/test/cross_runtime_features_test.exs:112">
P2: The test claims to validate per-event serial handler execution, but it only asserts global concurrency and can miss regressions in event-level handler serialness.</violation>

<violation number="2" location="abx_bus/test/cross_runtime_features_test.exs:122">
P2: Sleeping exactly the configured timeout creates a flaky boundary-condition test; use a duration clearly above the timeout.</violation>
</file>

<file name="abx_bus/lib/abx_bus/event.ex">

<violation number="1" location="abx_bus/lib/abx_bus/event.ex:104">
P1: `new/2` passes metadata override keys into `struct!/2`, causing runtime errors when callers provide allowed `event_*` overrides.</violation>
</file>

<file name="abx_bus/test/event_bus_find_test.exs">

<violation number="1" location="abx_bus/test/event_bus_find_test.exs:95">
P2: This test uses a fixed sleep for synchronization, which makes it timing-sensitive and can cause flaky failures in CI.</violation>
</file>

<file name="abx_bus/test/event_handler_test.exs">

<violation number="1" location="abx_bus/test/event_handler_test.exs:20">
P2: This test claims to verify cancellation of non-winning handlers, but it only checks the winning result. Add an assertion that slow/losing handlers are canceled; otherwise cancellation regressions can pass unnoticed.</violation>
</file>

<file name="abx_bus/lib/abx_bus.ex">

<violation number="1" location="abx_bus/lib/abx_bus.ex:124">
P1: Race condition (TOCTOU) between the completion check and waiter registration. If the event completes between `EventStore.get` and `EventStore.add_waiter`, the waiter is registered after `notify_waiters` already fired, so `await` hangs forever.

Fix: `add_waiter` should atomically check completion status and either return the completed event immediately or register the waiter under a single serialization point (e.g., a GenServer call to EventStore, or re-check after insert).</violation>
</file>

<file name="abx_bus/test/parent_tracking_test.exs">

<violation number="1" location="abx_bus/test/parent_tracking_test.exs:81">
P2: This test has no assertion for the stated behavior, so regressions in explicit `parent_id` handling would go undetected.</violation>
</file>

<file name="abx_bus/config/config.exs">

<violation number="1" location="abx_bus/config/config.exs:4">
P2: These `default_*` config values are currently unused, so this new config block does not affect bus behavior and can mislead users into thinking defaults are configurable here.</violation>
</file>

<file name="abx_bus/test/dispatch_defaults_test.exs">

<violation number="1" location="abx_bus/test/dispatch_defaults_test.exs:52">
P2: The max-concurrency tracker uses a non-atomic compare/set and can lose updates, making this test flaky and potentially masking real concurrency violations.</violation>
</file>

<file name="abx_bus/lib/abx_bus/event_store.ex">

<violation number="1" location="abx_bus/lib/abx_bus/event_store.ex:40">
P1: Non-atomic read-modify-write on a public ETS table. `update/2` does `get` → `Map.merge` → `put`, but since these are called directly from `EventWorker` and `BusServer` processes concurrently, a racing update will silently overwrite the other's changes (e.g., handler results stomp a status transition or vice versa). Serialize updates through the GenServer, or restructure to use `:ets.update_element/3` for field-level atomic updates.</violation>

<violation number="2" location="abx_bus/lib/abx_bus/event_store.ex:53">
P1: Same non-atomic read-modify-write race as `update/2`. `update_fun/2` is used in `BusServer` to decrement `event_pending_bus_count` — a lost update here means the counter never reaches zero and the event is never marked completed.</violation>

<violation number="3" location="abx_bus/lib/abx_bus/event_store.ex:195">
P1: Race condition between past search and future waiter registration. An event emitted between `search_past` returning nil and `add_find_waiter` being called will be missed entirely — it's not in the past scan results and the waiter isn't registered yet for `resolve_find_waiters` to notify. Register the waiter *before* scanning past events, then clean it up if a past match is found.</violation>
</file>

<file name="abx_bus/test/event_bus_forwarding_test.exs">

<violation number="1" location="abx_bus/test/event_bus_forwarding_test.exs:28">
P2: The circular-forwarding test sets up only a linear chain, so it does not validate stale in-flight cleanup under an actual forwarding cycle.</violation>

<violation number="2" location="abx_bus/test/event_bus_forwarding_test.exs:113">
P2: This conditional makes the concurrency assertion optional; the test can pass without validating the expected behavior.</violation>
</file>

<file name="abx_bus/test/event_bus_timeout_test.exs">

<violation number="1" location="abx_bus/test/event_bus_timeout_test.exs:94">
P2: This test uses handler sleep durations equal to the event timeout, which makes the assertion nondeterministic at the timeout boundary.</violation>
</file>

<file name="abx_bus/lib/abx_bus/bus_server.ex">

<violation number="1" location="abx_bus/lib/abx_bus/bus_server.ex:185">
P2: TOCTOU race in `ensure_bus_pid_table/0`: two concurrent `init/1` calls can both see `:undefined` and race on `:ets.new`, crashing the second caller. Wrap the creation in a `try`/`catch` to handle the already-exists case.</violation>

<violation number="2" location="abx_bus/lib/abx_bus/bus_server.ex:232">
P1: History list is never trimmed when `max_history_drop` is `true` (the default). The `check_backpressure` function allows events through, but no code ever drops old entries. This is an unbounded memory leak. Trim the list after prepending.</violation>

<violation number="3" location="abx_bus/lib/abx_bus/bus_server.ex:448">
P1: `spawn_link` without trapping exits will crash the BusServer GenServer if any EventWorker fails unexpectedly. Use `Process.monitor/1` with `spawn/1` instead so the GenServer receives a `{:DOWN, ref, ...}` info message that can be handled gracefully, rather than being killed by the linked exit.</violation>

<violation number="4" location="abx_bus/lib/abx_bus/bus_server.ex:518">
P2: `length(state.history)` is O(n) and is called on every `emit`. As history grows this becomes a bottleneck. Track the count in a separate field or use a bounded data structure.</violation>
</file>

Architecture diagram

sequenceDiagram
    participant Client
    participant API as AbxBus (Public API)
    participant Bus as BusServer (GenServer)
    participant Store as EventStore (ETS)
    participant Lock as LockManager
    participant Worker as EventWorker (Process)
    participant Handler as User Callback

    Note over Client,Handler: NEW: Event Emission Flow
    Client->>API: emit(bus, event)
    API->>API: NEW: Resolve parent_id (Process Dictionary)
    API->>Bus: call :emit
    Bus->>Store: NEW: put(event) & index to bus
    Bus->>Bus: Enqueue event (O(1) :queue)
    
    Note over Bus,Worker: NEW: Execution Cycle
    Bus->>Lock: NEW: Resolve concurrency policy
    opt Concurrency available (e.g., :bus_serial)
        Bus->>Worker: NEW: spawn_link(run/5)
        Worker->>Worker: NEW: Set Context (EventID/Bus)
        Worker->>Handler: execute callback
        
        Note over Handler,Worker: NEW: Queue-Jump Logic
        opt Nested Emission & Await
            Handler->>API: emit(child_event)
            API->>Bus: call :emit (Child)
            Handler->>API: await(child_event)
            API->>Bus: NEW: cast :jump_queue (event_id)
            Bus->>Bus: NEW: Dequeue child & spawn Worker immediately
            Bus->>Worker: spawn_link(child)
            Worker->>Store: update status :completed
            Store-->>API: Notify completion (ref)
            API-->>Handler: Return child result
        end

        Handler-->>Worker: result
        Worker-->>Bus: NEW: send :event_worker_done
    end

    alt NEW: Timeout Path
        Worker->>Worker: Monitors event/handler timers
        opt Timeout reached
            Worker->>Worker: Terminate task
            Worker->>Store: NEW: Mark as :error (Aborted/Cancelled)
            Worker-->>Bus: send :event_worker_done
        end
    end

    Bus->>Bus: NEW: Dequeue next event
    Note over Bus,Store: History management & Backpressure

Loading

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}

[cubic-dev-ai]

11 issues found across 29 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="abxbus-elixir/test/event_bus_find_test.exs">

<violation number="1" location="abxbus-elixir/test/event_bus_find_test.exs:96">
P2: This readiness check races with `find/2` waiter registration, so the future-event test can still fail intermittently in CI.</violation>
</file>

<file name="abxbus-elixir/lib/abxbus/bus_server.ex">

<violation number="1" location="abxbus-elixir/lib/abxbus/bus_server.ex:153">
P1: The `started` flag is set to `false` by `stop/2` but never checked in `emit` or `maybe_process_next`. The bus continues to accept and process events after being stopped. Either guard `emit` with `if not state.started, do: {:reply, {:error, :stopped}, state}`, or use `{:stop, :normal, :ok, state}` to actually terminate the GenServer.</violation>
</file>

<file name="abxbus-elixir/test/event_bus_locking_test.exs">

<violation number="1" location="abxbus-elixir/test/event_bus_locking_test.exs:80">
P2: These concurrency assertions are timing-based, so the new tests can fail nondeterministically on slower CI runners.</violation>
</file>

<file name="abxbus-elixir/lib/abxbus/event_worker.ex">

<violation number="1" location="abxbus-elixir/lib/abxbus/event_worker.ex:336">
P2: `spawn_link` is called before `Process.flag(:trap_exit, true)`, creating a small race window. If the linked process crashes before trap_exit is enabled, the parent is killed. Move `Process.flag(:trap_exit, true)` before the `spawn_link` call.</violation>

<violation number="2" location="abxbus-elixir/lib/abxbus/event_worker.ex:384">
P1: `Process.flag(:trap_exit, false)` unconditionally disables exit trapping instead of restoring the previous value. `maybe_with_handler_timeout` in the same file does this correctly with `old_trap = Process.flag(:trap_exit, true)` … `Process.flag(:trap_exit, old_trap)`. Apply the same pattern here.</violation>
</file>

<file name="abxbus-elixir/lib/abxbus.ex">

<violation number="1" location="abxbus-elixir/lib/abxbus.ex:97">
P1: The fast-path completion check only matches `:completed` status. If an event has errored before `await` is called (and waiters were already notified), this waiter will never receive a message, causing the caller to hang indefinitely. The check should also match `:error` (and any other terminal status).</violation>
</file>

<file name="abxbus-elixir/lib/abxbus/lock_manager.ex">

<violation number="1" location="abxbus-elixir/lib/abxbus/lock_manager.ex:152">
P1: Monitor global-lock holders; a crashing BusServer can leave global-serial mode stuck busy forever.</violation>

<violation number="2" location="abxbus-elixir/lib/abxbus/lock_manager.ex:164">
P1: Keying semaphores only by name makes `:bus` scope global across buses.</violation>
</file>

<file name="abxbus-elixir/lib/abxbus/event.ex">

<violation number="1" location="abxbus-elixir/lib/abxbus/event.ex:191">
P1: Allow `event_parent_id` through the metadata override path; otherwise `use Abxbus.Event` events cannot preserve an explicit parent ID.</violation>

<violation number="2" location="abxbus-elixir/lib/abxbus/event.ex:266">
P1: Generate a real UUID event_id here; the current timestamp-random format breaks cross-runtime event_id validation.</violation>
</file>

<file name="abxbus-elixir/lib/abxbus/event_store.ex">

<violation number="1" location="abxbus-elixir/lib/abxbus/event_store.ex:129">
P2: TOCTOU race in `resolve_find_waiters`: `lookup` + `delete_object` on the shared ETS table is not atomic. When multiple buses emit matching events concurrently, the same waiter can receive duplicate `:find_match` messages. Consider serializing this through the EventStore GenServer or using `:ets.take/2` (as done in `notify_waiters`) to atomically read-and-delete.</violation>
</file>

Architecture diagram

sequenceDiagram
    participant C as Client
    participant API as Abxbus API
    participant Store as EventStore (ETS)
    participant Bus as BusServer (GenServer)
    participant Lock as LockManager (GenServer)
    participant Worker as EventWorker (Process)

    Note over C,Worker: NEW: Elixir/OTP Event Bus Flow (Bus-Serial Example)

    C->>API: emit(event)
    API->>API: NEW: Extract Parent ID from P-Dict
    API->>Store: NEW: put_or_merge(event)
    API->>Bus: NEW: GenServer.call(:emit)
    
    alt Bus is Idle
        Bus->>Worker: NEW: spawn_link(run/5)
    else Bus is Busy
        Bus->>Bus: Enqueue event (:queue)
    end
    Bus-->>API: Return Event Metadata
    API-->>C: Return Event

    Note over Worker: Execution Context
    Worker->>Worker: NEW: Set P-Dict (:abx_current_event_id)
    Worker->>Lock: NEW: resolve_concurrency_policy()
    Lock-->>Worker: config (timeouts, concurrency)

    rect rgb(240, 240, 240)
        Note over Worker,API: Handler Execution
        Worker->>Worker: Execute Handler Fn
        
        opt Handler emits child
            Worker->>API: emit(child_event)
            API->>Store: NEW: update lineage (Parent -> Child)
            API->>Bus: GenServer.call(:emit)
        end

        opt Handler awaits child (Queue Jump)
            Worker->>API: await(child_event)
            API->>Store: NEW: add_waiter(child_id, self())
            API->>Bus: NEW: trigger_queue_jump(child_event)
            Bus->>Worker: NEW: spawn child worker immediately
            API->>API: NEW: receive (Blocks parent handler)
        end
    end

    Worker->>Bus: NEW: send(:event_worker_done)
    Bus->>Store: Mark event :completed
    Store->>API: NEW: notify_waiters (Signal unblock)
    
    Bus->>Bus: NEW: Dequeue next event
    
    opt Global Serial Mode
        Bus->>Lock: NEW: acquire_global() / release_global()
    end

    alt Failure Path (Timeout)
        Worker->>Worker: NEW: timer.exit after event_timeout
        Bus->>Store: NEW: Mark handlers as :error (Aborted)
    end

Loading

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}

[cubic-dev-ai]

P2: These concurrency assertions are timing-based, so the new tests can fail nondeterministically on slower CI runners.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At abxbus-elixir/test/event_bus_locking_test.exs, line 80:

<comment>These concurrency assertions are timing-based, so the new tests can fail nondeterministically on slower CI runners.</comment>

<file context>
@@ -0,0 +1,306 @@
+      max_b = :atomics.new(1, [])
+      max_global = :atomics.new(1, [])
+
+      sync_event = :erlang.make_ref()
+
+      make_handler = fn source_counter, source_max ->
</file context>

[cubic-dev-ai]

0 issues found across 1 file (changes from recent commits).

_{Requires human review: Auto-approval blocked by 14 unresolved issues from previous reviews.}

[cubic-dev-ai]

3 issues found across 9 files (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="abxbus-elixir/lib/abxbus/middlewares/wal.ex">

<violation number="1" location="abxbus-elixir/lib/abxbus/middlewares/wal.ex:70">
P1: `build/1` is no longer backward compatible for multiple WAL instances; later calls overwrite the global default agent for earlier buses.</violation>

<violation number="2" location="abxbus-elixir/lib/abxbus/middlewares/wal.ex:76">
P2: Make ETS table creation race-safe; the current `whereis`/`new` sequence can crash when two registrations happen concurrently.</violation>
</file>

<file name="abxbus-elixir/lib/abxbus/bus_server.ex">

<violation number="1" location="abxbus-elixir/lib/abxbus/bus_server.ex:347">
P2: Depth tracking is deleted too early for forwarded multi-bus events; this can reset recursion depth to 0 on still-pending buses.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}

[cubic-dev-ai]

P1: build/1 is no longer backward compatible for multiple WAL instances; later calls overwrite the global default agent for earlier buses.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At abxbus-elixir/lib/abxbus/middlewares/wal.ex, line 70:

<comment>`build/1` is no longer backward compatible for multiple WAL instances; later calls overwrite the global default agent for earlier buses.</comment>

<file context>
@@ -37,42 +39,96 @@ defmodule Abxbus.Middlewares.WAL do
+  """
+  def build(agent) do
+    ensure_config_table()
+    :ets.insert(@wal_config_table, {:__wal_default__, agent})
+    __MODULE__
+  end
</file context>

[cubic-dev-ai]

0 issues found across 2 files (changes from recent commits).

_{Requires human review: Auto-approval blocked by 16 unresolved issues from previous reviews.}

[cubic-dev-ai]

0 issues found across 1 file (changes from recent commits).

_{Requires human review: Auto-approval blocked by 15 unresolved issues from previous reviews.}

[cubic-dev-ai]

You're iterating quickly on this pull request. To help protect your rate limits, cubic has paused automatic reviews on new pushes for now—when you're ready for another review, comment @cubic-dev-ai review.

[devin-ai-integration]

Devin Review found 2 new potential issues.

View 33 additional findings in Devin Review.

[devin-ai-integration]

🟡 JSON decoder decode_array incorrectly accepts } as array terminator

The decode_array function at abxbus-elixir/lib/abxbus/json.ex:98 matches "}" (closing brace) as a valid terminator for JSON arrays, in addition to the correct "]" at line 99. In JSON, } should only close objects, while ] closes arrays. This means malformed JSON like [1, 2, 3} would be silently accepted and parsed as [1, 2, 3] instead of producing an error. More critically, in edge cases involving nested structures, this could cause the parser to consume a } that belongs to an enclosing object, leading to incorrect parse results or errors downstream. While this built-in JSON parser is only used when Jason is unavailable, it's still a correctness issue in the parser logic.

Suggested change

	defp decode_array(<<"}", rest::binary>>, acc), do: {Enum.reverse(acc), rest}
	defp decode_array(<<"]", rest::binary>>, acc), do: {Enum.reverse(acc), rest}

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🔴 await early-return bypasses tree-completion check, returning before children complete

When await/2 is called on an event whose event_status is already :completed or :error, the early-return path at abxbus-elixir/lib/abxbus.ex:100-109 returns immediately without checking whether the event's children have also completed. This violates the documented contract: "Blocks the calling process until the event (and all its children) are complete."

How the bug gets triggered

The normal (non-early) path works correctly — waiters registered before the event completes only get notified via notify_waiters in do_mark_tree_complete (abxbus-elixir/lib/abxbus/bus_server.ex:702-704), which checks that all children are terminal before notifying. But the early-return path at line 100 only checks the event's own status, not its children.

Triggering scenario:

1. Parent handler emits a child event without awaiting it (fire-and-forget)
2. Parent handler returns → parent's event_status set to :completed
3. maybe_mark_tree_complete finds children NOT complete → does NOT call notify_waiters
4. External code calls await(parent) → reads :completed status → returns immediately
5. Child is still running but the caller already got the "completed" event

Prompt for agents

The early-return path in Abxbus.await/2 (abxbus.ex:99-109) checks only the event's own status to decide whether to return immediately, but doesn't verify that all children have also completed. The tree-aware notification logic in BusServer.do_mark_tree_complete (bus_server.ex:686-714) correctly waits for children, but it only controls the notification-based path. The early-return path bypasses it entirely.

To fix this, the early-return check at line 100 needs to also verify that all children of the event are in a terminal state before returning immediately. One approach: add a helper like `tree_complete?(event_id)` that checks EventStore.children_of and recursively verifies all descendants are terminal, and include it in the guard condition. If the tree is NOT complete, fall through to the waiter-based path instead of returning early.

The same issue exists in `event_completed/2` at abxbus.ex:142-149 which has the same early-return pattern.

---

Was this helpful? React with 👍 or 👎 to provide feedback.