Skip to content

Socket and WebSocket connections

SharedData provides two persistent transports for synchronizing a table: TCP sockets and WebSockets. A receiver writes incoming records into its own local table. Neither transport shares memory across hosts or exposes the semantic HTTP query API. WebSockets also support a separate dictionary-stream protocol.

This guide describes the current connection interfaces and implementation. It does not report a live connection test or establish transport capacity. See the performance comparison for measured evidence and open measurements.

Choose the connection

Path Client interface Payload Authentication
Local table shdata.table(...) Mapped structured records Local filesystem/process access
HTTP queries ClientAPI or /api/data/... Query responses HTTP token permissions; HTTP guide
WebSocket table sync records.subscribe(...), records.publish(...), method='websocket' LZ4-compressed structured record batches Raw token in JSON login; WebSocket permission checks
TCP table sync The same table methods with method='socket' Four-byte network-order length followed by an LZ4 frame Fernet-wrapped shared token; see the compatibility note below
WebSocket stream ClientWebSocket.subscribe_stream(...), publish_stream(...) LZ4-compressed BSON dictionaries WebSocket token permissions

Table sync and dictionary streams use different message formats. A stream subscription is not a table subscription, and the TCP server implements table sync only.

Prepare both ends

  • Provision the table at the same database, period, source and name on both ends. Match the structured dtype exactly: field order, widths, types and byte order. The table wire format carries record bytes, not a schema migration. Receiving records updates the destination; start with an isolated table when checking an integration.
  • Configure each process's local storage through configuration. Opening an existing table without creation parameters does not provision a missing one. See tables.
  • Supply credentials from the environment or an approved secret store. The examples below contain no credential values. Do not put tokens in URLs or logs.
  • Use a secure WebSocket endpoint (wss) for remote access. Raw TCP has no built-in TLS; deploy it on a trusted transport with the access controls the installation requires. Fernet protects the login token, not the record stream.

The table-sync server currently opens tables in its server-side SharedData namespace (the stock entry points use master). SyncTable.init_client() does not forward the login's user to shdata.table(). The WebSocket permission check does include that user. Do not assume changing a login field selects a separate server-side namespace; align the configured namespace and permissions with the server's table ownership.

Server entry points

These are entry-point examples for an isolated installation. In a deployment managed by the Scheduler and WorkerPool, register and manage the routine through that control plane; do not start another copy beside an existing service. SYNC_BIND_HOST and SYNC_BIND_PORT below are example environment variables chosen by the operator, not built-in SharedData settings.

# WebSocket: aiohttp application served by Gunicorn; WebSocket route is /.
python -m SharedData.IO.ServerWebSocket \
  --host "$SYNC_BIND_HOST" --port "$SYNC_BIND_PORT" --nproc 1

# TCP: the entry point takes one comma-separated host,port argument.
python -m SharedData.IO.ServerSocket "$SYNC_BIND_HOST,$SYNC_BIND_PORT"

The WebSocket entry point does not configure TLS itself. A TLS terminator must forward the WebSocket upgrade and support long-lived connections. Client SHAREDDATA_CAFILE selects a CA file when needed; the connection helper also reads PROXY_ENDPOINT, PROXY_USER and PROXY_PWD for proxy configuration.

Subscribe to a table over WebSocket

The following runs in a long-lived receiver process. SYNC_* variables are example application configuration. The configured table must already exist; its user must match the server namespace described above. The normal token is read from SHAREDDATA_TOKEN by the table message builder.

import os
import time
from SharedData.SharedData import SharedData

shdata = SharedData(__file__, user=os.environ['SYNC_USER'])
records = shdata.table(
    os.environ['SYNC_DATABASE'], os.environ['SYNC_PERIOD'],
    os.environ['SYNC_SOURCE'], os.environ['SYNC_TABLE'])

records.subscribe(
    host=os.environ['SYNC_HOST'],
    port=int(os.environ['SYNC_PORT']),
    method='websocket', protocol='wss',
    lookbacklines=1000, snapshot=True, bandwidth=1_000_000)

# The helper starts a subscription thread. Keep its owning process alive.
while True:
    time.sleep(1)

SYNC_HOST is a hostname, without the scheme. The public table wrapper's protocol default is http; specify ws or wss explicitly. Use ws only where the connection's transport protection is otherwise appropriate.

The table helpers retry after disconnects. They do not expose a general stop/join API or an acknowledgement that the destination is caught up. Supervise the owning process and check destination values and freshness; starting a helper thread is not evidence of successful synchronization.

Publish a table over WebSocket

records.publish() uses the same host, port, method and synchronization options. The current high-level WebSocket publisher reads one JSON response; the server sends a login acknowledgement and then table metadata. The helper does not consume that second response to establish the receiver's count. Do not rely on it to negotiate a receiver-based resume position.

For an integration that needs those metadata, this explicit sequence consumes both frames before publishing. It uses the existing records prepared above as the sender. Run it in its own foreground process, instead of starting the subscription example at the same time. SYNC_WS_ENDPOINT is an example full wss://data.example.com/ endpoint supplied by the operator.

