Standard library / primitives

WebSocket client

Anything that streams — an RPC subscription, an exchange's order feed, a Discord/Slack gateway, the mempool — used to mean cron + polling. The WebSocket client replaces that with a live connection. It is a general transport, not a blockchain feature: it lives in the standard library next to the HTTP client, and it is gated by the same net(host) capability and scope.

It is not a minimal one-connection-at-a-time client: it multiplexes thousands of feeds in a single thread (readiness-driven via epoll/kqueue/IOCP — ~0 CPU while idle), reconnects on its own with bounded backoff and resubscribe, detects half-open connections via keepalive, and bounds memory with backpressure. The two axes of scale: ws_select (vertical fan-in, N feeds in 1 thread) and parallel_map (horizontal fan-out, N workers × 1 connection).

websocket.syn

-- Doc example: the WebSocket client is a GENERAL transport (RPC subscriptions,
-- exchange feeds, chat, mempool — anything that streams), gated by the SAME
-- net(host) capability as HTTP. A connection to an undeclared host is refused at
-- the capability check, before any socket opens. (Opening a live feed needs real
-- net, which the docs sandbox denies — so this example proves the gate; the live
-- round-trip and the WS→SSE bridge are shown in the page prose.)
intent: "doc example: WebSocket client capability gating"
require net("stream.allowed.com")
-- a grant with a port covers that port only
require net("127.0.0.1:2")
require secret("WS_KEY") for net("stream.allowed.com")    -- v0.6.46+: the key may go only there

task connect_undeclared()
    give ws_connect("wss://feed.evil.example.com/ws")   -- host not declared → denied

task bad_scheme()
    give ws_connect("https://stream.allowed.com/ws")    -- WebSocket needs ws:// or wss://

test "a WebSocket to an undeclared host is blocked (same net gate as HTTP)"
    assert_error(connect_undeclared)

task connect_other_port()
    give ws_connect("ws://127.0.0.1:3/ws")              -- net("127.0.0.1:2") does not cover :3

test "an explicit port is part of the grant"
    assert_error(connect_other_port)

test "the scheme must be ws:// or wss:// — an http(s) URL is a directed error"
    assert_error(bad_scheme)

-- ws_select multiplexes many feeds in one thread (Go's `select {}` in
-- one call). Its opts are validated up front — a typo doesn't become a silent
-- misconfiguration. These are PURE checks (no live server; the multiplex, reconnect
-- and keepalive round-trips are exercised against a local mock in the test suite).

task bad_on_full()
    -- on_full only accepts "block" (default), "drop_oldest", "error"
    give ws_connect("wss://stream.allowed.com/ws", nothing, {"on_full": "explode"})

task unknown_opt()
    give ws_connect("wss://stream.allowed.com/ws", nothing, {"reconect": {}})   -- typo

test "ws_select over an empty set is nothing (no feeds → nothing to wait for)"
    assert_eq(ws_select([], 0), nothing)

test "ws_status of an unknown handle is \"closed\" (never lies, never panics)"
    assert_eq(ws_status(999999), "closed")

test "connect opts are validated: a bad on_full / an unknown option error clearly"
    assert_error(bad_on_full)
    assert_error(unknown_opt)

-- An EVM node's eth_subscribe is plain JSON-RPC over the socket: each notification is
-- a JSON text whose params.result is a raw log ("0x…" topics and data). Decoding it
-- is pure — the same abi_decode_log that reads evm_logs and receipts.
test "an eth_subscribe log notification decodes by parameter name"
    let transfer be {"name": "Transfer", "inputs": [
        {"name": "from", "type": "address", "indexed": true},
        {"name": "to", "type": "address", "indexed": true},
        {"name": "value", "type": "uint256", "indexed": false}]}
    -- the subscribe frame you would ws_send
    let frame be json_encode({"jsonrpc": "2.0", "id": 1, "method": "eth_subscribe",
        "params": ["logs", {"topics": [abi_event_topic(transfer)]}]})
    assert(contains(frame, "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"))
    -- a notification as ws_recv hands it over (msg["data"])
    let data be json_encode({"jsonrpc": "2.0", "method": "eth_subscription",
        "params": {"subscription": "0x9c", "result": {
            "address": "0x6fac4d18c912343bf86fa7049364dd4e424ab9c0",
            "topics": [abi_event_topic(transfer),
                "0x0000000000000000000000007e5f4552091a69125d5dfcb7b8c2659029395bdf",
                "0x0000000000000000000000005aaeb6053f3e94c9b9a09f33669435e7ef1beaed"],
            "data": "0x00000000000000000000000000000000000000000000000000000000000003e8"}}})
    let note be json_decode(data)
    assert_eq(note["method"], "eth_subscription")
    let ev be abi_decode_log(transfer, note["params"]["result"], nothing)
    assert_eq(ev.from, "0x7E5F4552091A69125d5DfCb7b8C2659029395Bdf")
    assert_eq(ev.value, 1000)

