Add TypeScript implementation and comprehensive test suite#23
Closed
pirate wants to merge 115 commits intobrowser-use:mainfrom
Closed
Add TypeScript implementation and comprehensive test suite#23pirate wants to merge 115 commits intobrowser-use:mainfrom
pirate wants to merge 115 commits intobrowser-use:mainfrom
Conversation
Add support for middlewares to hook into event bus handler lifecycle
implement swappable EventHistory storage backend
Updated the description to clarify the library's functionality and similarities to JS event systems.
Revise README description for bubus library
…ead of process-until-event
Add `@retry` decorator to bubus-ts
Verifies that multiple events can be emitted and raced to completion via Promise.all([e1.done(), e2.done(), e3.done()]) across three scenarios: parallel concurrency, bus-serial concurrency, and inside a handler via queue-jump. https://claude.ai/code/session_01FFkpZPxvvoexboUF2Uo4oQ
pirate
force-pushed
the
claude/test-parallel-events-3720c
branch
from
46ab953 to
7c3f883
Compare
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
8 issues found across 65 files
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="bubus-ts/tests/error_handling.test.ts">
<violation number="1" location="bubus-ts/tests/error_handling.test.ts:172">
P2: The test excludes forwarding handlers by checking `handler_name !== 'dispatch'`, but bound functions are named `"bound dispatch"`. This can select the forwarding handler instead of the bus A handler, making the test brittle.</violation>
</file>
<file name="bubus-ts/src/lock_manager.ts">
<violation number="1" location="bubus-ts/src/lock_manager.ts:71">
P2: AsyncSemaphore.release() decrements in_use before waking a waiter, allowing a new acquire() to slip in and exceed the semaphore size. This can break mutual exclusion under contention.</violation>
</file>
<file name="bubus-ts/prettier.config.js">
<violation number="1" location="bubus-ts/prettier.config.js:8">
P2: `prettier.config.js` uses ESM `export default` but the repo lacks a `package.json` with `"type": "module"`, so Prettier will treat this file as CommonJS and fail to load it. Use CommonJS export or add `type: module`.</violation>
</file>
<file name="bubus-ts/package.json">
<violation number="1" location="bubus-ts/package.json:28">
P2: Test script uses POSIX-style inline env var assignment, which fails under Windows cmd.exe (npm default), preventing tests from running cross-platform.</violation>
</file>
<file name="README.md">
<violation number="1" location="README.md:16">
P3: README Python quickstart uses `bus.emit` and `SomeEvent({some_data: 132})`, but Python `EventBus` exposes `dispatch()` and events are constructed with keyword args. This snippet is not runnable and will mislead users.</violation>
</file>
<file name="bubus/service.py">
<violation number="1" location="bubus/service.py:628">
P2: Backpressure uses event_history for pending counts, but cleanup_event_history can evict started/pending events when history exceeds max_history_size. That undercounts inflight load and allows dispatch to accept more than the 100‑pending limit, defeating the backpressure mechanism.</violation>
<violation number="2" location="bubus/service.py:654">
P2: PENDING lifecycle hook is fired in a background task while STARTED/COMPLETED are awaited, so middleware can observe STARTED/COMPLETED before PENDING. This introduces out‑of‑order lifecycle events and a race in middleware state transitions.</violation>
</file>
<file name="bubus-ts/src/types.ts">
<violation number="1" location="bubus-ts/src/types.ts:67">
P2: `getStringTypeName` reads `_def.type`, but Zod stores the schema kind in `_def.typeName`. This makes `kind` empty for normal schemas and forces `'unknown'` for all type names, breaking `event_result_type` inference.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
|
|
||
| // bus_a's handler succeeded | ||
| const bus_a_result = Array.from(event.event_results.values()).find( | ||
| (r) => r.eventbus_name === 'ErrorForwardA' && r.handler_name !== 'dispatch' |
There was a problem hiding this comment.
P2: The test excludes forwarding handlers by checking handler_name !== 'dispatch', but bound functions are named "bound dispatch". This can select the forwarding handler instead of the bus A handler, making the test brittle.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/tests/error_handling.test.ts, line 172:
<comment>The test excludes forwarding handlers by checking `handler_name !== 'dispatch'`, but bound functions are named `"bound dispatch"`. This can select the forwarding handler instead of the bus A handler, making the test brittle.</comment>
<file context>
@@ -0,0 +1,221 @@
+
+ // bus_a's handler succeeded
+ const bus_a_result = Array.from(event.event_results.values()).find(
+ (r) => r.eventbus_name === 'ErrorForwardA' && r.handler_name !== 'dispatch'
+ )
+ assert.ok(bus_a_result)
</file context>
bubus-ts/src/lock_manager.ts
Outdated
| if (this.size === Infinity) { | ||
| return | ||
| } | ||
| this.in_use = Math.max(0, this.in_use - 1) |
There was a problem hiding this comment.
P2: AsyncSemaphore.release() decrements in_use before waking a waiter, allowing a new acquire() to slip in and exceed the semaphore size. This can break mutual exclusion under contention.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/src/lock_manager.ts, line 71:
<comment>AsyncSemaphore.release() decrements in_use before waking a waiter, allowing a new acquire() to slip in and exceed the semaphore size. This can break mutual exclusion under contention.</comment>
<file context>
@@ -0,0 +1,377 @@
+ if (this.size === Infinity) {
+ return
+ }
+ this.in_use = Math.max(0, this.in_use - 1)
+ const next = this.waiters.shift()
+ if (next) {
</file context>
bubus-ts/prettier.config.js
Outdated
| printWidth: 140, | ||
| } | ||
|
|
||
| export default config |
There was a problem hiding this comment.
P2: prettier.config.js uses ESM export default but the repo lacks a package.json with "type": "module", so Prettier will treat this file as CommonJS and fail to load it. Use CommonJS export or add type: module.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/prettier.config.js, line 8:
<comment>`prettier.config.js` uses ESM `export default` but the repo lacks a `package.json` with `"type": "module"`, so Prettier will treat this file as CommonJS and fail to load it. Use CommonJS export or add `type: module`.</comment>
<file context>
@@ -0,0 +1,8 @@
+ printWidth: 140,
+}
+
+export default config
</file context>
bubus-ts/package.json
Outdated
| "lint": "pnpm run format:check && eslint . && pnpm run typecheck", | ||
| "format": "prettier --write .", | ||
| "format:check": "prettier --check .", | ||
| "test": "NODE_OPTIONS='--expose-gc' node --expose-gc --test --import tsx tests/**/*.test.ts", |
There was a problem hiding this comment.
P2: Test script uses POSIX-style inline env var assignment, which fails under Windows cmd.exe (npm default), preventing tests from running cross-platform.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/package.json, line 28:
<comment>Test script uses POSIX-style inline env var assignment, which fails under Windows cmd.exe (npm default), preventing tests from running cross-platform.</comment>
<file context>
@@ -0,0 +1,63 @@
+ "lint": "pnpm run format:check && eslint . && pnpm run typecheck",
+ "format": "prettier --write .",
+ "format:check": "prettier --check .",
+ "test": "NODE_OPTIONS='--expose-gc' node --expose-gc --test --import tsx tests/**/*.test.ts",
+ "prepack": "pnpm run build",
+ "release:dry-run": "pnpm publish --access public --dry-run --no-git-checks",
</file context>
bubus/service.py
Outdated
| self._active_event_ids.add(event.event_id) | ||
| if self.middlewares: | ||
| loop = asyncio.get_running_loop() | ||
| loop.create_task(self._on_event_change(event, EventStatus.PENDING)) |
There was a problem hiding this comment.
P2: PENDING lifecycle hook is fired in a background task while STARTED/COMPLETED are awaited, so middleware can observe STARTED/COMPLETED before PENDING. This introduces out‑of‑order lifecycle events and a race in middleware state transitions.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus/service.py, line 654:
<comment>PENDING lifecycle hook is fired in a background task while STARTED/COMPLETED are awaited, so middleware can observe STARTED/COMPLETED before PENDING. This introduces out‑of‑order lifecycle events and a race in middleware state transitions.</comment>
<file context>
@@ -565,71 +648,244 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
+ self._active_event_ids.add(event.event_id)
+ if self.middlewares:
+ loop = asyncio.get_running_loop()
+ loop.create_task(self._on_event_change(event, EventStatus.PENDING))
+ if logger.isEnabledFor(logging.INFO):
+ logger.info(
</file context>
bubus/service.py
Outdated
| queue_size = self.event_queue.qsize() if self.event_queue else 0 | ||
| pending_in_history = sum(1 for e in self.event_history.values() if e.event_status in ('pending', 'started')) | ||
| pending_in_history = 0 | ||
| for existing_event in self.event_history.values(): |
There was a problem hiding this comment.
P2: Backpressure uses event_history for pending counts, but cleanup_event_history can evict started/pending events when history exceeds max_history_size. That undercounts inflight load and allows dispatch to accept more than the 100‑pending limit, defeating the backpressure mechanism.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus/service.py, line 628:
<comment>Backpressure uses event_history for pending counts, but cleanup_event_history can evict started/pending events when history exceeds max_history_size. That undercounts inflight load and allows dispatch to accept more than the 100‑pending limit, defeating the backpressure mechanism.</comment>
<file context>
@@ -546,7 +624,12 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
queue_size = self.event_queue.qsize() if self.event_queue else 0
- pending_in_history = sum(1 for e in self.event_history.values() if e.event_status in ('pending', 'started'))
+ pending_in_history = 0
+ for existing_event in self.event_history.values():
+ if not self._is_event_complete_fast(existing_event):
+ pending_in_history += 1
</file context>
bubus-ts/src/types.ts
Outdated
| visited.add(value) | ||
|
|
||
| const def = (value as unknown as { _def?: Record<string, unknown> })._def ?? {} | ||
| const kind = typeof def.type === 'string' ? def.type : '' |
There was a problem hiding this comment.
P2: getStringTypeName reads _def.type, but Zod stores the schema kind in _def.typeName. This makes kind empty for normal schemas and forces 'unknown' for all type names, breaking event_result_type inference.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/src/types.ts, line 67:
<comment>`getStringTypeName` reads `_def.type`, but Zod stores the schema kind in `_def.typeName`. This makes `kind` empty for normal schemas and forces `'unknown'` for all type names, breaking `event_result_type` inference.</comment>
<file context>
@@ -0,0 +1,94 @@
+ visited.add(value)
+
+ const def = (value as unknown as { _def?: Record<string, unknown> })._def ?? {}
+ const kind = typeof def.type === 'string' ? def.type : ''
+ if (!kind) return 'unknown'
+
</file context>
README.md
Outdated
| It "just works" with an intuitive, but powerful event JSON format + dispatch API that's consistent across both languages and scales consistently from one even up to millions: | ||
| ```python | ||
| bus.on(SomeEvent, some_function) | ||
| bus.emit(SomeEvent({some_data: 132})) |
There was a problem hiding this comment.
P3: README Python quickstart uses bus.emit and SomeEvent({some_data: 132}), but Python EventBus exposes dispatch() and events are constructed with keyword args. This snippet is not runnable and will mislead users.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At README.md, line 16:
<comment>README Python quickstart uses `bus.emit` and `SomeEvent({some_data: 132})`, but Python `EventBus` exposes `dispatch()` and events are constructed with keyword args. This snippet is not runnable and will mislead users.</comment>
<file context>
@@ -1,12 +1,35 @@
+It "just works" with an intuitive, but powerful event JSON format + dispatch API that's consistent across both languages and scales consistently from one even up to millions:
+```python
+bus.on(SomeEvent, some_function)
+bus.emit(SomeEvent({some_data: 132}))
+```
+
</file context>
Suggested change
| bus.emit(SomeEvent({some_data: 132})) | |
| bus.dispatch(SomeEvent(some_data=132)) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR adds a complete TypeScript/JavaScript port of the bubus event bus library alongside the existing Python implementation, along with extensive test coverage for both languages. The TypeScript implementation maintains API parity with Python while adapting to JavaScript idioms and async patterns.
Key Changes
TypeScript Implementation
bubus-ts/src/):event_bus.ts: Full EventBus implementation with concurrency control, handler management, and event lifecyclebase_event.ts: BaseEvent class with Zod schema validation and event metadata trackingevent_result.ts: EventResult tracking for handler execution outcomesevent_handler.ts: EventHandler registration and execution managementlock_manager.ts: Concurrency primitives (semaphores, locks) supporting global-serial, bus-serial, and parallel modesasync_context.ts: AsyncLocalStorage integration for context propagationlogging.ts: Event tree visualization utilitiestypes.ts: TypeScript type definitions and utilitiesTest Coverage
Python tests (
tests/):test_find.py: 1500+ lines testing event discovery with past/future/child_of semanticstest_context_propagation.py: ContextVar propagation through dispatch and handlerstest_event_history_mirroring.py: SQLite middleware for event history persistencetest_event_result_standalone.py: EventResult model validationtest_eventbus.pyandtest_handler_timeout.pywith additional scenariosTypeScript tests (
bubus-ts/tests/):eventbus_basics.test.ts: Constructor, defaults, and basic operationsfind.test.ts: Event discovery with past/future/child_of filteringcomprehensive_patterns.test.ts: Forwarding, async/sync dispatch, parent trackingtimeout.test.ts: Handler timeouts and error metadatalocking.test.ts: Concurrency modes and FIFO correctnesscontext_propagation.test.ts: AsyncLocalStorage context propagationperformance.test.ts: 50k event throughput benchmarksInfrastructure & Documentation
Middleware system (
bubus/middlewares.py):EventBusMiddleware: Lifecycle hooks for event and result state transitionsLoggerEventBusMiddleware: Structured logging integrationWALEventBusMiddleware: Write-ahead logging for durabilitySQLiteHistoryMirrorMiddleware: Persistent event history snapshotsEvent history (
bubus/event_history.py): Bounded history container with configurable capacityTypeScript tooling:
package.json: npm configuration with zod, uuid, and test dependenciestsconfig.json: Strict TypeScript configurationeslint.config.js,prettier.config.js: Code quality toolsDocumentation:
README.mdwith multi-language positioningbubus-ts/README.md: TypeScript-specific differences and gotchasModel & Service Updates
bubus/models.py:EventStatusenum (PENDING, STARTED, COMPLETED)_event_dispatch_contextfor ContextVar capture at dispatch time_event_is_complete_flagfor completion trackingbubus/service.py:EventBusMiddlewarebase class with lifecycle hooksEventHistoryclassNotable Implementation Details
event.done()in TypeScript (vsawait eventin Python) triggers immediate processing on all buses where the event is queuedhttps://claude.ai/code/session_01FFkpZPxvvoexboUF2Uo4oQ
Summary by cubic
Adds a complete TypeScript port of the bubus event bus with API parity to Python, plus a comprehensive cross-language test suite. Also adds a handler-level retry decorator, first() for racing handlers, middleware hooks, a pluggable event history backend, unified find(), and dispatch-time context propagation in Python.
New Features
Migration
Written for commit 7c3f883. Summary will update on new commits.