Synsemadocsv0.6.xENES

Biblioteca estándar

Cliente WebSocket

Cualquier cosa que streamea — una suscripción RPC, el feed de órdenes de un exchange, un gateway de Discord/Slack, la mempool — antes significaba cron + polling. El cliente WebSocket lo reemplaza con una conexión en vivo. Es un transporte general, no una feature de blockchain: vive en la biblioteca estándar al lado del cliente HTTP, y lo gatea la misma capability net(host) con el mismo scope.

No es un cliente mínimo de una-conexión-por-vez: multiplexa miles de feeds en un solo thread (readiness-driven vía epoll/kqueue/IOCP — CPU ~0 mientras está ocioso), reconecta solo con backoff acotado y resubscribe, detecta conexiones half-open por keepalive, y acota la memoria con backpressure. Los dos ejes de escala: ws_select (fan-in vertical, N feeds en 1 thread) y parallel_map (fan-out horizontal, N workers × 1 conexión).

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]"))

El modelo de capabilities (sin puerta nueva)§

ws_connect necesita net(host) — deny-by-default, con scope por hostname, exactamente como http_get. No hay un permiso específico de WebSocket: un WebSocket es transporte, gateado como cualquier otra salida de red.

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

Enviar, recibir, cerrar§

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)

Multiplexar miles de feeds: ws_select (el event-loop)§

Un ws_recv espera UNA conexión. Para mirar N feeds sin quemar un thread por cada uno, ws_select espera la primera que tenga datos — el select {} de Go sobre channels, en una sola llamada. Es readiness-driven (epoll/kqueue/IOCP vía mio): CPU ~0 mientras está ocioso (duerme en el kernel, no busy-spinea) y escala a miles.

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"])

Resiliencia: reconexión y keepalive (opt-in)§

Sin estas opciones, el comportamiento es el de siempre — nada silencioso. Con ellas, un agente de vida larga sobrevive caídas y detecta sockets muertos sin plumbing en userland.

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}})
La frontera del motor síncrono. No hay hilo de fondo por conexión (rompería el aislamiento CSP y la promesa de "sin thread por conexión"). El keepalive y la reconexión avanzan mientras el programa está dentro de ws_select/ws_recv/ws_status — por eso el patrón es un while con ws_select. Un ws_send suelto sin un recv posterior no hace tickear los timers.

Fan-out con parallel_map: N workers × 1 conexión§

ws_select es el eje vertical (muchos feeds, un thread). El horizontal es parallel_map: cada worker recibe su propio intérprete (hereda las caps) y su propio registro de WebSockets — abre su feed, lo procesa y devuelve. Los handles no cruzan workers (aislamiento CSP): nunca compartas un handle entre 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

Nunca cuelga, no se lo puede inundar, no busy-spinea§

Una conexión muerta jamás cuelga al agente: connect, el handshake y cada ws_recv/ws_select están acotados por el timeout — vencido → nothing, no un programa trabado. Memoria acotada bajo entrada hostil: cada mensaje tiene tope (default 16 MiB, techo 64 MiB) y la cola inbound por conexión está acotada en las dos dimensiones — mensajes (max_queue, default 1024) y bytes (max_queue_bytes, default 64 MiB) — así que una inundación de frames grandes no puede OOMearte más que una de frames chicos. La política por default al tocar cualquiera de los topes es backpressure TCP real: se deja de leer el socket, la ventana TCP frena al peer — sin pérdida de datos ni OOM. Las alternativas son explícitas: "drop_oldest" descarta los más viejos hasta que el nuevo quepa, y "error" drena lo ya encolado y después emerge un error atrapable que nombra el overflow — jamás un descarte silencioso. La entrega entre feeds es equitativa round-robin: una conexión parlanchina con backlog no puede hambrear a las demás. Y ws_select ocioso duerme en el kernel (una sola espera de readiness), no gira en vacío. Un tope blando de conexiones por intérprete (SYNSEMA_WS_MAX_CONNS, default 4096) evita que un loop abra 100k sockets.

Subprotocolos§

opts.subprotocols (lista) negocia el header Sec-WebSocket-Protocol; el subprotocolo acordado queda en 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

Bajo serve: el puente WS→SSE§

El patrón de servidor más común es un puente: una ruta abre un WebSocket saliente hacia un RPC o exchange, y reenvía lo que recibe al navegador por SSE. Esto es de las pocas cosas que no se pueden mostrar del todo en run/test (necesita un servidor corriendo y un upstream en vivo), así que acá va la forma:

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)

El navegador abre un EventSource("/live") común; el servidor sostiene el WebSocket upstream. Las conexiones WebSocket entrantes (un servidor WS) son un bloque route … socket desde el engine v0.6.7 — la misma familia ws_* funciona sobre el handle entrante, y select espera al feed upstream, al socket del navegador, a un proceso hijo y al bus de eventos en una sola llamada: ver Apps agénticas.

Qué desbloquea§

Como es una primitiva general, no es solo para blockchain. Cualquier programa Synsema puede ahora construir sobre un stream bidireccional en vivo: dashboards en tiempo real, un agente reaccionando a un gateway de webhooks, herramientas colaborativas, un grabador de market-data, un cliente de chat. Donde antes escribías un loop de polling, ahora sostenés una conexión abierta.

Un ejemplo concreto: la suscripción eth_subscribe de un nodo EVM (newHeads/logs) se compone hoy en userland — ws_connect al endpoint WS del nodo, ws_send del frame JSON-RPC de subscribe, ws_recv + json_decode de cada notificación, y cada log se decodifica con el fragmento ABI del evento (abi_decode_log, el mismo que lee evm_logs y los 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 es el nombre del método JSON-RPC del nodo, así que conserva el prefijo eth_. Las lecturas one-shot (nonce, fees, saldos, un rango de logs, receipts) van por el lado de lectura tipado de blockchain.