test "a key in the URL is a secret bound to the host: another host is refused before connecting (engine v0.6.46+)"
    let k be secret("WS_KEY", "wk-demo-123")
    let failed be ""
    try
        ws_connect(`ws://127.0.0.1:2/v1/{k}`)
    recover e
        set failed to e
    assert(contains(failed, "secret(WS_KEY) is bound to [stream.allowed.com]"))

The capability model (no new door)§

ws_connect needs net(host) — deny-by-default, scoped by hostname, exactly like http_get. There is no WebSocket-specific permission: a WebSocket is transport, gated like any other network egress.

Undeclared host → denied at the capability check, before any socket opens.
Inside a sandbox (capabilities emptied) → denied.
wss:// validates the server certificate against the OS root CAs, just like https://.
A grant with a port covers that port only: require net("node.example.com:8546") does not cover :8547; net("node.example.com") covers any port.
A wildcard grant reaches only public addresses (engine v0.6.46+): with net("*"), ws_connect("ws://127.0.0.1:8546/") is the error `ws_connect(): 127.0.0.1 resolved to 127.0.0.1, a private address; a net grant with a wildcard reaches only public addresses. Name it: require net("127.0.0.1:8546")…` — checked again on every reconnection. Ranges and how to name a private host: HTTP client.
A key in the URL (engine v0.6.46+): bind the secret to the host — require secret("K") for net("stream.example.com") — and ` ws_connect(wss://stream.example.com/v1/{k}) works; a text frame the server sends back with the key in it arrives masked as [secret K]`. Unbound, the URL is an error that gives that line. See Secrets.

require net("stream.exchange.com")
let conn be ws_connect("wss://stream.exchange.com/ws")     -- opaque handle

Send, receive, close§

ws_send(conn, json_encode({"op": "subscribe", "channel": "trades"}))  -- text or bytes
let msg be ws_recv(conn, 5)          -- next message, or `nothing` after the 5s timeout
-- msg is {"type": "text" | "binary" | "close", "data": …}
when msg != nothing and msg["type"] == "text"
    print(msg["data"])
ws_close(conn)                       -- clean close frame (idempotent)

ws_connect(url, headers?, opts?) → an opaque handle (like a db_open connection). headers is a text→text map — a secret value (e.g. bearer(...)) materializes only at the socket, exactly like HTTP headers. opts = {"timeout", "max_message_size", "subprotocols", "max_queue", "max_queue_bytes", "on_full", "reconnect", "keepalive"} (see below).
ws_recv(conn, timeout?) → the message map, or nothing on timeout (default 30s, the same criterion as wait_for). It never blocks forever; ping/pong frames are handled transparently. A peer that resets the connection arrives as {type: "close", data: "connection reset by peer"} (engine v0.6.45+), not as an error.
ws_send(conn, data) sends text (if data is text) or binary (if bytes). A secret is refused — reveal() it first if you truly must.
ws_close(conn) sends a clean close frame; closing an already-closed handle is a no-op.

Multiplexing thousands of feeds: `ws_select` (the event loop)§

A ws_recv waits on ONE connection. To watch N feeds without burning a thread per connection, ws_select waits for the first one with data — Go's select {} over channels, in a single call. It is readiness-driven (epoll/kqueue/IOCP via mio): ~0 CPU while idle (it sleeps in the kernel, never busy-spins) and scales to thousands.

