proper-wse: a fork of WSE (WebSocket Engine)

This is a fork of wse-server, published on PyPI as proper-wse and maintained for the Proper web framework. It still imports as wse_server, so don't install it next to wse-server. Only the server is published; the clients in this repository are upstream's.

Changes from wse-server 2.4.3:

• Broadcast frames over 1 KiB are shared by every connection instead of copied into each one (silvermpx/wse#74).
• max_pending_handshakes; past it, a 503 counted in wse_handshakes_dropped_total (silvermpx/wse#73). Logs go to stderr.
• allowed_origins: handshakes from other sites' pages get a 403.
• topic_backlog(topic): how far behind delivery is on a topic.
• connection_backlogs(min_pending) and abort_connection(conn_id): find the clients that stopped reading (bytes pending, bytes written so far) and end them at once, which a Close frame queued behind their backlog can't do.
• The pending-bytes count of a connection drops with each write to its socket, not when a whole backlog is written.
• Wheels for free-threaded Python (cp314t) besides abi3.

High-performance WebSocket server built in Rust with native clustering, E2E encryption, message recovery, presence tracking, and real-time fan-out. Exposed to Python via PyO3 with zero GIL overhead on the data path. Ships with Rust-accelerated utilities for application code: priority queues, rate limiters, event sequencing, compression, cryptography, and JWT.

Features

Server

End-to-End Encryption

Cluster Protocol

Presence Tracking

Message Recovery

Client SDKs (Python + TypeScript/React)

Transport Security

Quick Start

pip install wse-server

from wse_server import RustWSEServer, rust_jwt_encode
import time, threading

server = RustWSEServer(
    "0.0.0.0", 5007,
    max_connections=10_000,
    jwt_secret=b"replace-with-a-strong-secret-key!",
    jwt_issuer="my-app",
    jwt_audience="my-api",
)
server.enable_drain_mode()
server.start()

def handle_events(srv):
    while True:
        for ev in srv.drain_inbound(256, 50):
            if ev[0] == "auth_connect":
                srv.subscribe_connection(ev[1], ["updates"])
            elif ev[0] == "msg":
                print(f"Message from {ev[1]}: {ev[2]}")
            elif ev[0] == "disconnect":
                print(f"Disconnected: {ev[1]}")

threading.Thread(target=handle_events, args=(server,), daemon=True).start()

while server.is_running():
    time.sleep(1)

Generate a test token:

token = rust_jwt_encode(
    {"sub": "user-1", "iss": "my-app", "aud": "my-api",
     "exp": int(time.time()) + 3600, "iat": int(time.time())},
    b"replace-with-a-strong-secret-key!",
)

Server Configuration

RustWSEServer constructor parameters:

API Reference

Lifecycle

server.start()                          # Start the server
server.stop()                           # Graceful shutdown
server.is_running()                     # Check server status (bool)

Event Handling

Drain mode (recommended) - events are queued in a lock-free crossbeam channel. Python polls in batches, acquiring the GIL once per batch.

server.enable_drain_mode()              # Switch to batch-polling mode (call before start)
events = server.drain_inbound(256, 50)  # Poll up to 256 events, wait up to 50ms

Each event is a tuple: (event_type, conn_id, payload)

Callback mode - alternative to drain mode. Callbacks are invoked via spawn_blocking per event.

server.set_callbacks(on_connect, on_message, on_disconnect)

Sending Messages

server.send(conn_id, text)                      # Send text to one connection
server.send_bytes(conn_id, data)                 # Send binary to one connection
server.send_event(conn_id, event_dict)           # Send structured event (auto-serialized, deduped, rate-checked)

server.broadcast_all(text)                       # Send to every connected client (text)
server.broadcast_all_bytes(data)                 # Send to every connected client (binary)
server.broadcast_local(topic, text)              # Fan-out to topic subscribers on this instance
server.broadcast(topic, text)                    # Fan-out to topic subscribers across all cluster nodes

Topic Subscriptions

server.subscribe_connection(conn_id, ["prices", "news"])              # Subscribe to topics
server.subscribe_connection(conn_id, ["chat"], {"status": "online"})  # Subscribe with presence data
server.subscribe_connection(conn_id, ["tasks"], queue_group="workers") # Subscribe with queue group (round-robin)
server.unsubscribe_connection(conn_id, ["news"])                      # Unsubscribe from specific topics
server.unsubscribe_connection(conn_id, None)                          # Unsubscribe from all topics
server.get_topic_subscriber_count("prices")                           # Subscriber count for a topic

Subscriptions are cleaned up automatically on disconnect. In cluster mode, interest changes are propagated to peers via SUB/UNSUB frames.

Queue groups: connections in the same queue_group receive messages round-robin instead of fanout. Normal subscribers (no queue group) still receive all messages. Useful for distributing work across a pool of consumers.

Topic ACL

Per-connection topic access control with glob pattern matching.

# Allow only "user:*" topics, deny everything else
server.set_topic_acl(conn_id, allow=["user:*"])

# Allow "data:*" but deny "data:internal:*"
server.set_topic_acl(conn_id, allow=["data:*"], deny=["data:internal:*"])

# Must be called before subscribe_connection
server.subscribe_connection(conn_id, ["data:prices"])   # allowed
server.subscribe_connection(conn_id, ["data:internal:audit"])  # denied

Deny patterns take precedence over allow patterns. Supports * (any characters) and ? (single character) wildcards. Applied at subscribe time.

Presence Tracking

Requires presence_enabled=True in the constructor.

# Query members in a topic
members = server.presence("chat-room")
# {"alice": {"data": {"status": "online"}, "connections": 2},
#  "bob":   {"data": {"status": "away"},   "connections": 1}}

# Lightweight counts (O(1), no iteration)
stats = server.presence_stats("chat-room")
# {"num_users": 2, "num_connections": 3}

# Update a user's presence data across all their subscribed topics
server.update_presence(conn_id, {"status": "away"})

Presence is tracked at the user level (JWT sub claim). Multiple connections from the same user share a single presence entry. presence_join fires on first connection, presence_leave on last disconnect. In cluster mode, presence state is synchronized across all nodes using CRDT last-write-wins resolution.

Message Recovery

Requires recovery_enabled=True in the constructor.

result = server.subscribe_with_recovery(
    conn_id, ["prices"],
    recover=True,
    epoch=client_epoch,       # From previous session
    offset=client_offset,     # From previous session
)
# {"topics": {"prices": {"epoch": "0000007b", "offset": 456, "recovered": True, "count": 12}}}

The server maintains per-topic ring buffers (power-of-2 capacity, bitmask indexing). When recovery is enabled, every topic broadcast is stamped with tp/e/o (topic, epoch, offset) fields, so clients track their per-topic position from the messages themselves, drop duplicates idempotently by (epoch, offset), and automatically request recovery when they detect an offset gap. On reconnect, the server replays missed messages from the ring buffer. If the gap is too large or the epoch has changed, the client receives a NotRecovered status and should re-subscribe from scratch.

Memory is managed with a global budget (default 256 MB), TTL eviction for idle buffers, and LRU eviction when over budget.

Cluster

# Join a cluster mesh with mTLS
server.connect_cluster(
    peers=["10.0.0.2:9999", "10.0.0.3:9999"],
    tls_ca="/etc/wse/ca.pem",
    tls_cert="/etc/wse/node.pem",
    tls_key="/etc/wse/node.key",
    cluster_port=9999,
)

# With gossip discovery (only seed addresses needed)
server.connect_cluster(
    peers=[],
    seeds=["10.0.0.2:9999"],
    cluster_addr="10.0.0.1:9999",
    cluster_port=9999,
)

server.cluster_connected()       # True if connected to at least one peer
server.cluster_peers_count()     # Number of active peer connections
server.cluster_info()            # List of connected peers (address, instance_id, connected)

Nodes form a full TCP mesh automatically. The cluster protocol uses a custom binary frame format with an 8-byte header, 12 message types, and capability negotiation during handshake. Features:

• Interest-based routing - SUB/UNSUB/RESYNC frames. Messages are only forwarded to peers with matching subscribers.
• Gossip discovery - PeerAnnounce/PeerList frames. New nodes need one seed address to join.
• mTLS - mutual TLS via rustls with P-256 certificates and WebPkiClientVerifier.
• zstd compression - payloads above 256 bytes compressed at level 1, capability-negotiated.
• Circuit breaker - 10 failures to open, 60s reset, 3 half-open probe calls.
• Heartbeat - 5s interval, 15s timeout, dead peer detection.
• Dead letter queue - 1000-entry ring buffer for failed cluster sends.
• Presence sync - PresenceUpdate/PresenceFull frames with CRDT conflict resolution.

Health Monitoring

health = server.health_snapshot()
# {
#     "connections": 150,
#     "inbound_queue_depth": 0,
#     "inbound_dropped": 0,
#     "uptime_secs": 3600.5,
#     "recovery_enabled": True,
#     "recovery_topic_count": 5,
#     "recovery_total_bytes": 1048576,
#     "cluster_connected": True,
#     "cluster_peer_count": 2,
#     "cluster_messages_sent": 50000,
#     "cluster_messages_delivered": 49950,
#     "cluster_messages_dropped": 0,
#     "cluster_bytes_sent": 1048576,
#     "cluster_bytes_received": 1024000,
#     "cluster_reconnect_count": 0,
#     "cluster_unknown_message_types": 0,
#     "cluster_dlq_size": 0,
#     "presence_enabled": True,
#     "presence_topics": 3,
#     "presence_total_users": 25,
# }

Connection Management

server.get_connection_count()        # Lock-free AtomicUsize read
server.get_connections()             # List all connection IDs (snapshot)
server.disconnect(conn_id)           # Force-disconnect a connection
server.drain(close_code=4300, close_reason="shutting down", timeout=10)  # Graceful drain
server.inbound_queue_depth()         # Events waiting to be drained
server.inbound_dropped_count()       # Events dropped due to full queue
server.get_cluster_dlq_entries()     # Retrieve failed cluster messages from dead letter queue

drain() sends a WebSocket Close frame to all connected clients and rejects new connections. The drain wait runs as a separate task, so the command processor stays responsive. Use for zero-downtime deployments and rolling restarts.

Security

JWT Authentication

Rust-native HS256 validation during the WebSocket handshake. Zero GIL acquisition, 0.01ms per decode.

Token delivery:

• Browser clients: access_token HTTP-only cookie (set by your login endpoint, attached automatically by the browser)
• Backend clients: Authorization: Bearer <token> header and/or access_token cookie
• API clients: Authorization: Bearer <token> header

Required claims: sub (user ID), exp (expiration). Recommended: iat (issued at). Optional: iss, aud (validated if configured).

End-to-End Encryption

Per-connection session keys via ECDH P-256 key exchange, AES-GCM-256 encryption, HKDF-SHA256 key derivation.

Wire format: E: prefix + 12-byte IV + AES-GCM ciphertext + 16-byte auth tag.

Enable on the client side - the server handles key exchange automatically during the handshake.

Rate Limiting

Per-connection token bucket: 100,000 token capacity, 10,000 tokens/second refill. Clients receive a rate_limit_warning at 20% remaining capacity, and RATE_LIMITED error when exceeded.

Deduplication

send_event() maintains a 50,000-entry AHashSet with FIFO eviction. Duplicate message IDs are dropped before serialization.

Zombie Detection

Server pings every connected client every 25 seconds. Connections with no activity for 60 seconds are force-closed.

Full security documentation: docs/SECURITY.md

Wire Protocol

WSE uses a custom wire protocol with categorized JSON messages:

Text frames: JSON with c field for category: "WSE" (system), "S" (snapshot), "U" (update)

Binary frames: C: (zlib compressed), M: (MessagePack), E: (AES-GCM encrypted), raw zlib (0x78 magic byte)

MessagePack transport: opt-in per connection via ?format=msgpack query parameter. Roughly 2x faster serialization and 30% smaller payloads.

Protocol negotiation: client_hello/server_hello handshake with feature discovery, capability advertisement, and version agreement.

Full protocol specification: docs/PROTOCOL.md

Compression

Two compression layers:

• Client-facing: zlib for messages above the configurable threshold (default 1024 bytes). Applied automatically by send_event().
• Inter-peer (cluster): zstd level 1 for payloads above 256 bytes. Capability-negotiated during handshake. Decompression output capped at 1 MB (MAX_FRAME_SIZE).

Client SDKs

Python

pip install wse-client

Full-featured async and sync client with connection pool, circuit breaker, auto-reconnect, E2E encryption, and msgpack binary transport.

from wse_client import connect

async with connect("ws://localhost:5007/wse", token="<jwt>") as client:
    await client.subscribe(["updates"])
    async for event in client:
        print(event.type, event.payload)

Sync interface:

from wse_client import SyncWSEClient

client = SyncWSEClient("ws://localhost:5007/wse", token="<jwt>")
client.connect()
client.subscribe(["updates"])

@client.on("updates")
def handle(event):
    print(event.payload)

client.run_forever()

Key features: 4 reconnect strategies (exponential, linear, fibonacci, adaptive), connection pool with health scoring and 3 load balancing strategies, circuit breaker, token bucket rate limiter, event sequencer with id dedup plus per-topic (epoch, offset) idempotent dedup and gap recovery, network quality monitoring (latency/jitter/packet loss).

See python-client/ for full source and examples.

TypeScript / React

npm install wse-client

Single React hook (useWSE) for connection lifecycle, subscriptions, and message dispatch.

import { useWSE } from 'wse-client';

function App() {
  const { sendMessage, connectionHealth } = useWSE(
    '<jwt-token>',
    ['updates'],
    { endpoints: ['ws://localhost:5007/wse'] },
  );

  return <div>Status: {connectionHealth}</div>;
}

Key features: offline queue with IndexedDB persistence, adaptive quality management, connection pool with health scoring, E2E encryption (Web Crypto API), message batching, 5 priority levels, Zustand store for external state access.

See ts-client/ for full source and examples.

Performance

Benchmarked on AMD EPYC 7502P (32 cores / 64 threads, 128 GB RAM), Ubuntu 24.04.

Sub-millisecond latency. Median 0.38ms with JWT authentication. Connection handshake: 0.53ms median (Rust JWT path).

Detailed results: Benchmarks | Fan-out | Rust Client | Python Client | TS Client

Examples

Working examples in the examples/ directory:

Documentation

• Architecture - server internals, PyO3 bridge, concurrency model
• Integration Guide - complete setup and API reference
• Wire Protocol - message format specification
• Cluster Protocol - TCP mesh, frame format, gossip, interest routing
• Rust Utilities - priority queues, rate limiters, sequencing, compression, crypto, JWT
• Security - JWT, encryption, mTLS, rate limiting, circuit breaker
• Deployment - production setup, Docker, Kubernetes, cluster configuration
• Migration Guide - upgrading from v1.x to v2.x
• Contributing - development setup and coding standards
• Changelog - version history

License

MIT