import asyncio
import json
import os
from SharedData.IO.ClientWebSocket import ClientWebSocket
from SharedData.IO.SyncTable import SyncTable

async def publish_table(records):
    session, websocket = await ClientWebSocket.get_websocket(
        os.environ['SYNC_WS_ENDPOINT'])
    try:
        message = SyncTable.publish_table_message(
            records.table, 1000, None, True, 1_000_000)
        await websocket.send_str(message)
        acknowledgement = await websocket.receive_json()
        if acknowledgement.get('message') != 'login success!':
            raise RuntimeError('WebSocket login rejected')
        metadata = await websocket.receive_json()
        if not {'count', 'mtime'} <= metadata.keys():
            raise RuntimeError('Missing receiver table metadata')
        client = json.loads(message)
        client.update(metadata)
        client['conn'] = websocket
        client['addr'] = ('configured receiver', None)
        client = SyncTable.init_client(client, records.table)
        await SyncTable.websocket_publish_loop(client)
    finally:
        await websocket.close()
        await session.close()

# After opening the existing sender table:
# asyncio.run(publish_table(records))

This is a continuous publisher, not a finite upload with a completion receipt. Connection setup, retries, deadlines and cancellation belong to the surrounding application when using these lower-level methods.

TCP socket authentication and connection

The TCP table server decrypts the login token with SHAREDDATA_SECRET_KEY and compares it with its configured SHAREDDATA_TOKEN. It does not use the WebSocket per-path permission tree. Treat access to this shared credential as access to the TCP table service, not as a scoped HTTP token.

The built-in table message builders put the client process's SHAREDDATA_TOKEN into JSON unchanged. They do not perform the Fernet wrapping the TCP server expects. Consequently, the same raw-token environment used for WebSockets is insufficient for a stock TCP helper.

For the stock TCP helper, provision a Fernet-wrapped wire token from the approved secret store into a dedicated client's SHAREDDATA_TOKEN; the server retains its underlying token and matching Fernet key. Keep that client separate from HTTP/WebSocket clients that require the raw token. Do not print either representation. Alternatively, a custom client can build the wrapped login in memory using Fernet(...).encrypt(...) and implement the framing below.

With that transport-specific credential provisioned, an existing receiver uses:

records.subscribe(
    host=os.environ['SYNC_HOST'], port=int(os.environ['SYNC_PORT']),
    method='socket', lookbacklines=1000,
    snapshot=True, bandwidth=1_000_000)

The sender can use records.publish(...) with the same TCP options. The helper creates its own thread, so its owning process must stay alive. As with WebSockets, verify the receiver's records; there is no per-batch durable commit receipt.

Synchronization controls and capacity

Argument Actual behavior
lookbacklines Default 1000. For an indexed table, selects the recent row window checked for changed mtime values. It is not a general cap on catch-up traffic.
lookbackdate For indexed tables, a pandas.Timestamp selects a date-based window. The message builder serializes it at day precision; a plain string is not serialized by that check.
snapshot=True For indexed tables, asks for an initial send of the selected lookback window. It is not an atomic table snapshot or automatically a full-table copy.
bandwidth Default 1e6: publisher pacing based on compressed body bytes per second, despite some older docstrings saying bits. Headers and transport overhead are additional.
method socket or websocket for the transports described here. api selects a separate HTTP synchronization path.
protocol Specify ws/wss for WebSocket clients. It does not add TLS to raw TCP.

New-row catch-up is driven by row counts, and append-only tables do not use the indexed mtime lookback logic. A lookback setting therefore does not bound every transfer. Updates outside the indexed lookback window can be missed; intermediate updates can be coalesced before they are observed. There is no event-by-event replay or exactly-once guarantee here.

The publisher selects records, converts batches to raw bytes, LZ4-compresses them and paces sends. Batches target 128 KiB of uncompressed records. TCP prefixes each compressed batch with a four-byte big-endian length and uses ping as its heartbeat marker; WebSocket batches are binary messages. The receiver decompresses and feeds complete records to read_stream(), which upserts indexed tables or extends append-only ones.

Lookback scanning, polling, compression, receiver apply work and subscriber fan-out affect capacity. Indexed operations still contend on the destination's per-table mutex. Mmap visibility and durable persistence are separate; a successful send is not proof of receiver application or persistence. Measure client-observed lag, delivered bytes, receiver correctness and error rates on an isolated workload before choosing a production rate.

WebSocket dictionary streams

ClientWebSocket.subscribe_stream(database, period, source, tablename, ...) is an async generator of decoded dictionaries. Pass user, endpoint and token explicitly or supply the supported environment defaults: SHAREDDATA_WS_ENDPOINT and SHAREDDATA_TOKEN.

await ClientWebSocket.publish_stream(...) returns an asyncio.Queue and an asyncio.Task; enqueue dictionaries and supervise that task in the same event loop. Enqueuing is not a delivery receipt. Stream messages use compressed BSON, not the raw table-record format. Offsets and consumer groups belong to the underlying stream and are separate from table-sync lookback and row counts.