require net("feed.example.com")
let feeds be {"trades": ws_connect("wss://feed.example.com/trades"),
              "book": ws_connect("wss://feed.example.com/book")}   -- name→handle map
let live be true
while live
    let m be ws_select(feeds, 30)          -- first ready of ALL feeds, or nothing on timeout
    when m == nothing
        set live to false                   -- 30s idle → stop (tune to your feed)
    when m != nothing
        -- m adds "conn" (WHICH handle fired) and, for a map, "name"
        when m["type"] == "close"
            print("feed " + m["name"] + " dropped")   -- you know exactly which to resubscribe
        when m["type"] == "text"
            print(m["name"] + ": " + m["data"])

ws_select(conns, timeout?) → {conn, type, data, name?} of the first ready connection, or nothing on timeout. conns is a list of handles or a name→handle map (then the result carries name). A dropped connection surfaces as {type: "close", conn} and is retired — you know WHICH to resubscribe. A peer that goes away without closing (a reset: the process died, the network dropped; os error 10054 on Windows) is a close too, {type: "close", data: "connection reset by peer", conn} (engine v0.6.45+; before, it was an error that also cut the wait of every other connection in the same ws_select). A protocol violation (an invalid frame, a message over the size limit) is still a catchable error (with conn).
ws_select_all(conns, timeout?) → a list of every message ready this tick (one per connection; batch processing).
ws_broadcast(conns, data) → send the same message to many connections at once → returns the count.

Resilience: reconnection and keepalive (opt-in)§

Without these options the behavior is unchanged — nothing silent. With them, a long-lived agent survives drops and detects dead sockets without userland plumbing.

let conn be ws_connect("wss://feed.example.com/ws", nothing, {
    "reconnect": {"max_retries": 10, "backoff": 0.5, "on_reconnect": resubscribe},
    "keepalive": {"interval": 20, "timeout": 10}})

reconnect = {max_retries?, backoff?, backoff_max?, on_reconnect?} — transparent reconnection with bounded exponential backoff. on_reconnect is a task that runs after reconnecting (it receives the handle) — resubscribe there so you never lose your feed state. Every reconnect re-checks net(host) (never escalates scope) and wss re-validates the cert. Without reconnect, a drop surfaces as close.
keepalive = {interval, timeout?} — auto-ping every interval; no pong within timeout → the connection is dead → reconnect (if enabled) or close. Half-open detection most libraries skip.
ws_status(conn) → "open" | "reconnecting" | "closed". ws_stats(conn) → {sent, received, reconnects, queued, queued_bytes, last_pong_ago, status, subprotocol} — so a keeper of thousands of feeds can see itself (aligns with observability). Stats never lie: sent counts only messages that actually went out (or got queued to flush); last_pong_ago is seconds since the last pong (nothing if none yet). Any inbound traffic counts as liveness — a pong delayed behind a busy feed never falsely kills a live connection.

The sync-engine boundary. There is no background thread per connection (that would break CSP isolation and the "no thread-per-connection" promise). Keepalive and reconnection advance while the program is inside ws_select/ws_recv/ws_status — which is why the pattern is a while loop with ws_select. A lone ws_send with no following recv won't tick the timers.

Fan-out with `parallel_map`: N workers × 1 connection§

ws_select is the vertical axis (many feeds, one thread). The horizontal axis is parallel_map: each worker gets its own interpreter (inheriting caps) and its own WebSocket registry — it opens its feed, processes it, returns. Handles do not cross workers (CSP isolation): never share a handle between workers.

task watch(url)
    let c be ws_connect(url)
    let m be ws_recv(c, 30)
    ws_close(c)
    give m
let results be parallel_map(watch, thousands_of_urls)   -- thousands of feeds across the pool

It never hangs, can't be flooded, never busy-spins§

