@causal-order/transport

WebSocket + JSON ingress with acknowledged delivery, bounded shutdown, and event normalization for the causal-order stack.

Published package version: v0.2.2

Stack Position

nodes -> @causal-order/transport -> @causal-order/dedupe -> causal-order -> application

Transport owns WebSocket connection lifecycle, the versioned JSON protocol, delivery acknowledgment, peer evidence, and normalization into the canonical event contract. Deduplication and causal ordering remain downstream concerns.

Deployments requiring bounded outage buffering, health-aware routing, controlled recovery, and replay place @causal-order/monitor between transport and dedupe:

nodes -> @causal-order/transport -> @causal-order/monitor -> @causal-order/dedupe -> causal-order -> application

Monitor owns buffering and replay. Replayed events return through dedupe before causal ordering.

Install and Requirements

npm install @causal-order/transport @causal-order/dedupe causal-order

• Supported Node.js release lines: 22 and 24
• ESM-only output

The monitor-enabled stack requires Node.js 22.13.0 or newer because @causal-order/monitor uses the built-in node:sqlite module.

Server

Register one primary delivery authority when downstream acceptance must control the acknowledgment returned to the sender.

import { WebSocketJsonTransport } from "@causal-order/transport";

const transport = new WebSocketJsonTransport({
  mode: "server",
  host: "127.0.0.1",
  port: 8080,
});

transport.onDelivery(async (event, context) => {
  await deliverToNextStage(event, context);
  return { outcome: "accepted" };
});

transport.onEvent((event, context) => {
  observeTransportEvent(event, context);
});

transport.onPeerState((state) => {
  console.log("peer", state.peerId, state.status);
});

transport.onError((error) => {
  console.error("transport", error.detail);
});

await transport.start();

onDelivery() is the delivery authority. onEvent() is optional observation and does not control acknowledgment. deliverToNextStage is application-provided; it normally hands the event to monitor or directly to dedupe. Without an onDelivery() handler, successful normalization and observer dispatch form the acknowledgment boundary.

The default normalizer is used when normalizeMessage is omitted. A custom normalizer may adapt another inbound JSON event shape:

const transport = new WebSocketJsonTransport({
  mode: "server",
  port: 8080,
  normalizeMessage(message) {
    return normalizeApplicationMessage(message);
  },
});

Client

Client transports accept normalized events containing bigint-backed causal metadata. Known bigint fields are encoded as decimal strings on the JSON wire. send() resolves only after the receiving transport's primary delivery handler acknowledges the transmission. A WebSocket write callback alone is not treated as remote delivery.

import { WebSocketJsonTransport } from "@causal-order/transport";

const client = new WebSocketJsonTransport({
  mode: "client",
  url: "ws://127.0.0.1:8080",
  peerId: "ingress",
});

await client.start();

await client.send({
  id: "edge-a-000000000042",
  nodeId: "edge-a",
  sequence: 42n,
  clock: {
    physicalTimeMs: 1781000000000n,
    logicalCounter: 0,
    nodeId: "edge-a",
  },
  parentEventId: "edge-a-000000000041",
  dependencyEventIds: ["edge-b-000000000017"],
  partition: "factory-1",
  payload: { temperature: 21 },
});

Payloads must be JSON-serializable. Arbitrary bigint values inside payload are not converted automatically.

In server mode, send(event) broadcasts to all connected peers. send(event, peerId) targets one connected peer.

Every unicast target receives a unique transmission ID. Broadcasts snapshot their target peers and wait for every target to acknowledge or fail. A partial broadcast failure rejects with a TransportOperationError whose cause retains the individual peer failures.

Default Wire Format

Transport 0.2.x sends a versioned event envelope:

{
  "schema": "causal-order/transport",
  "version": 1,
  "type": "event",
  "transmissionId": "tx-018f-example",
  "event": {
    "id": "edge-a-000000000042",
    "nodeId": "edge-a",
    "sequence": "42",
    "clock": {
      "physicalTimeMs": "1781000000000",
      "logicalCounter": 0,
      "nodeId": "edge-a"
    },
    "parentEventId": "edge-a-000000000041",
    "dependencyEventIds": ["edge-b-000000000017"],
    "partition": "factory-1",
    "payload": {
      "temperature": 21
    }
  }
}

