Skip to content

WebSocket Feed

Purpose

The WebSocketFeed worker (Virtufin.Worker.WebSocketFeed, worker type websocketfeed, general-workers): one outbound WebSocket connection as one WorkManager worker, venue-agnostic: every venue fact (login, subscription frames, symbol spelling, limits, routing keys, the topic-to-item rules) is config, filled in from a venue preset (market-data-devkit feeds/<preset>.json). It serves topics: it is told which topics to fill, maps each to a venue item, subscribes those items and routes every frame to its topic; it knows no consumer. It opens the socket when the worker starts, logs in and subscribes on every open, relays every frame that is one JSON value to pub/sub through its host's outbox under the worker lease, reconnects with backoff, and takes its commands through WorkManager. Ownership, placement, the lease, tags, lifecycle events and secrets are the worker's (Worker Management); this spec covers what the feed adds.

Related specs: - Worker Management — the worker lifecycle contract (StartAsync/ProcessAsync/StopAsync), the Worker Lease, Worker Commands (InvokeWorkerCommand), Topic Dependencies (a worker's input topics are its whole dependency), worker secrets and lifecycle events. - Pub/Sub Topics — the venue topics the frames are routed to, Ingest Time, the virtufin.worker.websocketfeed fragment's names.

Requirements

Requirement: Venue-Agnostic Feed

The feed's source SHALL contain no venue: no venue's name, login, frame format, symbol rule or limit. A venue's protocol SHALL be config: login step templates, subscription frame and item templates, a symbol format, item limits, routing-key templates and the topic-to-item rules (topicitems), kept per venue as a preset next to the venue's naming fragment (market-data-devkit feeds/<preset>.json, read with VenueFeeds.Config(preset, overrides, world) in C#, virtufin.marketdata_devkit.feeds.config(preset, overrides, world=...) in Python and feeds.config(preset, overrides, world) of @virtufin/marketdata-devkit in TypeScript, which render the symbol spelling and the topicitems rules from the venue's naming fragment, never from hand-written topics). The config SHALL be described by a JSON Schema (draft 2020-12) shipped with the feed (schema/websocketfeed.config.schema.json in its package, published on its docs site, x-feed-version = the feed's version); the schema describes the decoded config (a preset's shape), on the wire every value being a string (booleans and numbers as their text, arrays and objects as their JSON text). Every preset SHALL validate against it.

Scenario: A new venue

  • WHEN a venue is added
  • THEN it SHALL be a new preset (and naming fragment), never code in the feed

Scenario: Unknown template placeholder

  • WHEN a template uses a placeholder its key does not allow, or a JSON template does not render one JSON value
  • THEN the feed's start SHALL fail naming the key

Requirement: One Connection per Worker

Each WebSocket connection SHALL be one WebSocketFeed worker. Creating, starting, stopping, deleting, listing and tagging a connection SHALL be the WorkManager's worker RPCs (CreateWorker with the feed's config, secrets and tags, StartWorker, StopWorker, DeleteWorker, ListWorkers, the tag RPCs). A feed SHALL be created with autoreload=false, so a library release does not bounce its socket. Its only topic SHALL be its command topic: workers starting, stopping or (re)subscribing SHALL never change a feed.

Scenario: Open a connection

  • WHEN a client creates a WebSocketFeed worker with a url and starts it
  • THEN the feed SHALL open the socket in the background from StartAsync, report connecting, and report connected once the login and the subscriptions are done
  • AND the client SHALL wait for the status connected (a lifecycle status event or the worker's reported status) before relying on frames

Scenario: Close a connection

  • WHEN a client deletes the worker
  • THEN the feed SHALL send a normal close and abort the socket within the stop deadline, and its session state SHALL be deleted with the worker

Requirement: Feed URL Guard

The feed SHALL accept only ws:// and wss:// URLs and SHALL check the URL at start against the SSRF guard: loopback, private and link-local hosts SHALL be refused unless the Runtime pod's operator allows them (WEBSOCKETFEED_ALLOWED_HOSTS, WEBSOCKETFEED_ALLOW_PRIVATE_HOSTS); the worker's creator cannot.

Scenario: Private host refused

  • WHEN a feed is started with a URL whose host resolves to a private address and the operator has not allowed it
  • THEN the start SHALL fail and no socket SHALL be opened

Requirement: Login and Subscriptions on Every Open

On every socket open (start, reconnect, handoff) the feed SHALL run its login (handshake: the steps [{text|base64, expect?, fail?}]; a text step's {secret:NAME} placeholders filled from the worker secrets, resolved again on this open, and {config:KEY} from its config), then subscribe the items of every served topic with subscribeframe, then send its fixed subscribe frames and every frame sent with replay, before it relays anything and before it reports connected. The rendered login SHALL exist only for that open: never stored, logged or returned. A secret missing on an open SHALL fail that login (the reconnect backoff applies); a feed created without a secret its login names SHALL fail its start.

Scenario: Derived subscription

  • WHEN the feed opens with subscribeframe, subscribeitem and frameper configured
  • THEN it SHALL subscribe the (channel, venue symbol) item of every served topic, in served order, excluding items beyond maxitems/maxitemsperchannel
  • AND a serve SHALL send only the delta: unsubscribeframe for the items of removed topics, subscribeframe for the added ones

Scenario: Login frames are not relayed

  • WHEN frames arrive while the feed awaits a step's expect or fail match
  • THEN they SHALL NOT be published

Scenario: Login fails

  • WHEN a step's fail pattern matches, no expect match arrives within handshaketimeoutms, or the socket closes during the login
  • THEN the open SHALL count as a failed attempt under the reconnect backoff

Scenario: Handoff

  • WHEN a start finds the worker's saved session (a restart, reload or move to another Runtime pod)
  • THEN the open SHALL be a handoff, and a rejected login (a provider that still sees the previous socket and refuses a second one) SHALL be retried with backoff

Requirement: Reconnection

A dropped socket SHALL be reported reconnecting (with the attempt and the reason) and reopened with exponential backoff (1 s doubling to 30 s), running the login and subscriptions again. With autoreconnect=true (the default) the feed SHALL keep trying while the worker runs unless reconnectmaxattempts caps it; with autoreconnect=false, or when the attempts are exhausted, it SHALL report disconnected.

Scenario: Reconnect succeeds

  • WHEN a reopened socket completes its login and subscriptions
  • THEN the feed SHALL report connected and relay frames again

Requirement: Frame Publishing

Every received frame that is exactly one JSON value SHALL be published through the worker context's PublishAsync (the host's outbox, gated by the Worker Lease) to the topic its routing selects ({key_path, split_arrays, keys}: the key read at key_path, or the values of several comma-separated paths joined by |, picks the served topic whose routing key it is (its rule's routingkey, else routingkeyitem, rendered for its item), else a topic from the explicit keys; an explicit key wins; with split_arrays each element of an array frame on its own), else to publishtopic. Without a publish topic, a frame no key maps SHALL be dropped and counted. A frame that is not one JSON value SHALL be skipped, counted and logged, without affecting other frames.

Scenario: Frame envelope

  • WHEN a frame is published
  • THEN the CloudEvent SHALL have type com.virtufin.websocketfeed.message, source urn:com.virtufin:worker:websocketfeed (fragment virtufin.worker.websocketfeed), the extensions websocketid (the worker id) and websocketurl, ingesttime (Pub/Sub Topics spec, Ingest Time), time = the publish time, and the frame text as its data

Scenario: Ownership lost

  • WHEN the stop token fires (stop, delete, reload, lease lost, a move)
  • THEN the feed SHALL publish nothing more, and frames still queued SHALL be dropped

Requirement: Feed Commands

The feed SHALL take the commands serve, send, publish, unpublish and status through InvokeWorkerCommand (answered with com.virtufin.websocketfeed.response {command, ok, ...} or com.virtufin.websocketfeed.error) and on its command topic wrk.websocketfeed.command.<id> (answered on the command's replytopic only). serve, publish ({topic, routing?}), unpublish and a send with replay SHALL be saved in the worker's state, so reconnects and handoffs keep them, and SHALL win over the config on a later start.

Scenario: Send a frame

  • WHEN a client invokes send with {text}, {base64} or the frame itself
  • THEN the feed SHALL send one frame (text for text and JSON, binary for bytes); with replay: true it SHALL also send it on every later open

Scenario: Send with a reply

  • WHEN a client invokes send with expect and/or fail (regular expressions, matched like a login step's) and optionally timeoutms (default 5000, at most 60000), as fields next to text/base64 or as attributes
  • THEN the feed SHALL listen before it sends and answer with reply (the first inbound frame after the send that matches expect, or any frame when only fail is given: {text} or {base64} with its ingesttime) and replyms
  • AND a reply matching fail SHALL be an error quoting it, no matching frame within timeoutms an error, and a feed that is not connected an error (even with replay)
  • AND the reply SHALL still be published like every other frame; the patterns are the caller's, so the feed stays venue-agnostic

Scenario: Status

  • WHEN a client invokes status
  • THEN the feed SHALL answer its URL, state, detail, the login's step count and secret names, publish topic, routing (with the served topics' keys), its served topics (each with its channel, symbol and routing key), subscribed and excluded items per channel, its release state (releasewhenidleminutes, idlesince, releasing) and counters (received, published, skipped, opens), never a secret's value

Requirement: Feed Status

The feed SHALL report its connection state through context.ReportStatusAsync (a worker lifecycle event state = status on sys.workmanager.worker.lifecycle): connecting, connected, reconnecting, disconnected, degraded (items beyond maxitems or maxitemsperchannel, which the feed does not subscribe) and releasing (its self-release deleting it). There SHALL be no lifecycle topic of the feed's own.

Scenario: Venue limit exceeded

  • WHEN the served topics' items of a channel exceed its maxitemsperchannel (the alpaca preset: 10 orderbooks), or all items exceed maxitems
  • THEN the feed SHALL subscribe the first ones within the limit, report degraded naming the excluded ones and list them in status

Requirement: Served Topics

A feed SHALL serve a set of topics: config topics (a JSON array) from its first start, changed at runtime by the serve command ({"topics": [...]} sets the set, {"add": [...], "remove": [...]} changes it), saved in its worker state (winning over the config after a restart, reload or move). Each served topic SHALL map to one venue item by the first topicitems rule ({topic, channel?, routingkey?, values?}) whose dot-separated template it matches: a segment is a literal or one whole {name} placeholder; the rule binds exactly one of {pair} (a canonical pair, spelled by symbolformat/symbolcase) or {symbol} (a venue symbol as is); other bound names feed the channel template, through values where the venue spells them differently. A serve SHALL be all or nothing: a topic no rule covers, a {pair} that is not BASE_QUOTE, or a topic whose item or routing key another served topic has SHALL refuse the whole command naming every refused topic (one venue item, one topic); the same for config topics at start. An empty set SHALL unsubscribe everything and keep the socket open. The feed SHALL know no consumer: nothing in the platform matches topics to feeds; an orchestrator (a notebook, a script, the TUI) decides which feed serves which topic (market-data-devkit FeedResolver/FeedOrchestrator, Feed Orchestration below).

Scenario: Serve adds a topic

  • WHEN a client invokes serve with {"add": ["act.exchange.<venue>.orderbook.update.BTC_USDT"]} on a feed whose rules cover it
  • THEN the feed SHALL subscribe that topic's item, route its frames to the topic, save the set and answer the served topics, the added and removed ones and the subscribed and excluded items

Scenario: Serve refuses an unknown topic

  • WHEN a serve names a topic no topicitems rule covers
  • THEN the feed SHALL answer an error naming it and change nothing

Requirement: Self-Release

A feed SHALL live until its orchestrator deletes it, also while it serves no topic, unless it was created with releasewhenidleminutes (an integer >= 1; unset = never). With it, a feed that has served no topic for that many minutes SHALL close its socket, save that it is releasing and delete itself (DeleteWorker); a feed moved while releasing SHALL finish the release without opening its socket. The release SHALL be decided under the same lock as serve: a serve that makes the set non-empty within the grace period SHALL cancel the release (emptying it again starts the period over), and a serve after the decision SHALL be refused (status releasing). An orchestrator applying topics by feed name SHALL then create the feed again, so a release racing an update ends with exactly one feed serving the topics, never two and never none.

Scenario: Idle feed releases itself

  • WHEN a feed with releasewhenidleminutes=1 has served no topic for one minute
  • THEN it SHALL close its socket, report releasing and delete itself

Scenario: Release racing an update

  • WHEN an orchestrator serves topics to a feed of that name while the feed decides to release itself
  • THEN either the serve lands first and the feed keeps serving, or it is refused and the orchestrator creates a new feed of that name; afterwards exactly one feed serves the topics

Scenario: No release by default

  • WHEN a feed without releasewhenidleminutes serves no topic
  • THEN it SHALL stay connected and idle until it is deleted

Requirement: Feed Orchestration

market-data-devkit SHALL ship the orchestrator side (C# FeedResolver/FeedOrchestrator, Python feeds.resolve/feeds.apply/feeds.delete, TypeScript feeds.resolve/feeds.apply/feeds.delete of @virtufin/marketdata-devkit), the three behaving alike: they SHALL run the same conformance vectors (conformance/feed-resolver.v1.json): topics (or the topics workers listen to) to the feeds that serve them. A topic SHALL be recognised by the naming library (a venue-bound market entity's event) and served by its venue's preset (default: the preset named like the venue) when the preset maps its family; other topics SHALL be reported as unserved. The topics SHALL be grouped per (preset, world), sorted, and split over as few feeds as the preset's maxitems/maxitemsperchannel allow, a topic staying on its current feed while that has room; a venue's maxconnections (one at Alpaca) SHALL cap the feeds and the topics beyond it SHALL be reported (the plan is incomplete and SHALL NOT be applied), never dropped silently. Feed names SHALL be deterministic (<owner>-<preset>-<world>-<n>). Applying a plan SHALL, per feed name and owner, serve the running feed its topics or create it, and delete the owner's feeds the plan no longer names; it SHALL be idempotent and independent of whether the workers started before or after.

Scenario: Over a venue's connection limit

  • WHEN an orchestrator resolves eleven Alpaca order-book topics
  • THEN the plan SHALL hold one feed with ten of them and report the eleventh as over the limit, and applying it SHALL fail without creating anything