A dead connection never hangs the agent: connect, the handshake, and every ws_recv/ws_select are bounded by the timeout — lapsed → nothing, not a stuck program. Memory is bounded under hostile input: each message is capped (default 16 MiB, ceiling 64 MiB) and the per-connection inbound queue is bounded in both dimensions — messages (max_queue, default 1024) and bytes (max_queue_bytes, default 64 MiB) — so a flood of big frames can't OOM you any more than a flood of tiny ones. The default policy when either cap is hit is real TCP backpressure: it stops reading the socket, the TCP window throttles the peer — no data loss, no OOM. The alternatives are explicit: "drop_oldest" evicts the oldest until the new one fits, and "error" drains what's already queued and then surfaces a catchable error naming the overflow — never a silent drop. Delivery across feeds is round-robin fair: a chatty connection with a backlog cannot starve the others. And an idle ws_select sleeps in the kernel (a single readiness wait), it does not spin. A soft per-interpreter connection cap (SYNSEMA_WS_MAX_CONNS, default 4096) stops a runaway loop from opening 100k sockets.

Subprotocols§

opts.subprotocols (a list) negotiates the Sec-WebSocket-Protocol header; the agreed subprotocol lands in ws_stats(conn)["subprotocol"].

let c be ws_connect("wss://feed.example.com/ws", nothing, {"subprotocols": ["json", "cbor"]})
let agreed be ws_stats(c)["subprotocol"]   -- "json" if the server chose it

Under `serve`: the WS→SSE bridge§

The most common server pattern is a bridge: a route opens an outbound WebSocket to an RPC or exchange, then forwards what it receives to the browser over SSE. This is one of the few things you can't fully show in run/test (it needs a running server and a live upstream), so here is the shape:

require serve(8080)
require net("stream.exchange.com")

serve on 8080
    route "GET /live"
        let conn be ws_connect("wss://stream.exchange.com/ws")
        ws_send(conn, json_encode({"op": "subscribe", "channel": "trades"}))
        stream                        -- Server-Sent Events to the browser
            let live be true
            while live
                let msg be ws_recv(conn, 30)
                when msg == nothing or msg["type"] == "close"
                    set live to false
                when msg != nothing and msg["type"] == "text"
                    send msg["data"]  -- forward each upstream message downstream
        ws_close(conn)

The browser opens one plain EventSource("/live"); the server holds the upstream WebSocket. Incoming WebSocket connections (a WS server) are a route … socket block since engine v0.6.7 — the same ws_* family works on the incoming handle, and select waits on the upstream feed, the browser's socket, a child process and the event bus in one call: see Agentic apps.

What it unlocks§

Because it's a general primitive, it's not only for blockchain. Any Synsema program can now build on a live bidirectional stream: real-time dashboards, an agent reacting to a webhook gateway, collaborative tools, a market-data recorder, a chat client. Where before you wrote a polling loop, you now hold an open connection.

A concrete example: an EVM node's eth_subscribe (newHeads/logs) composes in userland today — ws_connect to the node's WS endpoint, ws_send the JSON-RPC subscribe frame, ws_recv + json_decode each notification, and decode each log with the event's ABI fragment (abi_decode_log, the same one that reads evm_logs and receipts):

require net("node.example.com")

let transfer be {"name": "Transfer", "inputs": [
    {"name": "from", "type": "address", "indexed": true},
    {"name": "to", "type": "address", "indexed": true},
    {"name": "value", "type": "uint256", "indexed": false}]}
let conn be ws_connect("wss://node.example.com/ws")
ws_send(conn, json_encode({"jsonrpc": "2.0", "id": 1, "method": "eth_subscribe",
    "params": ["logs", {"address": token, "topics": [abi_event_topic(transfer)]}]}))
let live be true
while live
    let msg be ws_recv(conn, 60)
    when msg == nothing or msg["type"] == "close"
        set live to false
    otherwise
        let note be json_decode(msg["data"])
        when get(note, "method") == "eth_subscription"
            -- params.result is the raw log: "0x…" topics and data
            let ev be abi_decode_log(transfer, note["params"]["result"], nothing)
            when ev != nothing
                print(ev.from + " → " + ev.to + ": " + text(ev.value))
ws_close(conn)

eth_subscribe is the node's JSON-RPC method name, so it keeps the eth_ prefix. One-shot reads (nonce, fees, balances, a range of logs, receipts) go through the typed blockchain read side.