The receiver returns a correlated result after primary delivery:

{
  "schema": "causal-order/transport",
  "version": 1,
  "type": "delivery-result",
  "transmissionId": "tx-018f-example",
  "outcome": "accepted"
}

Refusal and failure results carry the same transmissionId, a non-accepted outcome, and a machine-readable code. Application event IDs do not serve as transmission IDs, so retries and intentional duplicates remain independently accountable.

The receiver continues to accept the legacy flat 0.1.x event object as input, but legacy messages cannot participate in acknowledged root send() semantics.

The following aliases are also accepted:

| Canonical field | Accepted aliases | | --- | --- | | nodeId | node, sourceNodeId | | sequence | seq | | clock.physicalTimeMs | ts, timestampMs, physicalTimeMs | | clock.logicalCounter | logical, lc | | payload | body |

sequence, timestamps, and ingestedAt accept bigint values, safe integer numbers, or canonical decimal integer strings. Unsafe numbers and malformed integer strings are rejected.

Normalized events contain:

• id
• nodeId
• optional sequence
• clock.physicalTimeMs, clock.logicalCounter, and clock.nodeId
• payload
• optional partition, parentEventId, and dependencyEventIds
• optional traceId and ingestedAt

API

The main entry point, @causal-order/transport, exports:

• WebSocketJsonTransport
• normalizeTransportEventMessage()
• createEventId()
• createHarnessAdapter()
• transport, event, context, peer-state, and option types

The published testing adapter supports deterministic node disconnect/reconnect and accepts the root transport's bounded per-peer send capacity:

const adapter = await createHarnessAdapter({
  nodeIds: ["edge-a", "edge-b"],
  maxInFlightSendsPerPeer: 4096,
});

await adapter.setNodeConnectivity("edge-a", "disconnected");
const bufferedSend = adapter.send(event);
await adapter.setNodeConnectivity("edge-a", "connected");
await bufferedSend;

WebSocketJsonTransport implements this lifecycle:

interface TransportContract<T> {
  readonly state: TransportLifecycleState;
  start(): Promise<void>;
  stop(): Promise<void>;
  send(event: NormalizedTransportEvent<T>, targetPeerId?: string): Promise<void>;
  onDelivery(handler: TransportDeliveryHandler<T>): () => void;
  onEvent(handler: TransportEventHandler<T>): () => void;
  onPeerState(handler: TransportPeerStateHandler): () => void;
  onError(handler: TransportErrorHandler): () => void;
}

onDelivery() registers one primary delivery authority. Its result determines the correlated acknowledgment returned to the sender. It may be asynchronous and may accept or refuse delivery. onEvent() remains an observational notification; observer failures are reported without changing the primary delivery result.

Lifecycle state is idle, starting, started, stopping, stopped, or failed. Shutdown atomically rejects new sends, drains accepted sends and receive work, closes sockets, and resolves only when no transport callback can arrive later. A bounded drain failure rejects explicitly.

Timeout and pressure options include connectionTimeoutMs, acknowledgmentTimeoutMs, deliveryTimeoutMs, shutdownTimeoutMs, maxInFlightSends, maxInFlightSendsPerPeer, maxInFlightReceives, maxInFlightReceivesPerPeer, and maxBufferedAmountBytes.

Instances may restart after a completed stop. Automatic reconnect is not part of 0.2.1; applications must create or restart a client according to their own retry policy.

Testing Subpath

The package exports @causal-order/transport/testing for @causal-order/testing. It is validation tooling, not an application runtime API.

npm install --save-dev @causal-order/testing

The adapter uses root transport clients and exercises the same acknowledgment, pressure, and stop-barrier contract as the production API. Repository test profiles and release evidence are documented in VALIDATION.md.

Runtime Boundary

• This package provides transport and normalization. Deduplication and causal ordering remain the responsibility of @causal-order/dedupe and causal-order.
• WebSocket + JSON is the supported ingress path. Other wire protocols should use separate adapters built against the same normalized event contract.
• The required ws runtime is vendored behind an internal wrapper, so consumers do not install ws as an additional runtime dependency.
• Vendored source, license, version, and local-modification details are recorded in THIRD_PARTY_NOTICES.md.

License

MIT