Module websocket 

WebSocket-based streaming DML ingest endpoint.

Clients open a WebSocket connection per table. The first text frame authenticates the session and later frames carry unsigned batches of upsert / delete DML messages. Each websocket payload carries a monotonically increasing dml_batch_id.

Wire format (JSON over text WebSocket frames):

Client → Server (authenticated init):

{"type": "init", "timestamp": 1760000000000}

Client → Server (DML batch):

{
  "dml_batch_id": 1,
  "items": [
    {"op": "upsert", "data": {"id": 1, "name": "foo"}},
    {"op": "delete", "data": {"id": 1, "name": "foo"}}
  ]
}

Server → Client (ack):

{"ack": 1}

Server → Client (fatal — the connection will close):

{"fatal": "DML channel closed, please reconnect"}

Structs

DmlRequest

IngestService

Shared state for the ingest service.

InitRequest 🔒

PreparedDmlBatch 🔒

RawDmlBatchRequest 🔒

RawDmlRequest

Enums

DmlOp 🔒

ServerMessage

Constants

INGEST_DML_REQUEST_BUFFER_SIZE 🔒

INIT_MESSAGE_TYPE 🔒

WEBSOCKET_INIT_TIMEOUT 🔒

Functions

build_router

handle_connection 🔒

parse_and_validate_init_request 🔒

prepare_dml_batch_payload 🔒

send_fatal 🔒

send_server_message 🔒

try_handle_connection 🔒

validate_monotonic_dml_batch_id 🔒

validate_timestamp_skew 🔒

ws_handler

Type Aliases

ServiceRef

WsRx 🔒

WsTx 🔒