✨ Add WebSocket server to @effectionx/websocket - #222
Open
taras wants to merge 16 commits into
Open
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a WebSocket server stream API, bounded close operations, and awaitable ChangesWebSocket server and lifecycle updates
Estimated code review effort: 3 (Moderate) | ~30 minutes Sequence Diagram(s)sequenceDiagram
participant ServerFactory
participant WebSocketServer
participant ConnectionQueue
participant WebSocketResource
participant Client
ServerFactory->>WebSocketServer: create server
Client->>WebSocketServer: open connection
WebSocketServer->>WebSocketResource: wrap accepted socket
WebSocketServer->>ConnectionQueue: buffer connection
ConnectionQueue-->>Client: deliver connection resource
Client->>WebSocketResource: send message
WebSocketResource-->>Client: deliver response
Important Pre-merge checks failedPlease resolve all errors before merging. Addressing warnings is optional. ❌ Failed checks (1 error)
✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
commit: |
taras
commented
Jul 6, 2026
taras
commented
Jul 6, 2026
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@websocket/server.test.ts`:
- Around line 117-120: The test is accessing `.data` on `firstMessages.next()`
and `secondMessages.next()` without narrowing the iterator result, but the
`.value` can also be a `CloseEvent`. Update the `received` setup in
`server.test.ts` to first unwrap each `IteratorResult` and assert/narrow that
the yielded value is a `MessageEvent<string>` before reading `.data`, using the
existing `firstMessages` and `secondMessages` iterators so the type checker can
distinguish them from `CloseEvent`.
In `@websocket/server.ts`:
- Around line 85-140: Add a test for connection-level error isolation in
useWebSocketServer: verify that a single accepted socket error after the
connection is established does not take down the server or other active
connections. Use the useWebSocketServer, onConnection, and the per-connection
useWebSocket<T> task behavior to assert the errored socket is isolated while the
server keeps accepting/serving remaining connections.
- Around line 108-119: The per-connection handling in onConnection currently
runs inside scope.run without its own failure boundary, so errors from
useWebSocket or the subscription drain loop can bubble up and affect the server.
Wrap the body of onConnection in a connection-local error boundary or
try/catch-style effect handler so each raw WebSocket failure is isolated, and
make sure the connection setup, connections.add(connection), and
subscription.next() loop all stay within that per-client scope.
In `@websocket/websocket.ts`:
- Around line 161-163: The public send method in websocket.ts is relying on
contextual typing for both its parameter and its Operation<void> return, so make
the types explicit on the send method in the websocket object literal. Update
the send signature in the websocket factory/adapter implementation to declare
the data parameter type and the void-returning operation type directly, keeping
the method compatible with the surrounding interface while following the
explicit public function typing guideline.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 2406db94-4af3-4cb6-8f23-3452e452d764
📒 Files selected for processing (7)
websocket/README.mdwebsocket/mod.tswebsocket/package.jsonwebsocket/server.test.tswebsocket/server.tswebsocket/websocket.test.tswebsocket/websocket.ts
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
websocket/server.ts (1)
85-103: 🚀 Performance & Scalability | 🔵 TrivialUnbounded connection buffering could exhaust resources under connection floods.
createQueueis unlimited in size, and the accept loop wraps and stores every incoming socket viauseWebSocketregardless of whether anything ever reads fromserver.socket. If the consumer stalls or never iterates the stream, the server keeps accepting and holding live sockets (each with its own background tasks/listeners) indefinitely — a stalled or malicious client population could grow memory/FD usage without bound. Worth considering a cap on buffered connections (or backpressure signal) as a hardening follow-up.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@websocket/server.ts` around lines 85 - 103, The accept path in websocket/server.ts can buffer an unlimited number of live WebSocketResource entries, which risks unbounded memory and FD growth if server.socket is not being consumed. Update the createQueue/useWebSocket flow in the server accept loop to enforce a bounded connection buffer or apply backpressure so new sockets are not wrapped and retained indefinitely when the consumer stalls. Use the existing spawn, each(on(server, "connection")), and connections.add logic as the place to gate or reject excess connections.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@websocket/server.ts`:
- Around line 85-103: The accept path in websocket/server.ts can buffer an
unlimited number of live WebSocketResource entries, which risks unbounded memory
and FD growth if server.socket is not being consumed. Update the
createQueue/useWebSocket flow in the server accept loop to enforce a bounded
connection buffer or apply backpressure so new sockets are not wrapped and
retained indefinitely when the consumer stalls. Use the existing spawn,
each(on(server, "connection")), and connections.add logic as the place to gate
or reject excess connections.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: e247a1c9-f01f-47c8-9884-ac73b73888d2
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (5)
websocket/package.jsonwebsocket/server.test.tswebsocket/server.tswebsocket/tsconfig.jsonwebsocket/websocket.ts
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
websocket/server.test.ts (1)
24-36: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider extracting the repeated "accept a connection" boilerplate.
The
let { server, port } = yield* useTestServer(); let incoming = yield* server; let client = yield* connect(port); let connection = (yield* incoming.next()).value;sequence repeats across most tests in this file. A smallacceptConnection()helper (alongside the existinguseTestServer/connect/drainhelpers) would reduce duplication and keep new tests concise.Also applies to: 38-50, 101-113, 133-166
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@websocket/server.test.ts` around lines 24 - 36, The test setup in server.test.ts repeats the same “start server, wait for incoming, connect client, accept connection” boilerplate in multiple cases. Extract that sequence into a small helper such as acceptConnection() near the existing useTestServer, connect, and drain helpers, and update the affected tests to use it so the websocket tests stay concise and consistent.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@websocket/package.json`:
- Around line 17-19: The websocket package currently depends on the top-level
`@effectionx/node` entrypoint, which pulls in Node-only stream code through its
reexports and breaks browser/Deno consumers. Update the websocket package to
avoid importing the full `@effectionx/node` package and instead reference the
narrower `@effectionx/node/events` entrypoint, or separate the event helpers from
the Node stream adapter so `@effectionx/websocket` stays environment-agnostic.
In `@websocket/server.ts`:
- Around line 159-161: The teardown loop in the server shutdown path currently
closes each live connection sequentially via `connection.close(...)`, which can
stretch shutdown to N times the close timeout. Update the shutdown logic to run
the per-connection close effects concurrently using Effection’s `all` in the
`server.ts` teardown flow, and add `all` to the Effection import. Keep the
existing `close(1001, "server shutting down")` behavior inside the
per-connection task, and verify that the `close()` calls remain safe when
executed in parallel.
---
Outside diff comments:
In `@websocket/server.test.ts`:
- Around line 24-36: The test setup in server.test.ts repeats the same “start
server, wait for incoming, connect client, accept connection” boilerplate in
multiple cases. Extract that sequence into a small helper such as
acceptConnection() near the existing useTestServer, connect, and drain helpers,
and update the affected tests to use it so the websocket tests stay concise and
consistent.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 8430e035-504d-4dd1-aea7-18e16862e950
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (6)
websocket/package.jsonwebsocket/server.test.tswebsocket/server.tswebsocket/tsconfig.jsonwebsocket/websocket.test.tswebsocket/websocket.ts
cowboyd
requested changes
Jul 7, 2026
cowboyd
reviewed
Aug 7, 2026
cowboyd
approved these changes
Aug 7, 2026
Add useWebSocketServer(), the server counterpart of useWebSocket(). It yields a stream of incoming connections, each a full-duplex WebSocketResource, so client and server share the same handle type. The underlying server is supplied via a factory, keeping the package free of any concrete server dependency and platform-agnostic. BREAKING CHANGE: WebSocketResource.send() is now an Operation, invoked as `yield* socket.send(...)` on both client and server. Bumped to 3.0.0.
Phase 1: replace the unbounded wait for the peer close handshake in useWebSocket with timebox(), so a silent peer can no longer hang scope teardown. Adds @effectionx/timebox.
Phase 2: WebSocketResource gains a composable close(code?, reason?) operation; useWebSocketServer composes a 1001 "server shutting down" close for live connections on teardown. First-close-wins, so it takes precedence over the scope-exit 1000.
Phase 3: each accepted connection now runs inside a scoped() error boundary, so one socket erroring is contained instead of crashing the server. Failures are surfaced compositionally on a new server.errors stream rather than a callback.
The going-away shutdown yielded inside a `finally`, which the Async Teardown policy forbids: a halt that unwinds through a yielding `finally` comes back as `iterator.next()` and the frame leaves return-mode, losing the halt. Register it with `ensure()` instead, placed after the accept spawns so it still runs while their connections are alive. Close the live connections concurrently while here — sequential closes cost one close timeout per silent peer.
Hand back connections as a Subscription, not a Stream. A stream is stateless — subscribing is what allocates state and starts the work. This server does the opposite: it listens and buffers from the moment the resource is created, and every subscriber drew from the same shared queue, so two subscribers silently stole each other's connections. Typing it as a Subscription says what it actually is. The tests already read it that way, opening with `yield* server` purely to reach a subscription; that indirection is gone. Drop the `as unknown as WebSocketServerLike` cast. It was carried over from the client's `ws as unknown as WebSocket` cast, but a `ws` WebSocketServer already satisfies the interface structurally — the cast was never needed and only made the API look worse than it is. Name the two collections for their jobs: `accepted` is the delivery buffer that drains as connections are read, `live` is the roster of open connections closed on shutdown. A connection is in both until read. Stop describing `scoped` in terms of trap/delimiter, which are private concepts, and move the test's HTTP-server teardown out of a yielding `finally` into `ensure()` per the Async Teardown policy.
Matches how the rest of the repo consumes the package (see process/) and narrows the import to the entrypoint actually used, rather than pulling the Node stream adapter in through the barrel.
taras
force-pushed
the
feat/websocket-server
branch
from
August 12, 2026 04:17
73f7496 to
9590f22
Compare
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@websocket/package.json`:
- Line 17: Update the package.json files field to include dist, mod.ts,
server.ts, and websocket.ts, preserving dist and ensuring the package entry
point and runtime source files are published.
In `@websocket/README.md`:
- Around line 61-64: Update the useWebSocketServer() README example to consume
and monitor its errors channel in a separate task. Document that
connection-specific failures are reported there, while server-level errors
terminate the resource scope, so examples handle both outcomes.
In `@websocket/server.ts`:
- Around line 163-175: Update the teardown generator around server.close() to
await its completion callback before returning, using effection’s withResolvers
to bridge the callback into the existing yield-based flow. Preserve the current
concurrent connection closing and call server.close() only after those closes
complete.
- Around line 169-173: Update the shutdown path in the live-connection cleanup
around the connection.close call to use a browser-valid WebSocket close code,
replacing 1001 with 1000 or an application code in the 3000–4999 range while
preserving the existing shutdown reason and connection iteration.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 9c2a2d67-4d0a-4f08-a777-154f6da2d4b3
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (6)
websocket/README.mdwebsocket/package.jsonwebsocket/server.test.tswebsocket/server.tswebsocket/websocket.test.tswebsocket/websocket.ts
The isolation test asserted that the value on `errors` is the DOM `error` event, which held only up to effection 4.0. Since 4.1, `Err()` boxes a thrown non-Error in a `ThrownValueError` whose message is `String(value)` — "[object Object]" for an event — and keeps the original on `cause`. Rebasing onto main pulled effection 4.1.0 in, so the assertion started failing on CI. `useWebSocket` throws the raw event deliberately, for parity with the client, so unwrap at the reading end instead: prefer `cause`, fall back to the value. The peer range is `^3 || ^4` and the matrix exercises both ends, so both shapes have to keep working. Document the same on the `errors` stream, which promised a shape it no longer delivers.
`server.close()` takes a completion callback that was ignored, so teardown returned before the listening socket was released and a resource binding the same port next could lose the race with EADDRINUSE. The test's own `useHttp` helper already awaited its close callback, so the two paths disagreed. Also document what a close code may be. `close()` passes the code straight to the socket, and the legal set depends on the implementation: the WHATWG API allows only 1000 and 3000-4999 and throws InvalidAccessError otherwise, while `ws` takes the full RFC 6455 range. Shutdown's 1001 works on `ws` but would throw on a WHATWG-conformant socket, so say so on both the option and `WebSocketServerLike` rather than quietly changing what the peer observes.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
websocket/server.ts (1)
122-131: 🚀 Performance & Scalability | 🔵 TrivialDefine an overflow policy for
accepted.
createQueue()is unbounded, andaccepted.add()does not apply backpressure. If the consumer stalls, pendingWebSocketResourcevalues retain their sockets and can exhaust memory or file descriptors. Add admission control or close overflow connections, and document the limit.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@websocket/server.ts` around lines 122 - 131, Define a documented capacity for the accepted queue and enforce it before adding values in the resource’s connection-admission flow. When the limit is reached, apply an explicit overflow policy—either wait for capacity or close and remove the excess WebSocketResource—so stalled consumers cannot retain unbounded sockets; preserve normal accepted/live tracking for admitted connections.websocket/websocket.ts (1)
13-20: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winValidate
closeTimeoutbefore callingtimebox.
closeTimeoutreachestimeboxunchanged. RejectNaN,Infinity, and negative values, and document that0skips the close-handshake wait. Add tests for these boundaries.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@websocket/websocket.ts` around lines 13 - 20, Validate closeTimeout before passing it to timebox, rejecting NaN, Infinity, and negative values while accepting 0 as the value that skips the close-handshake wait. Update the closeTimeout documentation in UseWebSocketOptions to describe this behavior, and add boundary tests covering invalid values and zero.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@websocket/server.ts`:
- Around line 122-131: Define a documented capacity for the accepted queue and
enforce it before adding values in the resource’s connection-admission flow.
When the limit is reached, apply an explicit overflow policy—either wait for
capacity or close and remove the excess WebSocketResource—so stalled consumers
cannot retain unbounded sockets; preserve normal accepted/live tracking for
admitted connections.
In `@websocket/websocket.ts`:
- Around line 13-20: Validate closeTimeout before passing it to timebox,
rejecting NaN, Infinity, and negative values while accepting 0 as the value that
skips the close-handshake wait. Update the closeTimeout documentation in
UseWebSocketOptions to describe this behavior, and add boundary tests covering
invalid values and zero.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 16f5f460-6cf1-43ff-95ac-5e62888575bb
📒 Files selected for processing (4)
websocket/README.mdwebsocket/server.test.tswebsocket/server.tswebsocket/websocket.ts
Three termination properties the suite could not see, each checked by mutating the implementation and confirming the test fails: - Every live connection gets the going-away close, not just the first. The existing teardown test used a single client, so closing only `[...live][0]` still passed it. - Teardown does not complete until the server has finished closing. - Teardown stays bounded when peers never answer the close handshake, and the going-away 1001 wins over the scope-exit 1000. The last two drive a hand-rolled server, since a real peer cannot be made to withhold a close frame or defer a close callback on demand. That exposed two things worth writing down: connections emitted in the same tick as resource creation are missed because the accept loop has not subscribed yet (harmless for a real server, whose sockets arrive via I/O), and a fake socket has to leave `readyState` OPEN behind for the "first close wins" rule to mean anything. Note the port-rebinding hazard that motivated awaiting the close callback is not reproducible: Node frees the listening socket during `close()`, so rebinding succeeds either way. The callback reports when connections have finished, which is what the test asserts instead.
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.
Motivation
The
@effectionx/websocketpackage shipped only a client (useWebSocket()). There was no server counterpart, even though the package's own test file already implemented the full server-side mechanic by hand — awsWebSocketServerfiringconnectionevents, each raw socket wrapped withuseWebSocket(). This PR formalizes that pattern into a reusable server that pairs naturally with the existing client.Approach
New
useWebSocketServer()(websocket/server.ts) — the server counterpart ofuseWebSocket():WebSocketServerResource<T>: an EffectionSubscriptionwhose items are the same full-duplexWebSocketResourcethe client produces, so both sides share one handle type (iterate to receive,yield* connection.send()to reply).useWebSocket()to wrap each incoming socket.createQueue, so none are dropped between the server starting to listen and the consumer reading.error, mirroring the client.scopedso one bad socket does not take down the server, publishing what it threw on anerrorsstream.1001("going away").() => WebSocketServerLikefactory, so the package never imports a concrete server implementation and stays platform-agnostic. AwsWebSocketServersatisfies the interface structurally and can be passed with no cast.sendis now anOperation— invoked asyield* resource.send(...)on both client and server, keeping the sharedWebSocketResourcesymmetric and letting sends participate in structured concurrency.close(code?, reason?)is new onWebSocketResource, bounded by a configurablecloseTimeout(default1000ms) so a silent peer can never hang teardown.Warning
Breaking change:
WebSocketResource.send()changed from a synchronousvoidcall to anOperation, so callers must now writeyield* socket.send(...). Version bumped2.3.4 → 3.0.0.This one fails silently.
sendis a generator method now, so an un-yield*edsocket.send(data)still typechecks — discarding a return value is legal — but the body never runs and the message is never sent. There is no type error and no runtime error to catch it, so every call site has to be updated by hand. Everything else in the release is additive:useWebSocketServer,WebSocketServerLike,WebSocketServerResource,UseWebSocketOptions,close(), and the new optionaluseWebSocketparameters.Dependencies: this adds two workspace runtime dependencies —
@effectionx/node(foron/onceover the server'sEventEmitter, imported via the@effectionx/node/eventssubpath) and@effectionx/timebox(to bound the close handshake).wsremains a devDependency for tests only.Also included:
mod.tsre-exports the server; updated client README/tests toyield*their sends.server.test.ts(8 tests): connection delivery, message round-trips in both directions, close-propagation on client disconnect, server teardown closing live clients, explicit close code/reason, buffering before the first read, per-connection error isolation, and multiple simultaneous clients as distinct connections.Why a
Subscriptionrather than aStreamPer review: a
Streamis stateless — subscribing to it is what allocates state and starts the work. A server is the opposite. It begins listening and buffering the moment the resource is created, and every consumer draws from that one shared queue, so handing out aStreamimplied an independent replay per subscriber that does not exist. Typing it as aSubscriptionstates the real contract: reading a connection consumes it.Verification
wsserver against a real nativeWebSocketclient, exercising the pairing end-to-end.tsc -bandtsc -p tsconfig.check.jsonclean,biome lint/formatclean, tsconfig references in sync.Summary by CodeRabbit